Exec 377 store to local file

api/Pipfile

@@ -11,6 +11,7 @@ opentrons-shared-data = { editable = true, path = "../shared-data/python" }
 opentrons = { editable = true, path = "." }
 opentrons-hardware = { editable = true, path = "./../hardware", extras=["FLEX"] }
 numpy = "==1.22.3"
+performance-metrics = {file = "../performance-metrics", editable = true}
 
 [dev-packages]
 # atomicwrites and colorama are pytest dependencies on windows,

api/src/opentrons/__init__.py

@@ -21,8 +21,12 @@
     ROBOT_FIRMWARE_DIR,
 )
 from opentrons.util import logging_config
+from opentrons_shared_data.robot.dev_types import RobotTypeEnum
+from opentrons.config import get_performance_metrics_data_dir
+from opentrons.config.feature_flags import enable_performance_metrics
 from opentrons.protocols.types import ApiDeprecationError
 from opentrons.protocols.api_support.types import APIVersion
+from performance_metrics import RobotContextTracker
 
 from ._version import version
 
@@ -55,6 +59,8 @@ def __dir__() -> List[str]:
 
 SMOOTHIE_HEX_RE = re.compile("smoothie-(.*).hex")
 
+_robot_context_tracker: RobotContextTracker | None = None
+
 
 def _find_smoothie_file() -> Tuple[Path, str]:
     resources: List[Path] = []
@@ -83,7 +89,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
 
@@ -139,6 +144,22 @@ async def _create_thread_manager() -> ThreadManagedHardware:
     return thread_manager
 
 
+def get_robot_context_tracker() -> RobotContextTracker:
+    global _robot_context_tracker
+
+    if _robot_context_tracker:
+        return _robot_context_tracker
+    else:
+        robot_type = robot_configs.load().model
+        _robot_context_tracker = RobotContextTracker(
+            storage_dir=get_performance_metrics_data_dir(),
+            should_track=enable_performance_metrics(
+                RobotTypeEnum.robot_literal_to_enum(robot_type)
+            ),
+        )
+        return _robot_context_tracker
+
+
 async def initialize() -> ThreadManagedHardware:
     """
     Initialize the Opentrons hardware returning a hardware instance.

api/src/opentrons/cli/analyze.py

@@ -28,6 +28,9 @@
 )
 
 from opentrons_shared_data.robot.dev_types import RobotType
+from opentrons import get_robot_context_tracker
+
+_robot_context_tracker = get_robot_context_tracker()
 
 
 @click.command()
@@ -63,6 +66,7 @@ def _get_input_files(files_and_dirs: Sequence[Path]) -> List[Path]:
     return results
 
 
+@_robot_context_tracker.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/tests/opentrons/cli/test_cli.py

@@ -1,4 +1,6 @@
 """Test cli execution."""
+
+
 import json
 import tempfile
 import textwrap
@@ -9,8 +11,15 @@
 
 import pytest
 from click.testing import CliRunner
+from opentrons 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()
+context_tracker._should_track = True
+
+from opentrons.cli.analyze import analyze  # noqa: E402
 
 
 def _list_fixtures(version: int) -> Iterator[Path]:
@@ -242,3 +251,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_tracking_of_analyis_with_robot_context_tracker(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)
+
+    _get_analysis_result([protocol_source_file])
+
+    assert len(context_tracker._storage) == before_analysis + 1

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/robot_context_tracker.py

@@ -9,6 +9,7 @@
 from typing import Callable, TypeVar
 from typing_extensions import ParamSpec
 from collections import deque
+from performance_metrics.constants import PerformanceMetricsFilename
 from performance_metrics.datashapes import (
     RawContextData,
     RobotContextState,
@@ -21,13 +22,15 @@
 class RobotContextTracker:
     """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
+    def _track(self, state: RobotContextState) -> Callable:  # type: ignore
         """Decorator factory for tracking the execution duration and state of robot operations.
 
         Args:
@@ -63,13 +66,21 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
 
         return inner_decorator
 
+    def track_analysis(self) -> Callable:  # type: ignore
+        """Decorator for tracking the analysis of a protocol."""
+        return self._track(RobotContextState.ANALYZING_PROTOCOL)
+
+    def track_startup(self) -> Callable:  # type: ignore
+        """Decorator for tracking the startup of the robot."""
+        return self._track(RobotContextState.STARTING_UP)
+
     def store(self) -> None:
         """Returns the stored context data and clears the storage list."""
         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:
+        os.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

@@ -18,29 +18,36 @@
 @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:
     """Tests the tracking of various robot context states through RobotContextTracker."""
 
-    @robot_context_tracker.track(state=RobotContextState.STARTING_UP)
+    @robot_context_tracker._track(state=RobotContextState.STARTING_UP)
     def starting_robot() -> None:
         sleep(STARTING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.CALIBRATING)
+    @robot_context_tracker._track(state=RobotContextState.CALIBRATING)
     def calibrating_robot() -> None:
         sleep(CALIBRATING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.ANALYZING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.ANALYZING_PROTOCOL)
     def analyzing_protocol() -> None:
         sleep(ANALYZING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.RUNNING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.RUNNING_PROTOCOL)
     def running_protocol() -> None:
         sleep(RUNNING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.SHUTTING_DOWN)
+    @robot_context_tracker._track(state=RobotContextState.SHUTTING_DOWN)
     def shutting_down_robot() -> None:
         sleep(SHUTTING_DOWN_TIME)
 
@@ -78,11 +85,11 @@ def test_multiple_operations_single_state(
 ) -> None:
     """Tests tracking multiple operations within a single robot context state."""
 
-    @robot_context_tracker.track(state=RobotContextState.RUNNING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.RUNNING_PROTOCOL)
     def first_operation() -> None:
         sleep(RUNNING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.RUNNING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.RUNNING_PROTOCOL)
     def second_operation() -> None:
         sleep(RUNNING_TIME)
 
@@ -104,7 +111,7 @@ def test_exception_handling_in_tracked_function(
 ) -> None:
     """Ensures exceptions in tracked operations are handled correctly."""
 
-    @robot_context_tracker.track(state=RobotContextState.SHUTTING_DOWN)
+    @robot_context_tracker._track(state=RobotContextState.SHUTTING_DOWN)
     def error_prone_operation() -> None:
         sleep(SHUTTING_DOWN_TIME)
         raise RuntimeError("Simulated operation failure")
@@ -126,7 +133,7 @@ async def test_async_operation_tracking(
 ) -> None:
     """Tests tracking of an asynchronous operation."""
 
-    @robot_context_tracker.track(state=RobotContextState.ANALYZING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.ANALYZING_PROTOCOL)
     async def async_analyzing_operation() -> None:
         await asyncio.sleep(ANALYZING_TIME)
 
@@ -146,7 +153,7 @@ async def test_async_operation_timing_accuracy(
 ) -> None:
     """Tests the timing accuracy of an async operation tracking."""
 
-    @robot_context_tracker.track(state=RobotContextState.RUNNING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.RUNNING_PROTOCOL)
     async def async_running_operation() -> None:
         await asyncio.sleep(RUNNING_TIME)
 
@@ -165,7 +172,7 @@ async def test_exception_in_async_operation(
 ) -> None:
     """Ensures exceptions in tracked async operations are correctly handled."""
 
-    @robot_context_tracker.track(state=RobotContextState.SHUTTING_DOWN)
+    @robot_context_tracker._track(state=RobotContextState.SHUTTING_DOWN)
     async def async_error_prone_operation() -> None:
         await asyncio.sleep(SHUTTING_DOWN_TIME)
         raise RuntimeError("Simulated async operation failure")
@@ -187,11 +194,11 @@ async def test_concurrent_async_operations(
 ) -> None:
     """Tests tracking of concurrent async operations."""
 
-    @robot_context_tracker.track(state=RobotContextState.CALIBRATING)
+    @robot_context_tracker._track(state=RobotContextState.CALIBRATING)
     async def first_async_calibrating() -> None:
         await asyncio.sleep(CALIBRATING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.CALIBRATING)
+    @robot_context_tracker._track(state=RobotContextState.CALIBRATING)
     async def second_async_calibrating() -> None:
         await asyncio.sleep(CALIBRATING_TIME)
 
@@ -210,7 +217,7 @@ def test_no_tracking(tmp_path: Path) -> None:
     """Tests that operations are not tracked when tracking is disabled."""
     robot_context_tracker = RobotContextTracker(tmp_path, should_track=False)
 
-    @robot_context_tracker.track(state=RobotContextState.STARTING_UP)
+    @robot_context_tracker._track(state=RobotContextState.STARTING_UP)
     def operation_without_tracking() -> None:
         sleep(STARTING_TIME)
 
@@ -223,18 +230,17 @@ 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)
+    @robot_context_tracker._track(state=RobotContextState.STARTING_UP)
     def starting_robot() -> None:
         sleep(STARTING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.CALIBRATING)
+    @robot_context_tracker._track(state=RobotContextState.CALIBRATING)
     def calibrating_robot() -> None:
         sleep(CALIBRATING_TIME)
 
-    @robot_context_tracker.track(state=RobotContextState.ANALYZING_PROTOCOL)
+    @robot_context_tracker._track(state=RobotContextState.ANALYZING_PROTOCOL)
     def analyzing_protocol() -> None:
         sleep(ANALYZING_TIME)
 
@@ -244,7 +250,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