Sauravbanna/ml testing

.gitignore

@@ -58,7 +58,7 @@ src/external/external
 
 # Misc. Log Files
 *.log
-*.csv
+
 
 ### Node ###
 
@@ -162,3 +162,4 @@ src/external
 # macOS folder info
 .DS_Store
 
+*_venv/

scripts/compile_pip_requirements.sh

@@ -20,3 +20,4 @@ bazel run //software/embedded/ansible:requirements.update
 bazel run //software/simulated_tests:requirements.update
 bazel run //software/embedded/robot_diagnostics_cli:requirements.update
 bazel run //starlark/nanopb:requirements.update
+bazel run //software/ml:requirements.update

src/MODULE.bazel

@@ -70,6 +70,13 @@ pip.parse(
     requirements_lock = "//starlark/nanopb:requirements_lock.txt",
 )
 use_repo(pip, "nanopb_deps")
+pip.parse(
+    hub_name = "ml_deps",
+    python_interpreter = "/opt/tbotspython/bin/python",
+    python_version = "3.12",
+    requirements_lock = "//software/ml:requirements_lock.txt",
+)
+use_repo(pip, "ml_deps")
 
 ##############################################
 # Configure Dependencies

src/software/evaluation/loggers/BUILD

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

src/software/evaluation/loggers/pass_logger.py

@@ -0,0 +1,123 @@
+from software.thunderscope.proto_unix_io import ProtoUnixIO
+from software.thunderscope.constants import PassResultsConstants
+import os
+from proto.import_all_protos import *
+from software.evaluation.logs.event_log import (
+    Team as TeamEnum,
+)
+from software.evaluation.trackers.pass_log_tracker import (
+    PassLogTracker,
+)
+from software.evaluation.logs.pass_log import (
+    PassLog,
+)
+import queue
+
+
+class PassLogger:
+    """Class to track passes.
+
+    i.e When a pass happens, we want to log the game state and the pass itself
+    As well as the game state at certain intervals after the pass
+    to see the outcomes of the pass
+    """
+
+    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,
+    ):
+        """Initializes the pass results tracker
+
+        :param proto_unix_io: the proto unix io to use
+        :param friendly_colour_yellow: if the friendly color is yellow or not
+        :param out_file_name: name of file to write pass results to.
+                              If None, uses the value from constants
+        :param buffer_size: buffer size to use
+        """
+        self.friendly_colour_yellow = friendly_colour_yellow
+
+        # track pass results to this queue
+        self.pass_result_queue = queue.Queue(self.EVENT_BUFFER_SIZE)
+
+        self.pass_tracker = PassLogTracker(
+            proto_unix_io=proto_unix_io,
+            from_team=(
+                TeamEnum.YELLOW if self.friendly_colour_yellow else TeamEnum.BLUE
+            ),
+            event_queue=self.pass_result_queue,
+            buffer_size=buffer_size,
+        )
+
+        self.events_file_path = os.path.join(
+            PassResultsConstants.PASS_RESULTS_DIRECTORY_PATH,
+            PassResultsConstants.PASS_RESULTS_FILE_NAME
+            if out_file_name is None
+            else out_file_name,
+        )
+        self.events_file_handle = None
+
+    def _get_team(self, is_friendly: bool) -> TeamEnum:
+        """Gets the correct Team enum value for either the current friendly or enemy team
+
+        :param is_friendly: whether to return the team for friendly or enemy
+        :return: the corresponding Team Enum value
+        """
+        return (
+            TeamEnum.YELLOW
+            if (self.friendly_colour_yellow == is_friendly)
+            else TeamEnum.BLUE
+        )
+
+    def __enter__(self):
+        """Creates any missing directories and opens an append 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):
+        """Flushes content and closes the log file"""
+        if self.events_file_handle:
+            self.events_file_handle.flush()
+            self.events_file_handle.close()
+
+    def refresh(self) -> None:
+        """Refreshes the tracker so we stay up to date on new passes
+        and checks to see if any passes are older than their interval
+        """
+        self.pass_tracker.refresh()
+
+        if not self.events_file_handle:
+            return
+
+        while not self.pass_result_queue.empty():
+            try:
+                # Get item without blocking
+                pass_log = self.pass_result_queue.get_nowait()
+
+                self._log_pass(pass_log)
+            except queue.Empty:
+                return
+
+    def _log_pass(self, pass_log: PassLog):
+        """Logs a single pass to file
+
+        :param result: the result to log
+        """
+        if not self.events_file_handle:
+            return
+
+        try:
+            csv_row = pass_log.to_csv_row()
+            self.events_file_handle.write(csv_row + "\n")
+            self.events_file_handle.flush()
+
+        except (IOError, FileNotFoundError, PermissionError):
+            pass

src/software/evaluation/loggers/stats_logger.py

@@ -0,0 +1,166 @@
+import os
+
+from software.evaluation.trackers import (
+    PossessionTracker,
+    ShotTracker,
+    PassTracker,
+    TrackerBuilder,
+    RefereeTracker,
+    GoalieTracker,
+)
+from dataclasses import dataclass
+from software.thunderscope.proto_unix_io import ProtoUnixIO
+from software.thunderscope.constants import RuntimeManagerConstants
+from software.evaluation.logs.event_log import Team as TeamEnum, EventLog
+import logging
+from proto.import_all_protos import *
+import queue
+
+
+@dataclass
+class FSStats:
+    """Stats for how well a FullSystem is performing"""
+
+    num_yellow_cards: int = 0
+    num_red_cards: int = 0
+    num_scores: int = 0
+
+    num_shots_on_net: int = 0
+    num_enemy_shots_blocked: int = 0
+
+
+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=(
+                    TeamEnum.YELLOW if self.friendly_colour_yellow else TeamEnum.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=(
+                        TeamEnum.YELLOW
+                        if self.friendly_colour_yellow
+                        else TeamEnum.BLUE
+                    ),
+                    for_team=(
+                        TeamEnum.BLUE
+                        if self.friendly_colour_yellow
+                        else TeamEnum.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:
+        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/evaluation/logs/BUILD

@@ -0,0 +1,49 @@
+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 = "world_state_log",
+    srcs = ["world_state_log.py"],
+    data = [
+        "//software:py_constants.so",
+    ],
+    deps = [
+        ":log_interface",
+    ],
+)
+
+py_library(
+    name = "event_log",
+    srcs = ["event_log.py"],
+    deps = [
+        ":log_interface",
+        ":world_state_log",
+    ],
+)
+
+py_library(
+    name = "pass_log",
+    srcs = ["pass_log.py"],
+    deps = [
+        ":event_log",
+        ":log_interface",
+    ],
+)
+
+py_library(
+    name = "logs",
+    deps = [
+        ":event_log",
+        ":log_interface",
+        ":pass_log",
+        ":world_state_log",
+    ],
+)