administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-02-03 01:35:22 +03:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Fix ws mypy

Browse Source
This commit is contained in:
Nick Mowen 2022-11-22 15:06:09 -07:00
parent fa05e55333
commit 00c0514127
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
frigate/comms/ws.py
View File

@ -4,6 +4,8 @@ import json
import logging import logging
import threading import threading
from typing import Callable
from wsgiref.simple_server import make_server from wsgiref.simple_server import make_server
from ws4py.server.wsgirefserver import ( from ws4py.server.wsgirefserver import (
WebSocketWSGIHandler, WebSocketWSGIHandler,
@ -20,23 +22,23 @@ from frigate.config import FrigateConfig
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class WebSocketClient(Communicator): class WebSocketClient(Communicator): # type: ignore[misc]
"""Frigate wrapper for ws client.""" """Frigate wrapper for ws client."""
def __init__(self, config: FrigateConfig) -> None: def __init__(self, config: FrigateConfig) -> None:
self.config = config self.config = config
def subscribe(self, receiver) -> None: def subscribe(self, receiver: Callable) -> None:
self._dispatcher = receiver self._dispatcher = receiver
self.start() self.start()
def start(self): def start(self) -> None:
"""Start the websocket client.""" """Start the websocket client."""
class _WebSocketHandler(WebSocket): class _WebSocketHandler(WebSocket): # type: ignore[misc]
receiver = self._dispatcher receiver = self._dispatcher
def received_message(self, message): def received_message(self, message: WebSocket.received_message) -> None:
try: try:
json_message = json.loads(message.data.decode("utf-8")) json_message = json.loads(message.data.decode("utf-8"))
json_message = { json_message = {
@ -72,7 +74,7 @@ class WebSocketClient(Communicator):
) )
self.websocket_thread.start() self.websocket_thread.start()
def publish(self, topic: str, payload: str, _) -> None: def publish(self, topic: str, payload: str, _: bool) -> None:
try: try:
ws_message = json.dumps( ws_message = json.dumps(
{ {
@ -87,7 +89,7 @@ class WebSocketClient(Communicator):
self.websocket_server.manager.broadcast(ws_message) self.websocket_server.manager.broadcast(ws_message)
def stop(self): def stop(self) -> None:
self.websocket_server.manager.close_all() self.websocket_server.manager.close_all()
self.websocket_server.manager.stop() self.websocket_server.manager.stop()
self.websocket_server.manager.join() self.websocket_server.manager.join()