feat(api, performance-metrics): track analysis execution

api/Pipfile

@@ -48,3 +48,4 @@ pytest-profiling = "~=1.7.0"
 # TODO(mc, 2022-03-31): upgrade sphinx, remove this subdep pin
 jinja2 = ">=2.3,<3.1"
 hypothesis = "==6.96.1"
+performance-metrics = {file = "../performance-metrics", editable = true}

api/src/opentrons/__init__.py

@@ -23,9 +23,9 @@
 from opentrons.util import logging_config
 from opentrons.protocols.types import ApiDeprecationError
 from opentrons.protocols.api_support.types import APIVersion
-
 from ._version import version
 
+
 HERE = os.path.abspath(os.path.dirname(__file__))
 __version__ = version
 
@@ -83,7 +83,6 @@ def _get_motor_control_serial_port() -> Any:
         # TODO(mc, 2021-08-01): raise a more informative exception than
         # IndexError if a valid serial port is not found
         port = get_ports_by_name(device_name=smoothie_id)[0]
-
     log.info(f"Connecting to motor controller at port {port}")
     return port
 

api/src/opentrons/cli/analyze.py

@@ -28,6 +28,7 @@
 )
 
 from opentrons_shared_data.robot.dev_types import RobotType
+from opentrons.util.performance_helpers import track_analysis
 
 
 @click.command()
@@ -63,6 +64,7 @@ def _get_input_files(files_and_dirs: Sequence[Path]) -> List[Path]:
     return results
 
 
+@track_analysis
 async def _analyze(
     files_and_dirs: Sequence[Path],
     json_output: Optional[AsyncPath],

api/src/opentrons/config/__init__.py

@@ -284,6 +284,13 @@ class ConfigElement(NamedTuple):
         ConfigElementType.DIR,
         "The dir where module calibration is stored",
     ),
+    ConfigElement(
+        "performance_metrics_dir",
+        "Performance Metrics Directory",
+        Path("performance_metrics_data"),
+        ConfigElementType.DIR,
+        "The dir where performance metrics are stored",
+    ),
 )
 #: The available configuration file elements to modify. All of these can be
 #: changed by editing opentrons.json, where the keys are the name elements,
@@ -602,3 +609,7 @@ def get_tip_length_cal_path() -> Path:
 
 def get_custom_tiprack_def_path() -> Path:
     return get_opentrons_path("custom_tiprack_dir")
+
+
+def get_performance_metrics_data_dir() -> Path:
+    return get_opentrons_path("performance_metrics_dir")

api/src/opentrons/util/performance_helpers.py

@@ -0,0 +1,73 @@
+"""Performance helpers for tracking robot context."""
+
+from pathlib import Path
+from opentrons_shared_data.performance.dev_types import (
+    SupportsTracking,
+    F,
+    RobotContextState,
+)
+from opentrons_shared_data.robot.dev_types import RobotTypeEnum
+from opentrons.config import (
+    get_performance_metrics_data_dir,
+    robot_configs,
+    feature_flags as ff,
+)
+from typing import Callable, Type
+
+performance_metrics_dir: Path = get_performance_metrics_data_dir()
+should_track: bool = ff.enable_performance_metrics(
+    RobotTypeEnum.robot_literal_to_enum(robot_configs.load().model)
+)
+
+
+def _handle_package_import() -> Type[SupportsTracking]:
+    """Handle the import of the performance_metrics package.
+
+    If the package is not available, return a stubbed tracker.
+    """
+    try:
+        from performance_metrics import RobotContextTracker
+
+        return RobotContextTracker
+    except ImportError:
+        return StubbedTracker
+
+
+package_to_use = _handle_package_import()
+_robot_context_tracker: SupportsTracking | None = None
+
+
+class StubbedTracker(SupportsTracking):
+    """A stubbed tracker that does nothing."""
+
+    def __init__(self, storage_dir: Path, should_track: bool) -> None:
+        """Initialize the stubbed tracker."""
+        pass
+
+    def track(self, state: RobotContextState) -> Callable[[F], F]:
+        """Return the function unchanged."""
+
+        def inner_decorator(func: F) -> F:
+            """Return the function unchanged."""
+            return func
+
+        return inner_decorator
+
+    def store(self) -> None:
+        """Do nothing."""
+        pass
+
+
+def _get_robot_context_tracker() -> SupportsTracking:
+    """Singleton for the robot context tracker."""
+    global _robot_context_tracker
+    if _robot_context_tracker is None:
+        _robot_context_tracker = package_to_use(performance_metrics_dir, should_track)
+    return _robot_context_tracker
+
+
+def track_analysis(func: F) -> F:
+    """Track the analysis of a protocol."""
+    return _get_robot_context_tracker().track(RobotContextState.ANALYZING_PROTOCOL)(
+        func
+    )

api/tests/opentrons/cli/test_cli.py

@@ -1,4 +1,6 @@
 """Test cli execution."""
+
+
 import json
 import tempfile
 import textwrap
@@ -9,8 +11,17 @@
 
 import pytest
 from click.testing import CliRunner
+from opentrons.util.performance_helpers import _get_robot_context_tracker
+
 
-from opentrons.cli.analyze import analyze
+# Enable tracking for the RobotContextTracker
+# This must come before the import of the analyze CLI
+context_tracker = _get_robot_context_tracker()
+
+# Ignore the type error for the next line, as we're setting a private attribute for testing purposes
+context_tracker._should_track = True  # type: ignore[attr-defined]
+
+from opentrons.cli.analyze import analyze  # noqa: E402
 
 
 def _list_fixtures(version: int) -> Iterator[Path]:
@@ -242,3 +253,24 @@ def test_python_error_line_numbers(
     assert result.json_output is not None
     [error] = result.json_output["errors"]
     assert error["detail"] == expected_detail
+
+
+def test_track_analysis(tmp_path: Path) -> None:
+    """Test that the RobotContextTracker tracks analysis."""
+    protocol_source = textwrap.dedent(
+        """
+        requirements = {"apiLevel": "2.15"}
+
+        def run(protocol):
+            pass
+        """
+    )
+
+    protocol_source_file = tmp_path / "protocol.py"
+    protocol_source_file.write_text(protocol_source, encoding="utf-8")
+
+    before_analysis = len(context_tracker._storage)  # type: ignore[attr-defined]
+
+    _get_analysis_result([protocol_source_file])
+
+    assert len(context_tracker._storage) == before_analysis + 1  # type: ignore[attr-defined]

api/tests/opentrons/util/test_performance_helpers.py

@@ -0,0 +1,28 @@
+"""Tests for performance_helpers."""
+
+from pathlib import Path
+from opentrons_shared_data.performance.dev_types import RobotContextState
+from opentrons.util.performance_helpers import (
+    StubbedTracker,
+    _get_robot_context_tracker,
+)
+
+
+def test_return_function_unchanged() -> None:
+    """Test that the function is returned unchanged when using StubbedTracker."""
+    tracker = StubbedTracker(Path("/path/to/storage"), True)
+
+    def func_to_track() -> None:
+        pass
+
+    assert (
+        tracker.track(RobotContextState.ANALYZING_PROTOCOL)(func_to_track)
+        is func_to_track
+    )
+
+
+def test_singleton_tracker() -> None:
+    """Test that the tracker is a singleton."""
+    tracker = _get_robot_context_tracker()
+    tracker2 = _get_robot_context_tracker()
+    assert tracker is tracker2

performance-metrics/src/performance_metrics/__init__.py

@@ -1 +1,5 @@
 """Opentrons performance metrics library."""
+
+from .robot_context_tracker import RobotContextTracker
+
+__all__ = ["RobotContextTracker"]

performance-metrics/src/performance_metrics/constants.py

@@ -0,0 +1,14 @@
+"""Performance metrics constants."""
+
+from enum import Enum
+from pathlib import Path
+
+
+class PerformanceMetricsFilename(Enum):
+    """Performance metrics filenames."""
+
+    ROBOT_CONTEXT = "robot_context.csv"
+
+    def get_storage_file_path(self, base_path: Path) -> Path:
+        """Builds the full path to the file."""
+        return base_path / self.value

performance-metrics/src/performance_metrics/datashapes.py

@@ -1,40 +1,8 @@
 """Defines data classes and enums used in the performance metrics module."""
 
-from enum import Enum
 import dataclasses
 from typing import Tuple
-
-
-class RobotContextState(Enum):
-    """Enum representing different states of a robot's operation context."""
-
-    STARTING_UP = 0, "STARTING_UP"
-    CALIBRATING = 1, "CALIBRATING"
-    ANALYZING_PROTOCOL = 2, "ANALYZING_PROTOCOL"
-    RUNNING_PROTOCOL = 3, "RUNNING_PROTOCOL"
-    SHUTTING_DOWN = 4, "SHUTTING_DOWN"
-
-    def __init__(self, state_id: int, state_name: str) -> None:
-        self.state_id = state_id
-        self.state_name = state_name
-
-    @classmethod
-    def from_id(cls, state_id: int) -> "RobotContextState":
-        """Returns the enum member matching the given state ID.
-
-        Args:
-            state_id: The ID of the state to retrieve.
-
-        Returns:
-            RobotContextStates: The enum member corresponding to the given ID.
-
-        Raises:
-            ValueError: If no matching state is found.
-        """
-        for state in RobotContextState:
-            if state.state_id == state_id:
-                return state
-        raise ValueError(f"Invalid state id: {state_id}")
+from opentrons_shared_data.performance.dev_types import RobotContextState
 
 
 @dataclasses.dataclass(frozen=True)

performance-metrics/src/performance_metrics/robot_context_tracker.py

@@ -6,18 +6,24 @@
 
 from functools import wraps, partial
 from time import perf_counter_ns
-import os
+from os import makedirs
 from typing import Callable, TypeVar, cast
 
 
 from typing_extensions import ParamSpec
 from collections import deque
-from performance_metrics.datashapes import RawContextData, RobotContextState
+from performance_metrics.constants import PerformanceMetricsFilename
+from performance_metrics.datashapes import (
+    RawContextData,
+)
+from opentrons_shared_data.performance.dev_types import (
+    RobotContextState,
+    SupportsTracking,
+)
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
-
 def _get_timing_function() -> Callable[[], int]:
     """Returns a timing function for the current platform."""
     time_function: Callable[[], int]
@@ -35,13 +41,15 @@ def _get_timing_function() -> Callable[[], int]:
     return time_function
 
 
-class RobotContextTracker:
+class RobotContextTracker(SupportsTracking):
     """Tracks and stores robot context and execution duration for different operations."""
 
-    def __init__(self, storage_file_path: Path, should_track: bool = False) -> None:
+    def __init__(self, storage_dir: Path, should_track: bool = False) -> None:
         """Initializes the RobotContextTracker with an empty storage list."""
         self._storage: deque[RawContextData] = deque()
-        self._storage_file_path = storage_file_path
+        self.storage_file_path = (
+            PerformanceMetricsFilename.ROBOT_CONTEXT.get_storage_file_path(storage_dir)
+        )
         self._should_track = should_track
 
     def track(self, state: RobotContextState) -> Callable:  # type: ignore
@@ -85,8 +93,8 @@ def store(self) -> None:
         stored_data = self._storage.copy()
         self._storage.clear()
         rows_to_write = [context_data.csv_row() for context_data in stored_data]
-        os.makedirs(self._storage_file_path.parent, exist_ok=True)
-        with open(self._storage_file_path, "a") as storage_file:
+        makedirs(self.storage_file_path.parent, exist_ok=True)
+        with open(self.storage_file_path, "a") as storage_file:
             writer = csv.writer(storage_file)
             writer.writerow(RawContextData.headers())
             writer.writerows(rows_to_write)

performance-metrics/tests/performance_metrics/test_robot_context_tracker.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 import pytest
 from performance_metrics.robot_context_tracker import RobotContextTracker
-from performance_metrics.datashapes import RobotContextState
+from opentrons_shared_data.performance.dev_types import RobotContextState
 from time import sleep, time_ns
 from unittest.mock import patch
 
@@ -19,7 +19,14 @@
 @pytest.fixture
 def robot_context_tracker(tmp_path: Path) -> RobotContextTracker:
     """Fixture to provide a fresh instance of RobotContextTracker for each test."""
-    return RobotContextTracker(storage_file_path=tmp_path, should_track=True)
+    return RobotContextTracker(tmp_path, should_track=True)
+
+
+def test_file_path(tmp_path: Path, robot_context_tracker: RobotContextTracker) -> None:
+    """Tests the storage file path for the RobotContextTracker."""
+    assert (
+        robot_context_tracker.storage_file_path == tmp_path / "robot_context.csv"
+    ), "Storage file path should be correctly set."
 
 
 def test_robot_context_tracker(robot_context_tracker: RobotContextTracker) -> None:
@@ -242,8 +249,7 @@ def operation_without_tracking() -> None:
 
 async def test_storing_to_file(tmp_path: Path) -> None:
     """Tests storing the tracked data to a file."""
-    file_path = tmp_path / "test_file.csv"
-    robot_context_tracker = RobotContextTracker(file_path, should_track=True)
+    robot_context_tracker = RobotContextTracker(tmp_path, should_track=True)
 
     @robot_context_tracker.track(state=RobotContextState.STARTING_UP)
     def starting_robot() -> None:
@@ -263,7 +269,7 @@ def analyzing_protocol() -> None:
 
     robot_context_tracker.store()
 
-    with open(file_path, "r") as file:
+    with open(robot_context_tracker.storage_file_path, "r") as file:
         lines = file.readlines()
         assert (
             len(lines) == 4

robot-server/Pipfile

@@ -36,6 +36,7 @@ sqlalchemy2-stubs = "==0.0.2a21"
 # limited by tavern
 python-box = "==6.1.0"
 types-paho-mqtt = "==1.6.0.20240106"
+performance-metrics = {file = "../performance-metrics", editable = true}
 
 [packages]
 anyio = "==3.7.1"