Improve comms typing (#18599)

* Enable mypy for comms

* Make zmq data types consistent

* Cleanup inter process typing issues

* Cleanup embeddings typing

* Cleanup config updater

* Cleanup recordings updator

* Make publisher have a generic type

* Cleanup event metadata updater

* Cleanup event metadata updater

* Cleanup detections updater

* Cleanup websocket

* Cleanup mqtt

* Cleanup webpush

* Cleanup dispatcher

* Formatting

* Remove unused

* Add return type

* Fix tests

* Fix semantic triggers config typing

* Cleanup
This commit is contained in:
Nicolas Mowen
2025-08-16 10:20:33 -05:00
committed by Blake Blackshear
parent 1add72884a
commit fcf3824124
29 changed files with 168 additions and 128 deletions
+6 -8
View File
@@ -1,7 +1,7 @@
"""Facilitates communication between processes."""
from enum import Enum
from typing import Any, Optional
from typing import Any
from .zmq_proxy import Publisher, Subscriber
@@ -19,8 +19,7 @@ class DetectionPublisher(Publisher):
topic_base = "detection/"
def __init__(self, topic: DetectionTypeEnum) -> None:
topic = topic.value
def __init__(self, topic: str) -> None:
super().__init__(topic)
@@ -29,16 +28,15 @@ class DetectionSubscriber(Subscriber):
topic_base = "detection/"
def __init__(self, topic: DetectionTypeEnum) -> None:
topic = topic.value
def __init__(self, topic: str) -> None:
super().__init__(topic)
def check_for_update(
self, timeout: float = None
) -> Optional[tuple[DetectionTypeEnum, Any]]:
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 (DetectionTypeEnum[topic[len(self.topic_base) :]], payload)
return (topic[len(self.topic_base) :], payload)