refactoring(data): Refactoring inter-process data passing WIP

foosball/detectors/color.py

@@ -134,6 +134,7 @@ def detect(self, frame) -> DetectedBall:
             return DetectedBall(ball=detected_blob, frame=detection_frame)
         else:
             logger.error("Ball Detection not possible. Config is None")
+            return DetectedBall(ball=None, frame=None)
 
 
 class GoalColorDetector(GoalDetector[GoalColorConfig]):

foosball/hooks/__init__.py

@@ -4,70 +4,82 @@
 import random
 import threading
 import traceback
-from typing import Mapping, Callable
+from abc import ABC, abstractmethod
+from typing import Mapping, Self
 
 import urllib3
 import yaml
 
-from foosball.models import Team
-
 logger = logging.getLogger(__name__)
 
 
-@dataclasses.dataclass
-class Webhook:
-    method: str
-    url: str
-    json: dict = None
-    headers: Mapping[str, str] = None
-
-
-def load_goal_webhook() -> Callable[[Team], Webhook]:
-    filename = 'goal_webhook.yaml'
-
-    def to_webhook(webhook_dict: dict, team: Team):
-        webhook_dict['json']['team'] = team.value
-        return Webhook(**webhook_dict)
-    if os.path.isfile(filename):
-        with open(filename, 'r') as f:
-            wh = yaml.safe_load(f)
-            return lambda team: to_webhook(wh, team)
-    else:
-        logger.info("No goal webhook configured under 'goal_webhook.yaml'")
-
-
-generate_goal_webhook = load_goal_webhook()
-
-
-def webhook(whook: Webhook):
-    threading.Thread(
-        target=_webhook,
-        args=[whook],
-        daemon=True
-    ).start()
-
-
-def _webhook(whook: Webhook):
-    try:
-        headers = {} if whook.headers is None else whook.headers
-        if whook.json is not None and "content-type" not in headers:
-            headers['content_type'] = 'application/json'
-        response = urllib3.request(whook.method, whook.url, json=whook.json, headers=headers)
-        logger.debug(f"webhook response: {response.status}")
-    except Exception as e:
-        logger.error(f"Webhook failed - {e}")
-        traceback.print_exc()
-
-
-def play_random_sound(folder: str, prefix: str = './assets/audio'):
-    path = f'{prefix}/{folder}'
-    audio_file = random.choice(os.listdir(path))
-    play_sound(f"{path}/{audio_file}")
-
-
-def play_sound(sound_file: str):
-    from playsound import playsound
-    if os.path.isfile(sound_file):
-        playsound(sound_file, block=False)
-    else:
-        logger.warning(f"Audio not found: {sound_file}")
+class Hook(ABC):
+    @abstractmethod
+    def invoke(self, *args, **kwargs):
+        pass
+
+
+class AudioHook(Hook):
+
+    def __init__(self, folder: str):
+        super().__init__()
+        self.folder = folder
+
+    def invoke(self, *args, **kwargs):
+        AudioHook.play_random_sound(self.folder)
+
+    @staticmethod
+    def play_sound(sound_file: str):
+        from playsound import playsound
+        if os.path.isfile(sound_file):
+            playsound(sound_file, block=False)
+        else:
+            logger.warning(f"Audio not found: {sound_file}")
+
+    @staticmethod
+    def play_random_sound(folder: str, prefix: str = './assets/audio'):
+        path = f'{prefix}/{folder}'
+        audio_file = random.choice(os.listdir(path))
+        AudioHook.play_sound(f"{path}/{audio_file}")
+
+
+class Webhook(Hook):
+    def __init__(self, method: str = None, url: str = None, json: dict = None, headers: Mapping[str, str] = None, *args, **kwargs):
+        self.method: str = method
+        self.url: str = url
+        self.json: dict = json
+        self.headers: Mapping[str, str] = headers
+
+    def as_dict(self, json: dict = None) -> dict:
+        d = vars(self)
+        if json is not None:
+            d['json'] = d['json'] | json
+        return d
+
+    def invoke(self, json: dict, *args, **kwargs):
+        threading.Thread(
+            target=Webhook.call,
+            kwargs=self.as_dict(json),
+            daemon=True
+        ).start()
+
+    @staticmethod
+    def call(method: str, url: str, json: dict = None, headers: Mapping[str, str] = None):
+        try:
+            headers = {} if headers is None else headers
+            if json is not None and "content-type" not in headers:
+                headers['content_type'] = 'application/json'
+            response = urllib3.request(method, url, json=json, headers=headers)
+            logger.debug(f"webhook response: {response.status}")
+        except Exception as e:
+            logger.error(f"Webhook failed - {e}")
+            traceback.print_exc()
+
+    @classmethod
+    def load_webhook(cls, filename: str) -> Self:
+        if os.path.isfile(filename):
+            with open(filename, 'r') as f:
+                wh = yaml.safe_load(f)
+                return Webhook(**wh)
+        else:
+            logger.info("No goal webhook configured under 'goal_webhook.yaml'")

foosball/models.py

@@ -1,8 +1,7 @@
 import collections
 from dataclasses import dataclass
-from typing import TypeVar
 from enum import Enum
-from typing import Optional, Union, Generic
+from typing import Optional, Union
 
 import cv2
 import numpy as np
@@ -79,7 +78,7 @@ def area(self):
 
 @dataclass
 class DetectedBall:
-    ball: Blob
+    ball: Optional[Blob]
     frame: np.array
 
 
@@ -138,32 +137,22 @@ def to_string(self):
         return " - ".join([i.to_string() for i in self.infos])
 
 
-R = TypeVar('R')
-
-
 @dataclass
-class Result(Generic[R]):
-    info: InfoLog
-    data: R
+class TrackerResult:
+    frame: Frame
+    goals: Goals | None
+    ball_track: Track | None
+    ball: Blob | None
 
 
 @dataclass
-class TrackerResultData:
-    frame: CPUFrame
-    goals: Optional[Goals]
-    ball_track: Track
-    ball: Blob
-
-
-TrackerResult = Result[TrackerResultData]
-
-
-@dataclass
-class PreprocessorResultData:
-    original: CPUFrame
-    preprocessed: Optional[CPUFrame]
+class PreprocessorResult:
+    original: Frame
+    preprocessed: Optional[Frame]
     homography_matrix: Optional[np.ndarray]  # 3x3 matrix used to warp the image and project points
     goals: Optional[Goals]
 
 
-PreprocessorResult = Result[PreprocessorResultData]
+@dataclass
+class RendererResult:
+    frame: Optional[Frame]

foosball/pipe/BaseProcess.py

@@ -1,12 +1,12 @@
 import abc
+import datetime as dt
 import logging
 import multiprocessing
 import traceback
-import datetime as dt
-from queue import Empty, Full
 from dataclasses import dataclass
+from queue import Empty, Full
 
-from foosball.models import InfoLog, Result, R
+from foosball.models import InfoLog
 from foosball.pipe.Pipe import clear, SENTINEL
 
 
@@ -25,11 +25,9 @@ def add(self, name: str, data: any, info=InfoLog([])):
         else:
             self.info = InfoLog([])
 
-    def remove(self, name) -> Result[R]:
+    def remove(self, name) -> any:
         return self.kwargs.pop(name)
 
-
-
     def __init__(self, args=None, kwargs=None, timestamp=dt.datetime.now()):
         if kwargs is None:
             kwargs = dict()

foosball/source/__init__.py

@@ -4,6 +4,7 @@
 from multiprocessing import Queue
 from queue import Full
 from threading import Thread
+from typing import Tuple
 
 from foosball.models import Frame
 from foosball.pipe.BaseProcess import Msg
@@ -88,7 +89,7 @@ def close_capture(self) -> None:
         pass
 
     @abstractmethod
-    def dim(self) -> [int, int]:
+    def dim(self) -> Tuple[int, int]:
         pass
 
     def stop(self):

foosball/source/gear.py

@@ -47,4 +47,4 @@ def dim(self):
         width = int(self.gear.stream.stream.get(cv2.CAP_PROP_FRAME_WIDTH))
         height = int(self.gear.stream.stream.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
-        return [width, height]
+        return tuple((width, height))

foosball/source/opencv.py

@@ -24,4 +24,4 @@ def dim(self):
         width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
         height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
-        return [width, height]
+        return tuple((width, height))

foosball/tracking/__init__.py

@@ -33,8 +33,7 @@ class Tracking:
     def __init__(self, stream, dims: FrameDimensions, goal_detector: GoalDetector, ball_detector: BallDetector, headless=False, maxPipeSize=128, calibrationMode=None, goalGracePeriod=1.0, **kwargs):
         super().__init__()
         self.calibrationMode = calibrationMode
-
-        width, height = dims.scaled
+        (width, height) = dims.scaled
         mask = generate_frame_mask(width, height)
         gpu_flags = kwargs.get(GPU)
         self.preprocessor = PreProcessor(dims, goal_detector, mask=mask, headless=headless, useGPU='preprocess' in gpu_flags,

foosball/tracking/ai.py

@@ -1,6 +1,7 @@
 import logging
 import traceback
 from queue import Empty
+from typing import Tuple
 
 from imutils.video import FPS
 from vidgear.gears import WriteGear
@@ -35,8 +36,7 @@ def __init__(self, source: Source, dis, *args, **kwargs):
         original = self.source.dim()
         self.scale = kwargs.get(SCALE)
         scaled = self.scale_dim(original, self.scale)
-        self.dims = FrameDimensions(original, scaled, self.scale)
-
+        self.dims = FrameDimensions(original=original, scaled=scaled, scale=self.scale)
         self.goal_detector = GoalColorDetector(GoalColorConfig.preset())
         self.ball_detector = BallColorDetector(BallColorConfig.preset(kwargs.get(BALL)))
 
@@ -106,7 +106,8 @@ def step_frame():
                         self.logger.debug("received SENTINEL")
                         break
                     self.fps.update()
-                    frame = msg.kwargs.get('Renderer', None)
+                    result = msg.kwargs.get('Renderer', None)
+                    frame = result.frame if result is not None else None
                     info: InfoLog = msg.info
                     self.fps.stop()
                     fps = int(self.fps.fps())
@@ -167,9 +168,9 @@ def adjust_calibration(self):
                 self.tracking.config_input(self.calibration.config)
 
     @staticmethod
-    def scale_dim(dim, scale_percent):
+    def scale_dim(dim, scale_percent) -> Tuple[int, int]:
 
         # calculate the percent of original dimensions
         width = int(dim[0] * scale_percent)
         height = int(dim[1] * scale_percent)
-        return [width, height]
+        return tuple((width, height))