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

)

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