enhancement(zmq): Add zmq hook WIP

Add viewbox to Framestats

Author: DarwinsBuddy

foosball/tracking/analyzer/ScoreAnalyzer.py

@@ -72,6 +72,7 @@ def analyze(self, msg: Msg, timestamp: dt.datetime) -> [ScoreAnalyzerResult, [In
         goals = msg.data["Tracker"].goals
         track = msg.data["Tracker"].ball_track
         ball = msg.data["Tracker"].ball
+        viewbox = msg.data["Tracker"].viewbox
         team_scored = None
         try:
             self.check_reset_score()
@@ -103,6 +104,7 @@ def analyze(self, msg: Msg, timestamp: dt.datetime) -> [ScoreAnalyzerResult, [In
             score=self.score,
             ball=ball,
             dims=self.dims,
+            viewbox=viewbox,
             timestamp=timestamp.strftime("%Y-%m-%dT%H:%M:%S")
         )
         self.frame_hooks.invoke(stats.to_dict())

foosball/tracking/analyzer/models.py

@@ -2,13 +2,15 @@
 
 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
+    viewbox: list | None
     timestamp: str | None
 
     def to_dict(self) -> dict:
@@ -18,5 +20,6 @@ def to_dict(self) -> dict:
             "ball": self.ball.to_json() if self.ball else None,
             "score": self.score.to_json() if self.score else None,
             "dims": self.dims.to_json() if self.dims else None,
+            "viewbox": self.viewbox if self.viewbox else None,
             "timestamp": self.timestamp
         }

foosball/tracking/preprocess/__init__.py

@@ -68,6 +68,7 @@ def __init__(self, dims: FrameDimensions, goal_detector: GoalDetector, headless=
         self.calibration_out = Queue() if self.goals_calibration else None
         self.config_in = Queue() if self.goals_calibration else None
         self.goal_detector = goal_detector
+        self.viewbox = None
 
     def config_input(self, config: GoalColorConfig) -> None:
         if self.goals_calibration:
@@ -119,8 +120,10 @@ def process(self, msg: Msg) -> Msg:
 
                 self.frames_since_last_marker_detection = (self.frames_since_last_marker_detection + 1) % self.redetect_markers_frame_threshold
                 if len(self.markers) == 4:
+
                     # crop and warp
-                    preprocessed, self.homography_matrix = self.four_point_transform(frame, self.markers)
+                    preprocessed, self.homography_matrix, marker_pts = self.four_point_transform(frame, self.markers)
+                    self.viewbox = [pt.tolist() for pt in marker_pts]
                     if trigger_marker_detection:
                         # detect goals anew
                         goals_detection_result = self.goal_detector.detect(preprocessed)
@@ -150,7 +153,7 @@ def process(self, msg: Msg) -> Msg:
             traceback.print_exc()
         # Not passing original msg due to performance impact (copying whole frames, etc.)
         return Msg(info=info, data={
-            "Preprocessor": PreprocessorResult(original=self.iproc(frame), preprocessed=self.iproc(preprocessed), homography_matrix=self.homography_matrix, goals=self.goals)
+            "Preprocessor": PreprocessorResult(original=self.iproc(frame), preprocessed=self.iproc(preprocessed), homography_matrix=self.homography_matrix, goals=self.goals, viewbox=self.viewbox)
         })
 
     @staticmethod
@@ -181,7 +184,7 @@ def order_points(pts: np.ndarray) -> np.ndarray:
         # return the ordered coordinates
         return rect
 
-    def four_point_transform(self, frame: Frame, markers: list[Aruco]) -> tuple[Frame, [int, int]]:
+    def four_point_transform(self, frame: Frame, markers: list[Aruco]) -> tuple[Frame, [int, int], [int, int, int, int]]:
         pts = np.array([self.corners2pt(marker.corners) for marker in markers])
         # obtain a consistent order of the points and unpack them
         # individually
@@ -226,7 +229,7 @@ def four_point_transform(self, frame: Frame, markers: list[Aruco]) -> tuple[Fram
         # since we want to draw on the original image (later on), which is still originally not warped,
         # we want to have a function set to project from/onto the warped/un-warped version of the frame
         # for future reference. so we return the warped image and the used homography matrix
-        return warped, homography_matrix
+        return warped, homography_matrix, src_pts
 
 
 def project_point(pt: Point, homography_matrix: np.array, mode: WarpMode):

foosball/tracking/preprocess/models.py

@@ -12,3 +12,4 @@ class PreprocessorResult:
     preprocessed: Optional[Frame]
     homography_matrix: Optional[np.ndarray]  # 3x3 matrix used to warp the image and project points
     goals: Optional[Goals]
+    viewbox: Optional[list]

foosball/tracking/tracker/__init__.py

@@ -38,7 +38,7 @@ def close(self) -> None:
             clear(self.calibration_out)
             self.calibration_out.close()
 
-    def update_ball_track(self, detected_ball: Blob) -> Track:
+    def update_ball_track(self, detected_ball: Blob) -> Track | None:
         if detected_ball is not None:
             center = detected_ball.center
             # update the points queue (track history)
@@ -47,7 +47,7 @@ def update_ball_track(self, detected_ball: Blob) -> Track:
             self.ball_track.appendleft(None)
         return self.ball_track
 
-    def get_info(self, ball_track: Track) -> [Info]:
+    def get_info(self, ball_track: Track | None) -> [Info]:
         info = [
             Info(verbosity=Verbosity.DEBUG, title="Track length", value=f"{str(sum([1 for p in ball_track or [] if p is not None])).rjust(2, ' ')}"),
             Info(verbosity=Verbosity.TRACE, title="Calibration", value=f"{self.calibrationMode if self.calibrationMode is not None else 'off'}"),
@@ -75,6 +75,7 @@ def process(self, msg: Msg) -> Msg:
         ball = None
         goals = data.goals
         ball_track = None
+        viewbox = data.viewbox
         tracker_info = []
         try:
             if not self.off:
@@ -105,6 +106,6 @@ def process(self, msg: Msg) -> Msg:
             traceback.print_exc()
         # Not passing original msg due to performance impact (copying whole frames, etc.)
         if not self.verbose:
-            return Msg(info=tracker_info, data={"Tracker": TrackerResult(frame=data.original, goals=goals, ball_track=ball_track, ball=ball)})
+            return Msg(info=tracker_info, data={"Tracker": TrackerResult(frame=data.original, goals=goals, ball_track=ball_track, ball=ball, viewbox=viewbox)})
         else:
-            return Msg(info=tracker_info, data={"Tracker": TrackerResult(frame=data.preprocessed, goals=goals, ball_track=ball_track, ball=ball)})
+            return Msg(info=tracker_info, data={"Tracker": TrackerResult(frame=data.preprocessed, goals=goals, ball_track=ball_track, ball=ball, viewbox=viewbox)})

foosball/tracking/tracker/models.py

@@ -9,3 +9,4 @@ class TrackerResult:
     goals: Goals | None
     ball_track: Track | None
     ball: Blob | None
+    viewbox: list | None