sync with edge

Merge branch 'edge' into app_fix-robot-rename-regex

api/src/opentrons/config/gripper_config.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+from dataclasses import dataclass
+import logging
+from typing import Tuple
+from typing_extensions import Literal
+
+log = logging.getLogger(__name__)
+
+GripperName = Literal["gripper"]
+GripperModel = Literal["gripper_v1"]
+
+
+@dataclass(frozen=True)
+class GripperConfig:
+    gripper_offset: Tuple[float, float, float]
+    gripper_current: float
+    display_name: str
+    name: GripperName
+    max_travel: float
+    home_position: float
+    steps_per_mm: float
+    idle_current: float
+    model: GripperModel

api/src/opentrons/hardware_control/__init__.py

@@ -15,6 +15,7 @@
 from .pause_manager import PauseManager
 from .backends import Controller, Simulator
 from .pipette import Pipette
+from .gripper import Gripper
 from .types import (
     CriticalPoint,
     NoTipAttachedError,
@@ -27,15 +28,18 @@
 from .execution_manager import ExecutionManager
 from .threaded_async_lock import ThreadedAsyncLock, ThreadedAsyncForbidden
 from .protocols import HardwareControlAPI
+from .instrument_abc import AbstractInstrument
 
 ThreadManagedHardware = ThreadManager[HardwareControlAPI]
 SyncHardwareAPI = SynchronousAdapter[HardwareControlAPI]
 
 __all__ = [
     "API",
+    "AbstractInstrument",
     "Controller",
     "Simulator",
     "Pipette",
+    "Gripper",
     "PauseManager",
     "SynchronousAdapter",
     "HardwareControlAPI",

api/src/opentrons/hardware_control/backends/controller.py

@@ -34,7 +34,7 @@
 if TYPE_CHECKING:
     from opentrons_shared_data.pipette.dev_types import PipetteModel, PipetteName
     from ..dev_types import (
-        AttachedInstrument,
+        AttachedPipette,
         AttachedInstruments,
         InstrumentHardwareConfigs,
     )
@@ -173,7 +173,7 @@ async def fast_home(self, axes: Sequence[str], margin: float) -> Dict[str, float
 
     async def _query_mount(
         self, mount: Mount, expected: Union[PipetteModel, PipetteName, None]
-    ) -> AttachedInstrument:
+    ) -> AttachedPipette:
         found_model: Optional[
             PipetteModel
         ] = await self._smoothie_driver.read_pipette_model(  # type: ignore

api/src/opentrons/hardware_control/backends/ot3controller.py

@@ -81,7 +81,7 @@
 if TYPE_CHECKING:
     from opentrons_shared_data.pipette.dev_types import PipetteName, PipetteModel
     from ..dev_types import (
-        AttachedInstrument,
+        AttachedPipette,
         InstrumentHardwareConfigs,
     )
     from opentrons.drivers.rpi_drivers.dev_types import GPIODriverLike
@@ -325,7 +325,7 @@ async def fast_home(
 
     async def get_attached_instruments(
         self, expected: Dict[OT3Mount, PipetteName]
-    ) -> Dict[OT3Mount, AttachedInstrument]:
+    ) -> Dict[OT3Mount, AttachedPipette]:
         """Get attached instruments.
 
         Args:
@@ -344,7 +344,7 @@ def _synthesize_model_name(
 
         def _build_attached_instr(
             attached: ohc_tool_types.PipetteInformation,
-        ) -> AttachedInstrument:
+        ) -> AttachedPipette:
             return {
                 "config": pipette_config.load(
                     _synthesize_model_name(attached.name, attached.model)
@@ -354,7 +354,7 @@ def _build_attached_instr(
 
         def _generate_attached_instrs(
             attached: ohc_tool_types.ToolSummary,
-        ) -> Iterator[Tuple[OT3Mount, AttachedInstrument]]:
+        ) -> Iterator[Tuple[OT3Mount, AttachedPipette]]:
             if attached.left:
                 yield (OT3Mount.LEFT, _build_attached_instr(attached.left))
             if attached.right:
@@ -615,13 +615,9 @@ async def probe_network(self, timeout: float = 5.0) -> None:
         # when that method actually does canbus stuff
         instrs = await self.get_attached_instruments({})
         expected = {NodeId.gantry_x, NodeId.gantry_y, NodeId.head}
-        if instrs.get(OT3Mount.LEFT, cast("AttachedInstrument", {})).get(
-            "config", None
-        ):
+        if instrs.get(OT3Mount.LEFT, cast("AttachedPipette", {})).get("config", None):
             expected.add(NodeId.pipette_left)
-        if instrs.get(OT3Mount.RIGHT, cast("AttachedInstrument", {})).get(
-            "config", None
-        ):
+        if instrs.get(OT3Mount.RIGHT, cast("AttachedPipette", {})).get("config", None):
             expected.add(NodeId.pipette_right)
         present = await probe(self._messenger, expected, timeout)
         self._present_nodes = self._replace_head_node(present)

api/src/opentrons/hardware_control/backends/ot3simulator.py

@@ -47,8 +47,8 @@
 from opentrons_shared_data.pipette.dev_types import PipetteName, PipetteModel
 from opentrons.hardware_control.dev_types import (
     InstrumentHardwareConfigs,
-    InstrumentSpec,
-    AttachedInstrument,
+    PipetteSpec,
+    AttachedPipette,
 )
 from opentrons.drivers.rpi_drivers.dev_types import GPIODriverLike
 
@@ -116,7 +116,7 @@ def __init__(
 
         def _sanitize_attached_instrument(
             passed_ai: Optional[Dict[str, Optional[str]]] = None
-        ) -> InstrumentSpec:
+        ) -> PipetteSpec:
             if not passed_ai or not passed_ai.get("model"):
                 return {"model": None, "id": None}
             if passed_ai["model"] in pipette_config.config_models:
@@ -222,7 +222,7 @@ async def fast_home(
 
     def _attached_to_mount(
         self, mount: OT3Mount, expected_instr: Optional[PipetteName]
-    ) -> AttachedInstrument:
+    ) -> AttachedPipette:
         init_instr = self._attached_instruments.get(mount, {"model": None, "id": None})
         found_model = init_instr["model"]
         back_compat: List["PipetteName"] = []
@@ -273,7 +273,7 @@ def _attached_to_mount(
 
     async def get_attached_instruments(
         self, expected: Dict[OT3Mount, PipetteName]
-    ) -> Dict[OT3Mount, AttachedInstrument]:
+    ) -> Dict[OT3Mount, AttachedPipette]:
         """Get attached instruments.
 
         Args:

api/src/opentrons/hardware_control/backends/simulator.py

@@ -23,9 +23,9 @@
 if TYPE_CHECKING:
     from opentrons_shared_data.pipette.dev_types import PipetteName
     from ..dev_types import (
-        AttachedInstrument,
+        AttachedPipette,
         AttachedInstruments,
-        InstrumentSpec,
+        PipetteSpec,
         InstrumentHardwareConfigs,
     )
     from opentrons.drivers.rpi_drivers.dev_types import GPIODriverLike
@@ -119,7 +119,7 @@ def __init__(
 
         def _sanitize_attached_instrument(
             passed_ai: Optional[Dict[str, Optional[str]]] = None
-        ) -> InstrumentSpec:
+        ) -> PipetteSpec:
             if not passed_ai or not passed_ai.get("model"):
                 return {"model": None, "id": None}
             if passed_ai["model"] in config_models:
@@ -209,7 +209,7 @@ async def fast_home(self, axis: Sequence[str], margin: float) -> Dict[str, float
 
     def _attached_to_mount(
         self, mount: types.Mount, expected_instr: Optional[PipetteName]
-    ) -> AttachedInstrument:
+    ) -> AttachedPipette:
         init_instr = self._attached_instruments.get(mount, {"model": None, "id": None})
         found_model = init_instr["model"]
         back_compat: List["PipetteName"] = []

api/src/opentrons/hardware_control/dev_types.py

@@ -16,19 +16,25 @@
 from opentrons.drivers.types import MoveSplit
 from opentrons.types import Mount
 from opentrons.config.pipette_config import PipetteConfig
+from opentrons.config.gripper_config import GripperConfig
 
 
-class InstrumentSpec(TypedDict):
+class PipetteSpec(TypedDict):
     model: Union[PipetteModel, None]
     id: Optional[str]
 
 
-class AttachedInstrument(TypedDict):
+class AttachedPipette(TypedDict):
     config: Optional[PipetteConfig]
     id: Optional[str]
 
 
-AttachedInstruments = Dict[Mount, AttachedInstrument]
+class AttachedGripper(TypedDict):
+    config: Optional[GripperConfig]
+    id: Optional[str]
+
+
+AttachedInstruments = Dict[Mount, AttachedPipette]
 
 
 EIGHT_CHANNELS = Literal[8]

api/src/opentrons/hardware_control/gripper.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+""" Classes and functions for gripper state tracking
+"""
+from dataclasses import asdict, replace
+import logging
+from typing import Any, Dict, Optional, Union
+
+from opentrons.types import Point
+from opentrons.config import gripper_config
+from .instrument_abc import AbstractInstrument
+from .types import CriticalPoint
+
+RECONFIG_KEYS = {"quirks"}
+
+
+mod_log = logging.getLogger(__name__)
+
+
+class Gripper(AbstractInstrument[gripper_config.GripperConfig]):
+    """A class to gather and track gripper state and configs.
+
+    This class should not touch hardware or call back out to the hardware
+    control API. Its only purpose is to gather state.
+    """
+
+    DictType = Dict[str, Union[str, float, bool]]
+    #: The type of this data class as a dict
+
+    def __init__(
+        self,
+        config: gripper_config.GripperConfig,
+        gripper_id: Optional[str] = None,
+    ) -> None:
+        self._config = config
+        self._name = self._config.name
+        self._model = self._config.model
+        self._gripper_id = gripper_id
+        self._log = mod_log.getChild(
+            self._gripper_id if self._gripper_id else "<unknown>"
+        )
+        # cache a dict representation of config for improved performance of
+        # as_dict.
+        self._config_as_dict = asdict(config)
+
+    @property
+    def config(self) -> gripper_config.GripperConfig:
+        return self._config
+
+    def update_config_item(self, elem_name: str, elem_val: Any) -> None:
+        self._log.info(f"updated config: {elem_name}={elem_val}")
+        self._config = replace(self._config, **{elem_name: elem_val})
+        # Update the cached dict representation
+        self._config_as_dict = asdict(self._config)
+
+    @property
+    def name(self) -> gripper_config.GripperName:
+        return self._name
+
+    @property
+    def model(self) -> gripper_config.GripperModel:
+        return self._model
+
+    @property
+    def gripper_id(self) -> Optional[str]:
+        return self._gripper_id
+
+    def critical_point(self, cp_override: Optional[CriticalPoint] = None) -> Point:
+        """
+        The vector from the gripper's origin to its critical point. The
+        critical point for a pipette is the end of the nozzle if no tip is
+        attached, or the end of the tip if a tip is attached.
+        """
+        # TODO: add critical point implementation
+        return Point(0, 0, 0)
+
+    def __str__(self) -> str:
+        return f"{self._config.display_name}"
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__}: {self._config.display_name} {id(self)}"
+
+    def as_dict(self) -> "Gripper.DictType":
+        self._config_as_dict.update(
+            {
+                "name": self.name,
+                "model": self.model,
+                "gripper_id": self.gripper_id,
+            }
+        )
+        return self._config_as_dict

api/src/opentrons/hardware_control/instrument_abc.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Any, Optional, Generic, TypeVar
+
+from opentrons.types import Point
+from .types import CriticalPoint
+
+
+InstrumentConfig = TypeVar("InstrumentConfig")
+
+
+class AbstractInstrument(ABC, Generic[InstrumentConfig]):
+    """Defines the common methods of an instrument."""
+
+    @property
+    def name(self) -> str:
+        """Return name of the instrument."""
+        ...
+
+    @property
+    def model(self) -> str:
+        """Return model of the instrument."""
+        ...
+
+    @property
+    def config(self) -> InstrumentConfig:
+        """Instrument config in dataclass format."""
+        ...
+
+    @abstractmethod
+    def update_config_item(self, elem_name: str, elem_val: Any) -> None:
+        """Update instrument config item."""
+        ...
+
+    @abstractmethod
+    def critical_point(self, cp_override: Optional[CriticalPoint] = None) -> Point:
+        """Computate critical point of an instrument."""
+        ...

api/src/opentrons/hardware_control/pipette.py

@@ -15,6 +15,7 @@
 from opentrons.config import pipette_config, robot_configs
 from opentrons.config.types import RobotConfig, OT3Config
 from opentrons.drivers.types import MoveSplit
+from .instrument_abc import AbstractInstrument
 from .types import CriticalPoint, BoardRevision, OT3AxisKind
 
 
@@ -33,7 +34,7 @@
 mod_log = logging.getLogger(__name__)
 
 
-class Pipette:
+class Pipette(AbstractInstrument[pipette_config.PipetteConfig]):
     """A class to gather and track pipette state and configs.
 
     This class should not touch hardware or call back out to the hardware