Track Pass Results after Intervals

src/software/thunderscope/constants.py

@@ -421,7 +421,3 @@ class PassResultsConstants:
     FRIENDLY_POSSESSION_SCORE = 2
     ENEMY_POSSESSION_SCORE = -FRIENDLY_POSSESSION_SCORE
     NEUTRAL_SCORE = 0
-
-    # the time intervals to log results for after each pass
-    # so after a pass, wait X seconds and then log game state
-    INTERVALS_S = [1, 5, 10]

src/software/thunderscope/log/pass_results/BUILD

@@ -11,3 +11,15 @@ py_library(
         "//software/thunderscope/log/trackers:tracker",
     ],
 )
+
+py_library(
+    name = "pass_results",
+    srcs = [
+        "pass_event.py",
+        "pass_result_tracker.py",
+        "pass_results.py",
+    ],
+    deps = [
+        "//software/thunderscope/log/trackers:tracker",
+    ],
+)

src/software/thunderscope/log/pass_results/pass_event.py

@@ -6,14 +6,19 @@
     Team,
 )
 from dataclasses import dataclass
-from enum import auto, StrEnum
+from enum import IntEnum
 from proto.import_all_protos import *
 
 
-class PassResultType(StrEnum):
+class PassResultType(IntEnum):
     """Enum for the different types of pass results we want to log"""
 
-    RESULT_PRE = auto()
+    RESULT_1S = 1
+    RESULT_5S = 5
+    RESULT_10S = 10
+    RESULT_20S = 20
+    RESULT_30S = 30
+    RESULT_PRE = 0
 
 
 @dataclass

src/software/thunderscope/log/pass_results/pass_result_tracker.py

@@ -0,0 +1,167 @@
+from software.thunderscope.log.trackers.kick_tracker import PassTracker
+from software.thunderscope.log.pass_results.pass_event import (
+    PassResultType,
+    TrackedPassResult,
+)
+from software.thunderscope.log.trackers.tracked_event import (
+    EventType,
+    Team,
+    get_event_from_world,
+)
+from typing import override
+from dataclasses import dataclass
+from software.thunderscope.constants import PassResultsConstants
+from proto.import_all_protos import *
+from software.thunderscope.proto_unix_io import ProtoUnixIO
+import queue
+
+
+@dataclass
+class LoggedPass:
+    pass_: Pass
+    timestamp: float
+    score: float
+
+
+class PassResultTracker(PassTracker):
+    def __init__(
+        self,
+        proto_unix_io: ProtoUnixIO,
+        from_team: Team,
+        event_queue: queue.Queue,
+        for_team: Team | None = None,
+        **kwargs,
+    ):
+        """Initializes the PassResultTracker
+
+        :param proto_unix_io: the proto unix io to get the game state from
+        :param from_team: the team that this tracker is tracking from (events are from this team)
+        :param for_team: the team that this tracker is tracking for (events are for this team)
+                          default is same as the from_team, but can be different
+        :param event_queue: the queue to write events to
+        """
+        super().__init__(
+            proto_unix_io=proto_unix_io,
+            from_team=from_team,
+            for_team=for_team if for_team is not None else from_team,
+            event_queue=event_queue,
+            **kwargs,
+        )
+
+        self.score = 0
+
+        self.intervals = [item.value for item in PassResultType]
+        self.interval_labels = [item.name for item in PassResultType]
+
+        self.logged_passes_map: dict[int, list[LoggedPass]] = {
+            interval: [] for interval in self.intervals
+        }
+
+    @override
+    def refresh(self) -> None:
+        super().refresh()
+
+        self._update_pass_timestamps()
+
+    @override
+    def write_event(self, event_type: EventType) -> None:
+        """This method is called by the super-class when there is a new pass
+        Adds the pass to the first list of logged passes, at the smallest interval
+
+        :param event_type: (unused)
+        """
+        if not self.curr_pass:
+            return
+
+        if not self.cached_world:
+            return
+
+        self.logged_passes_map[PassResultsConstants.INTERVALS_S[0]].append(
+            LoggedPass(
+                pass_=self.curr_pass,
+                timestamp=self.cached_world.getMostRecentTimestamp().toSeconds(),
+                score=self.score,
+            )
+        )
+
+    def set_score(self, score: float) -> None:
+        """Sets the current game score to the given value
+
+        :param score: the new score
+        """
+        self.score = score
+
+    def _log_pass_result(
+        self, pass_result_type: PassResultType, pass_: Pass, score: float = 0.0
+    ) -> None:
+        """Logs a single pass result, with the given pass, score, and event type
+
+        :param pass_result_type: the result type corresponding to the interval after which
+                                we are logging the result
+        :param pass_: the pass whose results are being logged
+        :param score: the score for this pass's results compared to when the pass started
+        """
+        if not self.cached_world:
+            return
+
+        pass_event = get_event_from_world(
+            world_msg=self.cached_world_msg,
+            event_type=EventType.PASS,
+            from_team=self.from_team,
+            for_team=self.for_team,
+        )
+
+        event = TrackedPassResult(
+            pass_event=pass_event,
+            pass_result_type=pass_result_type,
+            pass_=pass_,
+            score=score,
+        )
+
+        self.event_queue.put(event)
+
+    def _log_if_over_interval(
+        self, logged_pass: LoggedPass, interval: float, pass_result_type: PassResultType
+    ) -> bool:
+        """Checks if the given pass we logged is older than the given interval
+        If so, logs a result with the given type with the correct score
+
+        :param logged_pass: the pass we previously logged, with the timestamp and score at the time
+        :param interval: the interval that the pass may be older than
+        :param pass_result_type: the result type corresponding to this interval
+        :return: True if the pass is older than interval (and therefore result has been logged)
+                 False if not
+        """
+        if (
+            self.cached_world_msg.time_sent.epoch_timestamp_seconds
+            - logged_pass.timestamp
+            > interval
+        ):
+            self._log_pass_result(
+                pass_result_type=pass_result_type,
+                pass_=logged_pass.pass_,
+                score=(self.score - logged_pass.score),
+            )
+
+            return True
+
+        return False
+
+    def _update_pass_timestamps(self):
+        """For all currently logged passes, check if the interval they belong to has passed
+        If so, log that pass result
+        And move them to the next interval if exists
+        """
+        for idx, interval in enumerate(self.intervals):
+            logged_passes = self.logged_passes_map[interval]
+
+            pass_result_type = self.interval_labels[idx]
+
+            # passes are in the list in chronological order
+            while self._log_if_over_interval(
+                logged_passes[0], interval, pass_result_type
+            ):
+                logged_pass = logged_passes.pop(0)
+
+                if idx < len(self.intervals) - 1:
+                    self.logged_passes_map[self.intervals[idx + 1]].append(logged_pass)