Merge branch 'master' of https://github.com/UBC-Thunderbots/Software into robocup_2026

Author: williamckha

src/software/py_constants.cpp

@@ -41,6 +41,9 @@ PYBIND11_MODULE(py_constants, m)
         ACCELERATION_DUE_TO_GRAVITY_METERS_PER_SECOND_SQUARED;
     m.attr("ENEMY_BALL_PLACEMENT_DISTANCE_METERS") = ENEMY_BALL_PLACEMENT_DISTANCE_METERS;
 
+    m.attr("BALL_TO_FRONT_OF_ROBOT_DISTANCE_WHEN_DRIBBLING") =
+        BALL_TO_FRONT_OF_ROBOT_DISTANCE_WHEN_DRIBBLING;
+
 
     m.attr("TACTIC_OVERRIDE_PATH")      = TACTIC_OVERRIDE_PATH;
     m.attr("PLAY_OVERRIDE_PATH")        = PLAY_OVERRIDE_PATH;

src/software/stats/loggers/BUILD

@@ -0,0 +1,9 @@
+package(default_visibility = ["//visibility:public"])
+
+py_library(
+    name = "stats_logger",
+    srcs = ["stats_logger.py"],
+    deps = [
+        "//software/stats/trackers:tracker",
+    ],
+)

src/software/stats/loggers/stats_logger.py

@@ -0,0 +1,152 @@
+import os
+
+from software.stats.trackers import (
+    PossessionTracker,
+    ShotTracker,
+    PassTracker,
+    TrackerBuilder,
+    RefereeTracker,
+    GoalieTracker,
+)
+from software.thunderscope.proto_unix_io import ProtoUnixIO
+from software.thunderscope.constants import RuntimeManagerConstants
+from software.stats.logs.event_log import EventLog
+import logging
+from proto.import_all_protos import *
+import queue
+from proto.ssl_gc_common_pb2 import Team
+
+
+class StatsLogger:
+    # From GoalieTacticConfig
+    INCOMING_SHOT_MIN_VELOCITY = 0.2
+
+    EVENT_BUFFER_SIZE = 100
+
+    def __init__(
+        self,
+        proto_unix_io: ProtoUnixIO,
+        friendly_colour_yellow: bool,
+        out_file_name: str | None = None,
+        buffer_size: int = 5,
+        record_enemy_stats: bool = False,
+    ):
+        """Initializes the FullSystem Stats Tracker
+
+        :param friendly_colour_yellow: if the friendly colour is yellow
+        :param out_file_name: name of file to write stats to.
+                              If None, uses the value from constants
+        :param buffer_size: the buffer size for protocol buffers
+        :param record_enemy_stats: if this should record both friendly and enemy stats or just friendly
+        """
+        self.friendly_colour_yellow = friendly_colour_yellow
+
+        self.events_file_path = os.path.join(
+            RuntimeManagerConstants.RUNTIME_EVENTS_DIRECTORY_PATH,
+            RuntimeManagerConstants.RUNTIME_EVENTS_FILE
+            if out_file_name is None
+            else out_file_name,
+        )
+        # initialized in setup()
+        self.events_file_handle = None
+
+        self.event_queue = queue.Queue(self.EVENT_BUFFER_SIZE)
+
+        # flag to turn off logging stats if needed
+        self.logging_enabled = True
+
+        self.tracker = (
+            TrackerBuilder(
+                proto_unix_io=proto_unix_io,
+                from_team=(Team.YELLOW if self.friendly_colour_yellow else Team.BLUE),
+                event_queue=self.event_queue,
+                buffer_size=buffer_size,
+            )
+            .add_tracker(PassTracker)
+            .add_tracker(ShotTracker)
+            .add_tracker(PossessionTracker)
+            .add_tracker(
+                RefereeTracker,
+                friendly_color_yellow=self.friendly_colour_yellow,
+                toggle_logging=self._toggle_logging,
+            )
+            .add_tracker(GoalieTracker, for_friendly=True)
+        )
+
+        self.record_enemy_stats = record_enemy_stats
+        if self.record_enemy_stats:
+            self.enemy_tracker = (
+                TrackerBuilder(
+                    proto_unix_io=proto_unix_io,
+                    from_team=(
+                        Team.YELLOW if self.friendly_colour_yellow else Team.BLUE
+                    ),
+                    for_team=(
+                        Team.BLUE if self.friendly_colour_yellow else Team.YELLOW
+                    ),
+                    event_queue=self.event_queue,
+                    buffer_size=buffer_size,
+                )
+                .add_tracker(
+                    RefereeTracker,
+                    friendly_color_yellow=(not self.friendly_colour_yellow),
+                    toggle_logging=self._toggle_logging,
+                )
+                .add_tracker(GoalieTracker, for_friendly=False)
+            )
+
+    def refresh(self) -> None:
+        """Refreshes the events for the game so far"""
+        self.tracker.refresh()
+
+        if not self.events_file_handle:
+            return
+
+        while not self.event_queue.empty():
+            try:
+                # Get item without blocking
+                event = self.event_queue.get_nowait()
+
+                self._write_event_to_file(event)
+            except queue.Empty:
+                return
+
+    def __enter__(self):
+        """Sets up the file resources for logging
+        Creates any missing directories and stores the file handle
+        """
+        # create temp stats directory if it doesn't exist
+        os.makedirs(os.path.dirname(self.events_file_path), exist_ok=True)
+
+        self.events_file_handle = open(self.events_file_path, "a")
+
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Writes all logs back to file, and cleans up any created file resources after logging"""
+        if self.events_file_handle:
+            self.events_file_handle.flush()
+            self.events_file_handle.close()
+
+    def _toggle_logging(self, should_log: bool) -> None:
+        """Turns logging off or on based on the given boolean
+
+        ;param should_log: True if logging should continue, False if not
+        """
+        self.logging_enabled = should_log
+
+    def _write_event_to_file(self, event: EventLog) -> None:
+        """Write the given stats to the given file
+
+        :param event: the event to write
+        """
+        if not self.events_file_handle:
+            return
+
+        try:
+            csv_row = event.to_csv_row()
+            self.events_file_handle.write(csv_row + "\n")
+            self.events_file_handle.flush()
+
+        except (IOError, FileNotFoundError, PermissionError):
+            logging.warning("Failed to write event to file")

src/software/stats/logs/BUILD

@@ -0,0 +1,46 @@
+package(default_visibility = ["//visibility:public"])
+
+py_library(
+    name = "log_interface",
+    srcs = ["log_interface.py"],
+    deps = [
+        "//proto:import_all_protos",
+        "//software/thunderscope:time_provider",
+    ],
+)
+
+py_library(
+    name = "type_utils",
+    srcs = ["type_utils.py"],
+    deps = [],
+)
+
+py_library(
+    name = "world_state_log",
+    srcs = ["world_state_log.py"],
+    data = [
+        "//software:py_constants.so",
+    ],
+    deps = [
+        ":log_interface",
+        ":type_utils",
+    ],
+)
+
+py_library(
+    name = "event_log",
+    srcs = ["event_log.py"],
+    deps = [
+        ":log_interface",
+        ":world_state_log",
+    ],
+)
+
+py_library(
+    name = "logs",
+    deps = [
+        ":event_log",
+        ":log_interface",
+        ":world_state_log",
+    ],
+)

src/software/stats/logs/event_log.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from enum import StrEnum, auto
+from proto.import_all_protos import *
+from proto.ssl_gc_common_pb2 import Team
+from typing import Any, override
+from software.stats.logs.log_interface import TimestampedEvalLog
+from software.stats.logs.world_state_log import WorldStateLog
+
+
+class EventType(StrEnum):
+    """Enum for the different types of events we want to track"""
+
+    PASS = auto()
+    SHOT_ON_GOAL = auto()
+    ENEMY_SHOT_ON_GOAL = auto()
+    SHOT_BLOCKED = auto()
+    FRIENDLY_POSSESSION_START = auto()
+    FRIENDLY_POSSESSION_END = auto()
+    ENEMY_POSSESSION_START = auto()
+    ENEMY_POSSESSION_END = auto()
+    GAME_START = auto()
+    GAME_END = auto()
+    GOAL_SCORED = auto()
+    YELLOW_CARD = auto()
+    RED_CARD = auto()
+
+
+@dataclass(kw_only=True)
+class EventLog(TimestampedEvalLog):
+    """Represents a single event being tracked, where and for whom the event is,
+    and the game state at the time of the event
+    """
+
+    event_type: EventType
+    from_team: Team
+    for_team: Team
+    world_state_log: WorldStateLog
+
+    num_cols = TimestampedEvalLog.get_num_cols() + 3 + WorldStateLog.get_num_cols()
+
+    @staticmethod
+    def from_world(
+        world_msg: World, event_type: EventType, from_team: Team, for_team: Team
+    ) -> EventLog:
+        """Creates an EventLog from a world protobuf message
+
+        :param world_msg: the world object containing the state of the game
+        :param event_type: the type of event being recorded
+        :param from_team: the team that the event is coming from
+        :param for_team: the team that the event is for
+        :return: a fully populated EventLog including world state
+        """
+        world_state_log = WorldStateLog.from_world(world_msg=world_msg)
+
+        return EventLog(
+            event_type=event_type,
+            from_team=from_team,
+            for_team=for_team,
+            world_state_log=world_state_log,
+        )
+
+    @override
+    @classmethod
+    def get_num_cols(cls) -> int:
+        return EventLog.num_cols
+
+    @override
+    def to_array(self) -> list[Any]:
+        return (
+            super().to_array()
+            + [
+                self.event_type.value,
+                self.from_team,
+                self.for_team,
+            ]
+            + self.world_state_log.to_array()
+        )
+
+    @staticmethod
+    @override
+    def from_csv_row(row_iter: Iterator[str]) -> EventLog | None:
+        """Parses a full CSV row into an EventLog."""
+        timestamp = float(next(row_iter))
+
+        event_type = EventType(next(row_iter))
+        from_team = int(next(row_iter))
+        for_team = int(next(row_iter))
+
+        world_state = WorldStateLog.from_csv_row(row_iter)
+
+        if not world_state:
+            return None
+
+        return EventLog(
+            timestamp=timestamp,
+            event_type=event_type,
+            from_team=from_team,
+            for_team=for_team,
+            world_state_log=world_state,
+        )

src/software/stats/logs/log_interface.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+from abc import abstractmethod, ABC
+from dataclasses import dataclass, field
+from proto.import_all_protos import *
+from typing import Iterator, Any, override
+from software.thunderscope.time_provider import time_provider_instance
+
+
+class IEvalLog(ABC):
+    @classmethod
+    @abstractmethod
+    def get_num_cols(cls) -> int:
+        """Gets the number of columns present in this log"""
+        raise NotImplementedError("Please use the appropriate subclass of log!")
+
+    @abstractmethod
+    def to_array(self) -> list[Any]:
+        """Converts this log to an array of elements"""
+        raise NotImplementedError("Please use the appropriate subclass of log!")
+
+    def to_csv_row(self):
+        """Converts this log into a Comma Separated Values string
+
+        :return: a string of values separated by commas
+        """
+        row_array = self.to_array()
+        assert len(row_array) == type(self).get_num_cols()
+
+        return ",".join([str(elem) for elem in row_array])
+
+    @staticmethod
+    @abstractmethod
+    def from_csv_row(row_iter: Iterator[str], **kwargs) -> IEvalLog | None:
+        """Converts a CSV row into an instance of this log
+
+        :param row_iter: an iterator representing a csv row, which returns elements one by one
+        :param **kwargs: any extra arguments needed for this log not present in the csv row
+        """
+        raise NotImplementedError("Please use the appropriate subclass of log!")
+
+
+@dataclass
+class TimestampedEvalLog(IEvalLog):
+    timestamp: float = field(default_factory=time_provider_instance.elapsed_time_ns)
+
+    def get_timestamp(self) -> float:
+        """Get this log's timestamp"""
+        return self.timestamp
+
+    @classmethod
+    @override
+    def get_num_cols(cls) -> int:
+        return 1
+
+    @override
+    def to_array(self) -> list[Any]:
+        return [self.timestamp]
+
+    @staticmethod
+    @override
+    def from_csv_row(row_iter: Iterator[str], **kwargs) -> TimestampedEvalLog | None:
+        return TimestampedEvalLog(timestamp=float(next(row_iter)))