Mirrors/blakeblackshear.frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2025-08-04 13:47:37 +02:00
Code Issues Projects Releases Wiki Activity

Make publisher have a generic type

Browse Source
This commit is contained in:
Nicolas Mowen 2025-06-06 08:35:30 -06:00
parent de7238df7f
commit d6f2d55525
3 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
frigate/comms/events_updater.py
View File

@ -7,7 +7,9 @@ from frigate.events.types import EventStateEnum, EventTypeEnum
from .zmq_proxy import Publisher, Subscriber from .zmq_proxy import Publisher, Subscriber
class EventUpdatePublisher(Publisher): class EventUpdatePublisher(
Publisher[tuple[EventTypeEnum, EventStateEnum, str, str, dict[str, Any]]]
):
"""Publishes events (objects, audio, manual).""" """Publishes events (objects, audio, manual)."""
topic_base = "event/" topic_base = "event/"
@ -16,9 +18,11 @@ class EventUpdatePublisher(Publisher):
super().__init__("update") super().__init__("update")
def publish( def publish(
self, payload: tuple[EventTypeEnum, EventStateEnum, str, str, dict[str, Any]] self,
payload: tuple[EventTypeEnum, EventStateEnum, str, str, dict[str, Any]],
sub_topic: str = "",
) -> None: ) -> None:
super().publish(payload) super().publish(payload, sub_topic)
class EventUpdateSubscriber(Subscriber): class EventUpdateSubscriber(Subscriber):
@ -30,7 +34,9 @@ class EventUpdateSubscriber(Subscriber):
super().__init__("update") super().__init__("update")
class EventEndPublisher(Publisher): class EventEndPublisher(
Publisher[tuple[EventTypeEnum, EventStateEnum, str, dict[str, Any]]]
):
"""Publishes events that have ended.""" """Publishes events that have ended."""
topic_base = "event/" topic_base = "event/"
@ -39,9 +45,11 @@ class EventEndPublisher(Publisher):
super().__init__("finalized") super().__init__("finalized")
def publish( def publish(
self, payload: tuple[EventTypeEnum, EventStateEnum, str, dict[str, Any]] self,
payload: tuple[EventTypeEnum, EventStateEnum, str, dict[str, Any]],
sub_topic: str = "",
) -> None: ) -> None:
super().publish(payload) super().publish(payload, sub_topic)
class EventEndSubscriber(Subscriber): class EventEndSubscriber(Subscriber):

4
frigate/comms/recordings_updater.py
View File

@ -14,7 +14,7 @@ class RecordingsDataTypeEnum(str, Enum):
recordings_available_through = "recordings_available_through" recordings_available_through = "recordings_available_through"
class RecordingsDataPublisher(Publisher): class RecordingsDataPublisher(Publisher[tuple[str, float]]):
"""Publishes latest recording data.""" """Publishes latest recording data."""
topic_base = "recordings/" topic_base = "recordings/"
@ -22,7 +22,7 @@ class RecordingsDataPublisher(Publisher):
def __init__(self, topic: RecordingsDataTypeEnum) -> None: def __init__(self, topic: RecordingsDataTypeEnum) -> None:
super().__init__(topic.value) super().__init__(topic.value)
def publish(self, payload: Any, sub_topic: str = "") -> None: def publish(self, payload: tuple[str, float], sub_topic: str = "") -> None:
super().publish(payload, sub_topic) super().publish(payload, sub_topic)

9
frigate/comms/zmq_proxy.py
View File

@ -2,7 +2,7 @@
import json import json
import threading import threading
from typing import Any, Optional from typing import Any, Generic, Optional, TypeVar
import zmq import zmq
@ -47,7 +47,10 @@ class ZmqProxy:
self.runner.join() self.runner.join()
class Publisher: T = TypeVar("T")
class Publisher(Generic[T]):
"""Publishes messages.""" """Publishes messages."""
topic_base: str = "" topic_base: str = ""
@ -58,7 +61,7 @@ class Publisher:
self.socket = self.context.socket(zmq.PUB) self.socket = self.context.socket(zmq.PUB)
self.socket.connect(SOCKET_PUB) self.socket.connect(SOCKET_PUB)
def publish(self, payload: Any, sub_topic: str = "") -> None: def publish(self, payload: T, sub_topic: str = "") -> None:
"""Publish message.""" """Publish message."""
self.socket.send_string(f"{self.topic}{sub_topic} {json.dumps(payload)}") self.socket.send_string(f"{self.topic}{sub_topic} {json.dumps(payload)}")