Skip to content

Commit

Permalink
chore: rename RobotContextTracker to RobotActivityTracker (#15800)
Browse files Browse the repository at this point in the history
# Overview

Closes https://opentrons.atlassian.net/browse/EXEC-623

Renames RobotContextTracker and RobotContextState to
RobotActivityTracker and RobotActivityState. This is to prevent
confusion with the `RobotContext` class added in
#15745

# Test Plan

- [x] Make sure linting and testing passes
  • Loading branch information
DerekMaggio authored Jul 25, 2024
1 parent 7c31fa7 commit 86d75bd
Show file tree
Hide file tree
Showing 9 changed files with 132 additions and 127 deletions.
30 changes: 15 additions & 15 deletions api/src/opentrons/util/performance_helpers.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Performance helpers for tracking robot context."""
"""Performance helpers for tracking robot activity."""

import functools
from pathlib import Path
Expand All @@ -12,7 +12,7 @@
)

if typing.TYPE_CHECKING:
from performance_metrics import RobotContextState, SupportsTracking
from performance_metrics import RobotActivityState, SupportsTracking


_UnderlyingFunctionParameters = typing.ParamSpec("_UnderlyingFunctionParameters")
Expand All @@ -36,7 +36,7 @@ def __init__(self, storage_location: Path, should_track: bool) -> None:

def track(
self,
state: "RobotContextState",
state: "RobotActivityState",
) -> typing.Callable[
[_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn]],
_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn],
Expand Down Expand Up @@ -72,35 +72,35 @@ def _handle_package_import() -> typing.Type["SupportsTracking"]:
If the package is not available, return a stubbed tracker.
"""
try:
from performance_metrics import RobotContextTracker
from performance_metrics import RobotActivityTracker

return RobotContextTracker
return RobotActivityTracker
except ImportError:
return _StubbedTracker


_package_to_use = _handle_package_import()
_robot_context_tracker: typing.Optional["SupportsTracking"] = None
_robot_activity_tracker: typing.Optional["SupportsTracking"] = None

# TODO: derek maggio (06-03-2024): investigate if _should_track should be
# reevaluated each time _get_robot_context_tracker is called. I think this
# reevaluated each time _get_robot_activity_tracker is called. I think this
# might get stuck in a state where after the first call, _should_track is
# always considered the initial value. It might miss changes to the feature
# flag. The easiest way to test this is on a robot when that is working.


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(
def _get_robot_activity_tracker() -> "SupportsTracking":
"""Singleton for the robot activity tracker."""
global _robot_activity_tracker
if _robot_activity_tracker is None:
_robot_activity_tracker = _package_to_use(
get_performance_metrics_data_dir(), _should_track
)
return _robot_context_tracker
return _robot_activity_tracker


def _track_a_function(
state_name: "RobotContextState",
state_name: "RobotActivityState",
func: _UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn],
) -> typing.Callable[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn]:
"""Track a function.
Expand All @@ -115,7 +115,7 @@ def _track_a_function(
Returns:
The decorated function.
"""
tracker: SupportsTracking = _get_robot_context_tracker()
tracker: SupportsTracking = _get_robot_activity_tracker()
wrapped = tracker.track(state=state_name)(func)

@functools.wraps(func)
Expand Down
6 changes: 3 additions & 3 deletions api/tests/opentrons/util/test_performance_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from pathlib import Path
from opentrons.util.performance_helpers import (
_StubbedTracker,
_get_robot_context_tracker,
_get_robot_activity_tracker,
)


Expand All @@ -19,6 +19,6 @@ def func_to_track() -> None:

def test_singleton_tracker() -> None:
"""Test that the tracker is a singleton."""
tracker = _get_robot_context_tracker()
tracker2 = _get_robot_context_tracker()
tracker = _get_robot_activity_tracker()
tracker2 = _get_robot_activity_tracker()
assert tracker is tracker2
8 changes: 4 additions & 4 deletions performance-metrics/src/performance_metrics/__init__.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
"""Opentrons performance metrics library."""

from ._robot_context_tracker import RobotContextTracker
from ._types import RobotContextState, SupportsTracking
from ._robot_activity_tracker import RobotActivityTracker
from ._types import RobotActivityState, SupportsTracking


__all__ = [
"RobotContextTracker",
"RobotContextState",
"RobotActivityTracker",
"RobotActivityState",
"SupportsTracking",
]
18 changes: 9 additions & 9 deletions performance-metrics/src/performance_metrics/_data_shapes.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,34 +4,34 @@
import typing
from pathlib import Path

from ._types import SupportsCSVStorage, StorableData, RobotContextState
from ._types import SupportsCSVStorage, StorableData, RobotActivityState
from ._util import get_timing_function

_timing_function = get_timing_function()


@dataclasses.dataclass(frozen=True)
class RawContextData(SupportsCSVStorage):
"""Represents raw duration data with context state information.
class RawActivityData(SupportsCSVStorage):
"""Represents raw duration data with activity state information.
Attributes:
- function_start_time (int): The start time of the function.
- duration_measurement_start_time (int): The start time for duration measurement.
- duration_measurement_end_time (int): The end time for duration measurement.
- state (RobotContextStates): The current state of the context.
- state (RobotActivityStates): The current state of the activity.
"""

state: RobotContextState
state: RobotActivityState
func_start: int
duration: int

@classmethod
def headers(self) -> typing.Tuple[str, str, str]:
"""Returns the headers for the raw context data."""
"""Returns the headers for the raw activity data."""
return ("state_name", "function_start_time", "duration")

def csv_row(self) -> typing.Tuple[str, int, int]:
"""Returns the raw context data as a string."""
"""Returns the raw activity data as a string."""
return (
self.state,
self.func_start,
Expand All @@ -40,9 +40,9 @@ def csv_row(self) -> typing.Tuple[str, int, int]:

@classmethod
def from_csv_row(cls, row: typing.Sequence[StorableData]) -> SupportsCSVStorage:
"""Returns a RawContextData object from a CSV row."""
"""Returns a RawActivityData object from a CSV row."""
return cls(
state=typing.cast(RobotContextState, row[0]),
state=typing.cast(RobotActivityState, row[0]),
func_start=int(row[1]),
duration=int(row[2]),
)
Expand Down
18 changes: 9 additions & 9 deletions performance-metrics/src/performance_metrics/_metrics_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,20 +13,20 @@


class MetricsStore(typing.Generic[T]):
"""Dataclass to store data for tracking robot context."""
"""Dataclass to store data for tracking robot activity."""

def __init__(self, metadata: MetricsMetadata) -> None:
"""Initialize the metrics store."""
self.metadata = metadata
self._data: typing.List[T] = []
self._data_store: typing.List[T] = []

def add(self, context_data: T) -> None:
def add(self, data: T) -> None:
"""Add data to the store."""
self._data.append(context_data)
self._data_store.append(data)

def add_all(self, context_data: typing.Iterable[T]) -> None:
def add_all(self, data: typing.Iterable[T]) -> None:
"""Add data to the store."""
self._data.extend(context_data)
self._data_store.extend(data)

def setup(self) -> None:
"""Set up the data store."""
Expand All @@ -40,9 +40,9 @@ def setup(self) -> None:

def store(self) -> None:
"""Clear the stored data and write it to the storage file."""
stored_data = self._data.copy()
self._data.clear()
rows_to_write = [context_data.csv_row() for context_data in stored_data]
stored_data = self._data_store.copy()
self._data_store.clear()
rows_to_write = [activity_data.csv_row() for activity_data in stored_data]
with open(self.metadata.data_file_location, "a") as storage_file:
logger.debug(
f"Writing {len(rows_to_write)} rows to {self.metadata.data_file_location}"
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Module for tracking robot context and execution duration for different operations."""
"""Module for tracking robot activity and execution duration for different operations."""

import inspect
from pathlib import Path
Expand All @@ -8,8 +8,8 @@
import typing

from ._metrics_store import MetricsStore
from ._data_shapes import RawContextData, MetricsMetadata
from ._types import SupportsTracking, RobotContextState
from ._data_shapes import RawActivityData, MetricsMetadata
from ._types import SupportsTracking, RobotActivityState
from ._util import get_timing_function

_UnderlyingFunctionParameters = typing.ParamSpec("_UnderlyingFunctionParameters")
Expand All @@ -22,20 +22,20 @@
_timing_function = get_timing_function()


class RobotContextTracker(SupportsTracking):
"""Tracks and stores robot context and execution duration for different operations."""
class RobotActivityTracker(SupportsTracking):
"""Tracks and stores robot activity and execution duration for different operations."""

METADATA_NAME: typing.Final[
typing.Literal["robot_context_data"]
] = "robot_context_data"
typing.Literal["robot_activity_data"]
] = "robot_activity_data"

def __init__(self, storage_location: Path, should_track: bool) -> None:
"""Initializes the RobotContextTracker with an empty storage list."""
self._store = MetricsStore[RawContextData](
"""Initializes the RobotActivityTracker with an empty storage list."""
self._store = MetricsStore[RawActivityData](
MetricsMetadata(
name=self.METADATA_NAME,
storage_dir=storage_location,
headers=RawContextData.headers(),
headers=RawActivityData.headers(),
)
)
self._should_track = should_track
Expand All @@ -45,7 +45,7 @@ def __init__(self, storage_location: Path, should_track: bool) -> None:

def track(
self,
state: RobotContextState,
state: RobotActivityState,
) -> typing.Callable[
[_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn]],
_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn],
Expand All @@ -56,7 +56,7 @@ def track(
Args:
func_to_track: The function to track.
state: The state of the robot context during the function execution.
state: The state of the robot activity during the function execution.
*args: The arguments to pass to the function.
**kwargs: The keyword arguments to pass to the function.
Expand Down Expand Up @@ -90,7 +90,7 @@ async def async_wrapper(
duration_end_time = perf_counter_ns()

self._store.add(
RawContextData(
RawActivityData(
func_start=function_start_time,
duration=duration_end_time - duration_start_time,
state=state,
Expand All @@ -116,7 +116,7 @@ def wrapper(
duration_end_time = perf_counter_ns()

self._store.add(
RawContextData(
RawActivityData(
func_start=function_start_time,
duration=duration_end_time - duration_start_time,
state=state,
Expand All @@ -130,7 +130,7 @@ def wrapper(
return inner_decorator

def store(self) -> None:
"""Returns the stored context data and clears the storage list."""
"""Returns the stored activity data and clears the storage list."""
if not self._should_track:
return
self._store.store()
6 changes: 3 additions & 3 deletions performance-metrics/src/performance_metrics/_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
]


RobotContextState = typing.Literal[
RobotActivityState = typing.Literal[
"ANALYZING_PROTOCOL",
"GETTING_CACHED_ANALYSIS",
"RUNNING_PROTOCOL",
Expand All @@ -21,15 +21,15 @@


class SupportsTracking(typing.Protocol):
"""Protocol for classes that support tracking of robot context."""
"""Protocol for classes that support tracking of robot activity."""

def __init__(self, storage_location: Path, should_track: bool) -> None:
"""Initialize the tracker."""
...

def track(
self,
state: "RobotContextState",
state: "RobotActivityState",
) -> typing.Callable[
[_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn]],
_UnderlyingFunction[_UnderlyingFunctionParameters, _UnderlyingFunctionReturn],
Expand Down
26 changes: 14 additions & 12 deletions performance-metrics/tests/performance_metrics/test_metrics_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@
from pathlib import Path
from time import sleep

from performance_metrics._robot_context_tracker import RobotContextTracker
from performance_metrics._data_shapes import RawContextData
from performance_metrics._robot_activity_tracker import RobotActivityTracker
from performance_metrics._data_shapes import RawActivityData

# Corrected times in seconds
STARTING_TIME = 0.001
Expand All @@ -16,38 +16,40 @@

async def test_storing_to_file(tmp_path: Path) -> None:
"""Tests storing the tracked data to a file."""
robot_context_tracker = RobotContextTracker(tmp_path, should_track=True)
robot_activity_tracker = RobotActivityTracker(tmp_path, should_track=True)

@robot_context_tracker.track("ROBOT_STARTING_UP")
@robot_activity_tracker.track("ROBOT_STARTING_UP")
def starting_robot() -> None:
sleep(STARTING_TIME)

@robot_context_tracker.track("CALIBRATING")
@robot_activity_tracker.track("CALIBRATING")
async def calibrating_robot() -> None:
sleep(CALIBRATING_TIME)

@robot_context_tracker.track("ANALYZING_PROTOCOL")
@robot_activity_tracker.track("ANALYZING_PROTOCOL")
def analyzing_protocol() -> None:
sleep(ANALYZING_TIME)

starting_robot()
await calibrating_robot()
analyzing_protocol()

robot_context_tracker.store()
robot_activity_tracker.store()

with open(robot_context_tracker._store.metadata.data_file_location, "r") as file:
with open(robot_activity_tracker._store.metadata.data_file_location, "r") as file:
lines = file.readlines()
assert len(lines) == 3, "All stored data should be written to the file."

split_lines: list[list[str]] = [
line.replace('"', "").strip().split(",") for line in lines
]
assert all(
RawContextData.from_csv_row(line) for line in split_lines
), "All lines should be valid RawContextData instances."
RawActivityData.from_csv_row(line) for line in split_lines
), "All lines should be valid RawActivityData instances."

with open(robot_context_tracker._store.metadata.headers_file_location, "r") as file:
with open(
robot_activity_tracker._store.metadata.headers_file_location, "r"
) as file:
headers = file.readlines()
assert len(headers) == 1, "Header should be written to the headers file."
assert tuple(headers[0].strip().split(",")) == RawContextData.headers()
assert tuple(headers[0].strip().split(",")) == RawActivityData.headers()
Loading

0 comments on commit 86d75bd

Please sign in to comment.