mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-03 05:16:50 +03:00
Nicolas MowenandGitHub
5e87d101da
CI / AMD64 Build (push) Canceled after 0s
CI / AMD64 Smoke Test (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
* Integrate state changes with review items * Cleanup and fixes * Improve report context management
44 lines
1.1 KiB
Python
44 lines
1.1 KiB
Python
"""Facilitates communication between processes."""
|
|
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from .zmq_proxy import Publisher, Subscriber
|
|
|
|
|
|
class DetectionTypeEnum(str, Enum):
|
|
all = ""
|
|
api = "api"
|
|
video = "video"
|
|
audio = "audio"
|
|
lpr = "lpr"
|
|
classification_state = "classification_state"
|
|
|
|
|
|
class DetectionPublisher(Publisher):
|
|
"""Simplifies receiving video and audio detections."""
|
|
|
|
topic_base = "detection/"
|
|
|
|
def __init__(self, topic: str) -> None:
|
|
super().__init__(topic)
|
|
|
|
|
|
class DetectionSubscriber(Subscriber):
|
|
"""Simplifies receiving video and audio detections."""
|
|
|
|
topic_base = "detection/"
|
|
|
|
def __init__(self, topic: str) -> None:
|
|
super().__init__(topic)
|
|
|
|
def check_for_update(
|
|
self, timeout: float | None = None
|
|
) -> tuple[str, Any] | tuple[None, None] | None:
|
|
return super().check_for_update(timeout)
|
|
|
|
def _return_object(self, topic: str, payload: Any) -> Any:
|
|
if payload is None:
|
|
return (None, None)
|
|
return (topic[len(self.topic_base) :], payload)
|