support the enviornmental sensor in the volumetric tests

hardware-testing/hardware_testing/drivers/asair_sensor.py

@@ -7,15 +7,15 @@
 import abc
 import codecs
 import logging
-import random
 import time
-from typing import Tuple
+from typing import Tuple, Optional
 from abc import ABC
 from dataclasses import dataclass
 
+from serial.tools.list_ports import comports  # type: ignore[import]
 import serial  # type: ignore[import]
 from serial.serialutil import SerialException  # type: ignore[import]
-
+from hardware_testing.data import ui
 log = logging.getLogger(__name__)
 
 USB_VID = 0x0403
@@ -66,6 +66,28 @@ def get_reading(self) -> Reading:
         ...
 
 
+def BuildAsairSensor() -> AsairSensorBase:
+    """Try to find and return an Asair sensor, if not found return a simulator."""
+    def _find_port() -> Optional[str]:
+        ports = comports()
+        for check_port in ports:
+            if (check_port.vid, check_port.pid) == (USB_VID, USB_PID):
+                return str(check_port.device)
+            continue
+        return None
+    ui.print_title("Connecting to Environmental sensor")
+    port = _find_port()
+    if port is not None:
+        try:
+            sensor = AsairSensor.connect(port)
+            ui.print_info(f"Found sensor on port {port}")
+            return sensor
+        except SerialException:
+            pass
+    ui.print_info(f"No sensor found with vid:pid {USB_VID}:{USB_PID}")
+    return SimAsairSensor()
+
+
 class AsairSensor(AsairSensorBase):
     """Asair sensor driver."""
 
@@ -152,6 +174,6 @@ class SimAsairSensor(AsairSensorBase):
 
     def get_reading(self) -> Reading:
         """Get a reading."""
-        temp = random.uniform(24.5, 25)
-        relative_hum = random.uniform(45, 40)
+        temp = 25.0
+        relative_hum = 50.0
         return Reading(temperature=temp, relative_humidity=relative_hum)

hardware-testing/hardware_testing/gravimetric/__main__.py

@@ -36,6 +36,7 @@
 from .measurement import DELAY_FOR_MEASUREMENT
 from .trial import TestResources
 from .tips import get_tips
+from hardware_testing.drivers import asair_sensor
 
 # FIXME: bump to v2.15 to utilize protocol engine
 API_LEVEL = "2.13"
@@ -251,6 +252,7 @@ def _main(args: argparse.Namespace, _ctx: ProtocolContext) -> None:
         tip_batch=helpers._get_tip_batch(_ctx.is_simulating()),
         git_description=get_git_description(),
         tips=get_tips(_ctx, pipette, args.tip, all_channels=all_channels_same_time),
+        env_sensor=asair_sensor.BuildAsairSensor(),
     )
 
     if args.photometric:

hardware-testing/hardware_testing/gravimetric/execute.py

@@ -208,6 +208,7 @@ def _record_measurement_and_store(m_type: MeasurementType) -> MeasurementData:
             trial.recorder,
             trial.pipette.mount,
             trial.stable,
+            trial.env_sensor,
             shorten=trial.inspect,
             delay_seconds=trial.scale_delay,
         )
@@ -376,6 +377,7 @@ def _calculate_evaporation(
         test_report,
         liquid_tracker,
         True,
+        resources.env_sensor,
     )
     ui.print_info(f"running {config.NUM_BLANK_TRIALS}x blank measurements")
     mnt = OT3Mount.RIGHT if resources.pipette.mount == "right" else OT3Mount.LEFT
@@ -507,6 +509,7 @@ def run(cfg: config.GravimetricConfig, resources: TestResources) -> None:
             test_report,
             liquid_tracker,
             False,
+            resources.env_sensor,
         )
         for volume in trials.keys():
             actual_asp_list_all = []

hardware-testing/hardware_testing/gravimetric/execute_photometric.py

@@ -109,7 +109,7 @@ def _tag(m_type: MeasurementType) -> str:
     def _record_measurement_and_store(m_type: MeasurementType) -> EnvironmentData:
         m_tag = _tag(m_type)
         m_data = read_environment_data(
-            trial.cfg.pipette_mount, trial.ctx.is_simulating()
+            trial.cfg.pipette_mount, trial.ctx.is_simulating(), trial.env_sensor
         )
         report.store_measurements_pm(trial.test_report, m_tag, m_data)
         _MEASUREMENTS.append(
@@ -396,6 +396,7 @@ def run(cfg: config.PhotometricConfig, resources: TestResources) -> None:
         resources.test_volumes,
         liquid_tracker,
         cfg,
+        resources.env_sensor,
     )
 
     try:

hardware-testing/hardware_testing/gravimetric/measurement/__init__.py

@@ -9,7 +9,7 @@
 
 from .record import GravimetricRecorder, GravimetricRecording
 from .environment import read_environment_data, EnvironmentData, get_average_reading
-
+from hardware_testing.drivers import asair_sensor
 
 RELATIVE_DENSITY_WATER: Final = 1.0
 
@@ -139,11 +139,12 @@ def record_measurement_data(
     recorder: GravimetricRecorder,
     mount: str,
     stable: bool,
+    env_sensor: asair_sensor.AsairSensorBase,
     shorten: bool = False,
     delay_seconds: int = DELAY_FOR_MEASUREMENT,
 ) -> MeasurementData:
     """Record measurement data."""
-    env_data = read_environment_data(mount, ctx.is_simulating())
+    env_data = read_environment_data(mount, ctx.is_simulating(), env_sensor)
     # NOTE: we need to delay some amount, to give the scale time to accumulate samples
     with recorder.samples_of_tag(tag):
         if ctx.is_simulating():

hardware-testing/hardware_testing/gravimetric/measurement/environment.py

@@ -7,6 +7,7 @@
     SensorResponseBad,
 )
 from hardware_testing.opentrons_api.types import OT3Mount
+from hardware_testing.drivers import asair_sensor
 
 
 @dataclass
@@ -21,7 +22,9 @@ class EnvironmentData:
     celsius_liquid: float
 
 
-def read_environment_data(mount: str, is_simulating: bool) -> EnvironmentData:
+def read_environment_data(
+    mount: str, is_simulating: bool, env_sensor: asair_sensor.AsairSensorBase
+) -> EnvironmentData:
     """Read blank environment data."""
     mnt = OT3Mount.LEFT if mount == "left" else OT3Mount.RIGHT
     try:
@@ -31,12 +34,12 @@ def read_environment_data(mount: str, is_simulating: bool) -> EnvironmentData:
         print(e)
         celsius_pipette = 25.0
         humidity_pipette = 50.0
-    # TODO: implement USB environmental sensors
+    env_data = env_sensor.get_reading()
     d = EnvironmentData(
         celsius_pipette=celsius_pipette,
         humidity_pipette=humidity_pipette,
-        celsius_air=25.0,
-        humidity_air=50.0,
+        celsius_air=env_data.temperature,
+        humidity_air=env_data.relative_humidity,
         pascals_air=1000,
         celsius_liquid=25.0,
     )

hardware-testing/hardware_testing/gravimetric/trial.py

@@ -11,6 +11,7 @@
 from . import helpers
 from . import report
 from hardware_testing.data import ui
+from hardware_testing.drivers import asair_sensor
 
 
 @dataclass
@@ -27,6 +28,7 @@ class VolumetricTrial:
     volume: float
     mix: bool
     acceptable_cv: Optional[float]
+    env_sensor: asair_sensor.AsairSensorBase
 
 
 @dataclass
@@ -76,6 +78,7 @@ class TestResources:
     tip_batch: str
     git_description: str
     tips: Dict[int, List[Well]]
+    env_sensor: asair_sensor.AsairSensorBase
 
 
 def build_gravimetric_trials(
@@ -89,6 +92,7 @@ def build_gravimetric_trials(
     test_report: report.CSVReport,
     liquid_tracker: LiquidTracker,
     blank: bool,
+    env_sensor: asair_sensor.AsairSensorBase,
 ) -> Dict[float, Dict[int, List[GravimetricTrial]]]:
     """Build a list of all the trials that will be run."""
     trial_list: Dict[float, Dict[int, List[GravimetricTrial]]] = {}
@@ -120,6 +124,7 @@ def build_gravimetric_trials(
                     scale_delay=cfg.scale_delay,
                     acceptable_cv=None,
                     cfg=cfg,
+                    env_sensor=env_sensor,
                 )
             )
     else:
@@ -153,6 +158,7 @@ def build_gravimetric_trials(
                             scale_delay=cfg.scale_delay,
                             acceptable_cv=None,
                             cfg=cfg,
+                            env_sensor=env_sensor,
                         )
                     )
     return trial_list
@@ -167,6 +173,7 @@ def build_photometric_trials(
     test_volumes: List[float],
     liquid_tracker: LiquidTracker,
     cfg: config.PhotometricConfig,
+    env_sensor: asair_sensor.AsairSensorBase,
 ) -> Dict[float, List[PhotometricTrial]]:
     """Build a list of all the trials that will be run."""
     trial_list: Dict[float, List[PhotometricTrial]] = {}
@@ -188,6 +195,7 @@ def build_photometric_trials(
                     cfg=cfg,
                     mix=cfg.mix,
                     acceptable_cv=None,
+                    env_sensor=env_sensor,
                 )
             )
     return trial_list