enhancement(zmq): Add zmq hook WIP

foosball/hooks/__init__.py

@@ -5,3 +5,11 @@ class Hook(ABC):
     @abstractmethod
     def invoke(self, *args, **kwargs):
         pass
+
+    @abstractmethod
+    def start(self, *args, **kwargs):
+        pass
+
+    @abstractmethod
+    def stop(self, *args, **kwargs):
+        pass

foosball/hooks/audio.py

@@ -11,6 +11,12 @@ def __init__(self, folder: str):
         super().__init__()
         self.folder = folder
 
+    def start(self, *args, **kwargs):
+        pass
+
+    def stop(self, *args, **kwargs):
+        pass
+
     def invoke(self, *args, **kwargs):
         AudioHook.play_random_sound(self.folder)
 

foosball/hooks/web.py → foosball/hooks/webhook.py

@@ -18,6 +18,12 @@ def __init__(self, method: str = None, url: str = None, json: dict = None, heade
         self.json: dict = json
         self.headers: Mapping[str, str] = headers
 
+    def start(self, *args, **kwargs):
+        pass
+
+    def stop(self, *args, **kwargs):
+        pass
+
     def as_dict(self, json: dict = None) -> dict:
         d = vars(self)
         if json is not None:

foosball/hooks/zmq/__init__.py

@@ -0,0 +1,44 @@
+import json
+import logging
+
+import zmq
+from .. import Hook
+
+class ZMQHook(Hook):
+
+    def __init__(self, host="localhost", port=5555, topic="ws"):
+        super().__init__()
+        self.zmq_pub = ZMQPub(address=f"tcp://{host}:{port}")
+        self.topic = topic
+
+    def invoke(self, message: dict | str, *args, **kwargs):
+        # TODO: send here FrameStats
+        self.zmq_pub.publish(message, topic=self.topic)
+
+    def start(self, *args, **kwargs):
+        pass
+
+    def stop(self, *args, **kwargs):
+        self.zmq_pub.close()
+
+
+class ZMQPub:
+    def __init__(self, address="tcp://127.0.0.1:5555"):
+        self.address = address
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.PUB)
+        logging.info(f"ZMQ Publisher started at {self.address}")
+        self.socket.bind(self.address)  # Bind the socket to the address
+
+    def publish(self, message: str | dict, topic=None):
+        msg = json.dumps(message) if type(message) is dict else message
+        logging.debug(f"ZMQ publish {msg} to {topic} ({self.address})")
+        if topic is not None:
+            self.socket.send_string(f"{topic} {msg}")
+        else:
+            # Send the message
+            self.socket.send_string(msg)
+
+    def close(self):
+        self.socket.close()
+        logging.info(f"ZMQ Publisher closed for {self.address}")

foosball/tracking/analyzer/ScoreAnalyzer.py

@@ -5,7 +5,8 @@
 from typing import Optional
 
 from . import AbstractAnalyzer
-from ...hooks.web import Webhook
+from ...hooks.zmq import ZMQHook
+from ...hooks.webhook import Webhook
 from ...hooks.audio import AudioHook
 from ...models import Team, Goals, Score, Track, Verbosity, Info
 from ...pipe.BaseProcess import Msg
@@ -20,9 +21,11 @@ class ScoreAnalyzerResult:
 
 class ScoreAnalyzer(AbstractAnalyzer):
     def close(self):
-        pass
+        if self.hooks is not None and isinstance(self.hooks, list):
+            for h in self.hooks:
+                h.stop()
 
-    def __init__(self, goal_grace_period_sec: float = 1.0, audio: bool = False, webhook: bool = False, *args, **kwargs):
+    def __init__(self, goal_grace_period_sec: float = 1.0, *args, **kwargs):
         super().__init__(name="ScoreAnalyzer")
         self.kwargs = kwargs
         self.goal_grace_period_sec = goal_grace_period_sec
@@ -31,7 +34,14 @@ def __init__(self, goal_grace_period_sec: float = 1.0, audio: bool = False, webh
         self.last_track_sighting: dt.datetime | None = None
         self.last_track: Optional[Track] = None
         self.goal_candidate = None
-        self.hooks = [AudioHook("goal"), Webhook.load_webhook('goal_webhook.yaml')]
+        # TODO: create them somewhere else and pass them down
+        self.hooks = [
+            AudioHook("goal"),
+            Webhook.load_webhook('goal_webhook.yaml'),
+            ZMQHook(host="localhost", port=5555, topic="ws")
+        ]
+        for h in self.hooks:
+            h.start()
 
     @staticmethod
     def is_track_empty(track: Track):

foosball/tracking/analyzer/analyze.py

@@ -8,12 +8,10 @@ class Analyzer(BaseProcess):
     def close(self):
         pass
 
-    def __init__(self, audio: bool = False, webhook: bool = False, goal_grace_period_sec: float = 1.0, *args, **kwargs):
+    def __init__(self, goal_grace_period_sec: float = 1.0, *args, **kwargs):
         super().__init__(name="Analyzer")
         self.kwargs = kwargs
-        self.analyzers = [ScoreAnalyzer(goal_grace_period_sec, audio, webhook, args, kwargs)]
-        self.audio = audio
-        self.webhook = webhook
+        self.analyzers = [ScoreAnalyzer(goal_grace_period_sec, args, kwargs)]
 
     def reset_score(self):
         for a in self.analyzers:

foosball/tracking/analyzer/models.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass
+
+from foosball.models import Goals, Track, Blob, Score, FrameDimensions
+
+
+@dataclass
+class FrameStats:
+    goals: Goals | None
+    track: Track | None
+    ball: Blob | None
+    score: Score | None
+    dims: FrameDimensions | None

requirements.txt

@@ -6,4 +6,5 @@ playsound==1.3.0
 PyGObject==3.48.2
 PyYAML==6.0.2
 tqdm==4.66.5
-vidgear==0.3.3
+vidgear==0.3.3
+pyzmq==26.2.0

requirements_dev.txt

@@ -10,4 +10,5 @@ pytest-timeout==2.3.1
 pytest_httpserver==1.1.0
 PyYAML==6.0.2
 tqdm==4.66.5
-vidgear==0.3.3
+vidgear==0.3.3
+pyzmq==26.2.0