feat(hardware-testing): add full features for volumetric scripts

hardware-testing/Makefile

@@ -88,6 +88,14 @@ test-cov:
 	$(pytest) $(tests) $(test_opts) $(cov_opts)
 	-$(MAKE) remove-patches-gravimetric
 
+.PHONY: test-photometric-single
+test-photometric-single:
+	-$(MAKE) apply-patches-gravimetric
+	$(python) -m hardware_testing.gravimetric --photometric --simulate --pipette 50 --channels 1 --tip 50
+	$(python) -m hardware_testing.gravimetric --photometric --simulate --pipette 50 --channels 1 --tip 50 --photoplate-col-offset 3
+	$(python) -m hardware_testing.gravimetric --photometric --simulate --pipette 50 --channels 1 --tip 50 --dye-well-col-offset 3
+	-$(MAKE) remove-patches-gravimetric
+
 .PHONY: test-photometric
 test-photometric:
 	-$(MAKE) apply-patches-gravimetric
@@ -115,7 +123,7 @@ test-gravimetric-multi:
 
 .PHONY: test-gravimetric-96
 test-gravimetric-96:
-	$(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 96 --tip 1000 --trials 1 --no-blank
+	$(python) -m hardware_testing.gravimetric --simulate --pipette 1000 --channels 96 --trials 2 --no-blank
 
 .PHONY: test-gravimetric
 test-gravimetric:
@@ -281,10 +289,10 @@ push-ot3-gravimetric:
 	$(MAKE) apply-patches-gravimetric
 	-$(MAKE) sync-sw-ot3
 	$(MAKE) remove-patches-gravimetric
-	scp $(ssh_helper) -r -O hardware_testing/labware/opentrons_flex_96_tiprack_50ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
-	scp $(ssh_helper) -r -O hardware_testing/labware/opentrons_flex_96_tiprack_200ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
-	scp $(ssh_helper) -r -O hardware_testing/labware/opentrons_flex_96_tiprack_1000ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
-	scp $(ssh_helper) -r -O hardware_testing/labware/radwag_pipette_calibration_vial/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
+	scp $(ssh_helper) -r hardware_testing/labware/opentrons_flex_96_tiprack_50ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
+	scp $(ssh_helper) -r hardware_testing/labware/opentrons_flex_96_tiprack_200ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
+	scp $(ssh_helper) -r hardware_testing/labware/opentrons_flex_96_tiprack_1000ul_adp/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
+	scp $(ssh_helper) -r hardware_testing/labware/radwag_pipette_calibration_vial/ root@$(host):/data/labware/v2/custom_definitions/custom_beta/
 
 .PHONY: apply-patches-gravimetric
 apply-patches-gravimetric:

hardware-testing/hardware_testing/data/csv_report.py

@@ -135,7 +135,7 @@ def store(self, *data: Any, print_results: bool = True) -> None:
         self._elapsed_time = time() - self._start_time
         for i, expected_type in enumerate(self._data_types):
             if data[i] is None:
-                self._data[i] = data[i]
+                self._data[i] = None
             else:
                 try:
                     self._data[i] = expected_type(data[i])
@@ -144,8 +144,8 @@ def store(self, *data: Any, print_results: bool = True) -> None:
                         f"[{self.tag}] unexpected data type {type(data[i])} "
                         f'with value "{data[i]}" at index {i}'
                     )
-        self._stored = True
-        if print_results and CSVResult in self._data_types:
+        self._stored = bool(None not in self._data)
+        if self._stored and print_results and CSVResult in self._data_types:
             print_csv_result(self.tag, CSVResult.from_bool(self.result_passed))
 
 

hardware-testing/hardware_testing/data/ui.py

@@ -1,6 +1,7 @@
 """Production QC User Interface."""
 from opentrons.hardware_control import SyncHardwareAPI
 from opentrons.hardware_control.types import StatusBarState
+from time import sleep
 
 PRINT_HEADER_NUM_SPACES = 4
 PRINT_HEADER_DASHES = "-" * PRINT_HEADER_NUM_SPACES
@@ -26,11 +27,16 @@ def get_user_ready(message: str) -> None:
     input(f"WAIT: {message}, press ENTER when ready: ")
 
 
-def alert_user_ready(message: str, hw: SyncHardwareAPI) -> None:
+def alert_user_ready(message: str, hw: SyncHardwareAPI, delay: int = 0) -> None:
     """Flash the ui lights on the ot3 and then use the get_user_ready."""
     hw.set_status_bar_state(StatusBarState.PAUSED)
     get_user_ready(message)
     hw.set_status_bar_state(StatusBarState.CONFIRMATION)
+    if delay > 0:
+        print_info(f"Please wait {delay} seconds:")
+        for sec in range(delay):
+            print(f" - {sec + 1}/{delay}")
+            sleep(0 if hw.is_simulator else 1)
 
 
 def print_title(title: str) -> None:

hardware-testing/hardware_testing/drivers/radwag/commands.py

@@ -142,3 +142,10 @@ class RadwagValueRelease(Enum):
     fast = 1
     fast_reliable = 2
     reliable = 3
+
+
+class RadwagAmbiant(Enum):
+    """Radwag ambiant enviornment states."""
+
+    unstable = 0
+    stable = 1

hardware-testing/hardware_testing/drivers/radwag/driver.py

@@ -1,14 +1,15 @@
 """Radwag Scale Driver."""
 from abc import ABC, abstractmethod
 from typing import Tuple, Optional
-
+import datetime
 from serial import Serial  # type: ignore[import]
 
 from .commands import (
     RadwagCommand,
     RadwagWorkingMode,
     RadwagFilter,
     RadwagValueRelease,
+    RadwagAmbiant,
     radwag_command_format,
 )
 from .responses import RadwagResponse, RadwagResponseCodes, radwag_response_parse
@@ -86,6 +87,7 @@ class RadwagScale(RadwagScaleBase):
     def __init__(self, connection: Serial) -> None:
         """Constructor."""
         self._connection = connection
+        self._raw_log = open("/data/testing_data/scale_raw.txt", "w")
 
     @classmethod
     def create(
@@ -102,6 +104,7 @@ def _write_command(self, cmd: str) -> None:
         cmd_str = radwag_command_format(cmd)
         cmd_bytes = cmd_str.encode("utf-8")
         send_len = self._connection.write(cmd_bytes)
+        self._raw_log.write(f"{datetime.datetime.now()} --> {cmd_bytes!r}\n")
         assert send_len == len(cmd_bytes), (
             f'Radwag command "{cmd}" ({str(cmd_bytes)} '
             f"bytes) only sent {send_len} bytes"
@@ -117,6 +120,7 @@ def _read_response(
             self._connection.timeout = prev_timeout
         else:
             response = self._connection.readline()
+        self._raw_log.write(f"{datetime.datetime.now()} <-- {response}\n")
         data = radwag_response_parse(response.decode("utf-8"), command)
         return data
 
@@ -150,6 +154,7 @@ def connect(self) -> None:
     def disconnect(self) -> None:
         """Disconnect."""
         self._connection.close()
+        self._raw_log.close()
 
     def read_serial_number(self) -> str:
         """Read serial number."""
@@ -185,6 +190,14 @@ def value_release(self, val_rel: RadwagValueRelease) -> None:
             res.code == RadwagResponseCodes.CARRIED_OUT
         ), f"Unexpected response code: {res.code}"
 
+    def ambiant(self, amb_rel: RadwagAmbiant) -> None:
+        """Set the value release type."""
+        cmd = RadwagCommand.SET_AMBIENT_CONDITIONS_STATE
+        res = self._write_command_and_read_response(cmd, append=str(amb_rel.value))
+        assert (
+            res.code == RadwagResponseCodes.CARRIED_OUT
+        ), f"Unexpected response code: {res.code}"
+
     def continuous_transmission(self, enable: bool) -> None:
         """Enable/disable continuous transmissions."""
         if enable:
@@ -272,6 +285,10 @@ def value_release(self, val_rel: RadwagValueRelease) -> None:
         """Value release."""
         return
 
+    def ambiant(self, amb_rel: RadwagAmbiant) -> None:
+        """Set the value release type."""
+        return
+
     def continuous_transmission(self, enable: bool) -> None:
         """Continuous transmission."""
         return