-
Notifications
You must be signed in to change notification settings - Fork 179
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore: create module for setting up performance metrics
- Loading branch information
1 parent
9583648
commit 2e3160e
Showing
4 changed files
with
111 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |