feat(hardware): hook up a way to toggle multi-sensor logic (#15557)

<!--
Thanks for taking the time to open a pull request! Please make sure
you've read the "Opening Pull Requests" section of our Contributing
Guide:


https://github.com/Opentrons/opentrons/blob/edge/CONTRIBUTING.md#opening-pull-requests

To ensure your code is reviewed quickly and thoroughly, please fill out
the sections below to the best of your ability!
-->

# Overview

<!--
Use this section to describe your pull-request at a high level. If the
PR addresses any open issues, please tag the issues here.
-->
Hardware testing wants a way to toggle different logic options for the
8/96 channel. This exposes' it in ot3 api so hardware-testing can use it
if they want
# Test Plan

<!--
Use this section to describe the steps that you took to test your Pull
Request.
If you did not perform any testing provide justification why.

OT-3 Developers: You should default to testing on actual physical
hardware.
Once again, if you did not perform testing against hardware, justify
why.

Note: It can be helpful to write a test plan before doing development

Example Test Plan (HTTP API Change)

- Verified that new optional argument `dance-party` causes the robot to
flash its lights, move the pipettes,
then home.
- Verified that when you omit the `dance-party` option the robot homes
normally
- Added protocol that uses `dance-party` argument to G-Code Testing
Suite
- Ran protocol that did not use `dance-party` argument and everything
was successful
- Added unit tests to validate that changes to pydantic model are
correct

-->

# Changelog

<!--
List out the changes to the code in this PR. Please try your best to
categorize your changes and describe what has changed and why.

Example changelog:
- Fixed app crash when trying to calibrate an illegal pipette
- Added state to API to track pipette usage
- Updated API docs to mention only two pipettes are supported

IMPORTANT: MAKE SURE ANY BREAKING CHANGES ARE PROPERLY COMMUNICATED
-->

# Review requests

<!--
Describe any requests for your reviewers here.
-->

# Risk assessment

<!--
Carefully go over your pull request and look at the other parts of the
codebase it may affect. Look for the possibility, even if you think it's
small, that your change may affect some other part of the system - for
instance, changing return tip behavior in protocol may also change the
behavior of labware calibration.

Identify the other parts of the system your codebase may affect, so that
in addition to your own review and testing, other people who may not
have the system internalized as much as you can focus their attention
and testing there.
-->

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: ryanthecoder <[email protected]>

analyses-snapshot-testing/tests/__snapshots__/analyses_snapshot_test/test_analysis_snapshot[e42e36e3ca][OT2_X_v2_13_None_None_PythonSyntaxError].json

@@ -28,7 +28,7 @@
  "msg": "No module named 'superspecialmagic'",
  "name": "superspecialmagic",
  "path": "None",
- "traceback": " File \"/usr/local/lib/python3.10/site-packages/opentrons/cli/analyze.py\", line N, in _do_analyze\n await runner.load(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/protocol_runner.py\", line N, in load\n self._protocol_executor.extract_run_parameters(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/python_protocol_wrappers.py\", line N, in extract_run_parameters\n return exec_add_parameters(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_add_parameters\n exec(protocol.contents, new_globs)\n\n File \"OT2_X_v2_13_None_None_PythonSyntaxError.py\", line N, in <module>\n"
+ "traceback": " File \"/usr/local/lib/python3.10/site-packages/opentrons/cli/analyze.py\", line N, in _do_analyze\n await orchestrator.load(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/run_orchestrator.py\", line N, in load\n await self._protocol_runner.load(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/protocol_runner.py\", line N, in load\n self._protocol_executor.extract_run_parameters(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocol_runner/python_protocol_wrappers.py\", line N, in extract_run_parameters\n return exec_add_parameters(\n\n File \"/usr/local/lib/python3.10/site-packages/opentrons/protocols/execution/execute_python.py\", line N, in exec_add_parameters\n exec(protocol.contents, new_globs)\n\n File \"OT2_X_v2_13_None_None_PythonSyntaxError.py\", line N, in <module>\n"
  },
  "errorType": "PythonException",
  "id": "UUID",

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

@@ -149,6 +149,7 @@ async def liquid_probe(
  output_format: OutputOptions = OutputOptions.can_bus_only,
  data_files: Optional[Dict[InstrumentProbeType, str]] = None,
  probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+ force_both_sensors: bool = False,
  ) -> float:
  ...
 

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

@@ -1361,6 +1361,7 @@ async def liquid_probe(
  output_option: OutputOptions = OutputOptions.can_bus_only,
  data_files: Optional[Dict[InstrumentProbeType, str]] = None,
  probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+ force_both_sensors: bool = False,
  ) -> float:
  if output_option == OutputOptions.sync_buffer_to_csv:
  if (
@@ -1403,6 +1404,7 @@ async def liquid_probe(
  can_bus_only_output=can_bus_only_output,
  data_files=data_files_transposed,
  sensor_id=sensor_id_for_instrument(probe),
+ force_both_sensors=force_both_sensors,
  )
  for node, point in positions.items():
  self._position.update({node: point.motor_position})

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

@@ -348,6 +348,7 @@ async def liquid_probe(
  output_format: OutputOptions = OutputOptions.can_bus_only,
  data_files: Optional[Dict[InstrumentProbeType, str]] = None,
  probe: InstrumentProbeType = InstrumentProbeType.PRIMARY,
+ force_both_sensors: bool = False,
  ) -> float:
  z_axis = Axis.by_mount(mount)
  pos = self._position

api/src/opentrons/hardware_control/ot3api.py

@@ -2574,6 +2574,7 @@ async def _liquid_probe_pass(
  probe_settings: LiquidProbeSettings,
  probe: InstrumentProbeType,
  z_distance: float,
+ force_both_sensors: bool = False,
  ) -> float:
  plunger_direction = -1 if probe_settings.aspirate_while_sensing else 1
  await self._backend.liquid_probe(
@@ -2585,6 +2586,7 @@ async def _liquid_probe_pass(
  probe_settings.output_option,
  probe_settings.data_files,
  probe=probe,
+ force_both_sensors=force_both_sensors,
  )
  end_pos = await self.gantry_position(mount, refresh=True)
  return end_pos.z
@@ -2607,6 +2609,7 @@ async def liquid_probe(
  max_z_dist: float,
  probe_settings: Optional[LiquidProbeSettings] = None,
  probe: Optional[InstrumentProbeType] = None,
+ force_both_sensors: bool = False,
  ) -> float:
  """Search for and return liquid level height.
 

api/tests/opentrons/hardware_control/test_ot3_api.py

@@ -843,6 +843,7 @@ async def test_liquid_probe(
  fake_settings_aspirate.output_option,
  fake_settings_aspirate.data_files,
  probe=InstrumentProbeType.PRIMARY,
+ force_both_sensors=False,
  )
 
  return_dict[head_node], return_dict[pipette_node] = 142, 142
@@ -913,6 +914,7 @@ async def test_multi_liquid_probe(
  fake_settings_aspirate.output_option,
  fake_settings_aspirate.data_files,
  probe=InstrumentProbeType.PRIMARY,
+ force_both_sensors=False,
  )
  assert mock_liquid_probe.call_count == 3
 
@@ -982,6 +984,7 @@ async def test_liquid_not_found(
  fake_settings_aspirate.output_option,
  fake_settings_aspirate.data_files,
  probe=InstrumentProbeType.PRIMARY,
+ force_both_sensors=False,
  )
  assert mock_liquid_probe.call_count == 3
 

hardware/opentrons_hardware/firmware_bindings/constants.py

@@ -375,6 +375,7 @@ class SensorOutputBinding(int, Enum):
  report = 0x02
  max_threshold_sync = 0x04
  auto_baseline_report = 0x08
+ multi_sensor_sync = 0x10
 
 
 @unique

hardware/opentrons_hardware/firmware_bindings/messages/payloads.py

@@ -495,6 +495,7 @@ class AddSensorLinearMoveBasePayload(AddLinearMoveRequestPayload):
 
  sensor_id: SensorIdField
  sensor_type: SensorTypeField
+ sensor_binding_flags: utils.UInt8Field
 
 
 @dataclass(eq=False)

hardware/opentrons_hardware/hardware_control/motion.py

@@ -56,6 +56,7 @@ class MoveGroupSingleAxisStep:
  move_type: MoveType = MoveType.linear
  sensor_type: Optional[SensorType] = None
  sensor_id: Optional[SensorId] = None
+ sensor_binding_flags: Optional[int] = None
 
  def is_moving_step(self) -> bool:
  """Check if this step involves any actual movement."""
@@ -137,6 +138,7 @@ def create_step(
  stop_condition: MoveStopCondition = MoveStopCondition.none,
  sensor_type_pass: Optional[SensorType] = None,
  sensor_id_pass: Optional[SensorId] = None,
+ sensor_binding_flags: Optional[int] = None,
 ) -> MoveGroupStep:
  """Create a move from a block.
 
@@ -165,6 +167,7 @@ def create_step(
  move_type=MoveType.get_move_type(stop_condition),
  sensor_type=sensor_type_pass,
  sensor_id=sensor_id_pass,
+ sensor_binding_flags=sensor_binding_flags,
  )
  return step
 

hardware/opentrons_hardware/hardware_control/move_group_runner.py

@@ -310,6 +310,7 @@ def _get_stepper_motor_message(
  # stop_condition = step.stop_condition.value
  assert step.sensor_type is not None
  assert step.sensor_id is not None
+ assert step.sensor_binding_flags is not None
  stop_condition = MoveStopCondition.sync_line
  sensor_move_payload = AddSensorLinearMoveBasePayload(
  request_stop_condition=MoveStopConditionField(stop_condition),
@@ -332,6 +333,7 @@ def _get_stepper_motor_message(
  ),
  sensor_type=SensorTypeField(step.sensor_type),
  sensor_id=SensorIdField(step.sensor_id),
+ sensor_binding_flags=UInt8Field(step.sensor_binding_flags),
  )
  return AddSensorLinearMoveRequest(payload=sensor_move_payload)
  else:

hardware/opentrons_hardware/hardware_control/tool_sensors.py

@@ -94,7 +94,14 @@ def _fix_pass_step_for_buffer(
  sensor_type: SensorType,
  sensor_id: SensorId,
  stop_condition: MoveStopCondition = MoveStopCondition.sync_line,
+ binding_flags: Optional[int] = None,
 ) -> MoveGroupStep:
+ if binding_flags is None:
+ binding_flags = (
+ SensorOutputBinding.auto_baseline_report
+ + SensorOutputBinding.sync
+ + SensorOutputBinding.report
+ )
  tool_nodes = [
  i
  for i in movers
@@ -114,6 +121,7 @@ def _fix_pass_step_for_buffer(
  stop_condition=MoveStopCondition.sensor_report,
  sensor_type_pass=sensor_type,
  sensor_id_pass=sensor_id,
+ sensor_binding_flags=binding_flags,
  )
  elif sensor_type == SensorType.capacitive:
  tool_move = create_step(
@@ -127,6 +135,7 @@ def _fix_pass_step_for_buffer(
  stop_condition=MoveStopCondition.sensor_report,
  sensor_type_pass=sensor_type,
  sensor_id_pass=sensor_id,
+ sensor_binding_flags=binding_flags,
  )
  for node in tool_nodes:
  move_group[node] = tool_move[node]
@@ -140,6 +149,7 @@ def _build_pass_step(
  sensor_type: SensorType,
  sensor_id: SensorId,
  stop_condition: MoveStopCondition = MoveStopCondition.sync_line,
+ binding_flags: Optional[int] = None,
 ) -> MoveGroupStep:
  move_group = create_step(
  distance={ax: float64(abs(distance[ax])) for ax in movers},
@@ -154,6 +164,7 @@ def _build_pass_step(
  stop_condition=stop_condition,
  sensor_type_pass=sensor_type,
  sensor_id_pass=sensor_id,
+ sensor_binding_flags=binding_flags,
  )
  return move_group
 
@@ -388,13 +399,24 @@ async def liquid_probe(
  can_bus_only_output: bool = False,
  data_files: Optional[Dict[SensorId, str]] = None,
  sensor_id: SensorId = SensorId.S0,
+ force_both_sensors: bool = False,
 ) -> Dict[NodeId, MotorPositionStatus]:
  """Move the mount and pipette simultaneously while reading from the pressure sensor."""
  log_files: Dict[SensorId, str] = {} if not data_files else data_files
  sensor_driver = SensorDriver()
  threshold_fixed_point = threshold_pascals * sensor_fixed_point_conversion
  # How many samples to take to level out the sensor
  num_baseline_reads = 20
+ sensor_binding = None
+ if sensor_id == SensorId.BOTH and force_both_sensors:
+ # this covers the case when we want to use both sensors in an AND configuration
+ # we don't think we'll use this but we want the ability to override the standard OR configuration
+ sensor_binding = (
+ SensorOutputBinding.auto_baseline_report
+ + SensorOutputBinding.sync
+ + SensorOutputBinding.report
+ + SensorOutputBinding.multi_sensor_sync
+ )
  pressure_sensors = await _setup_pressure_sensors(
  messenger,
  sensor_id,
@@ -412,6 +434,7 @@ async def liquid_probe(
  sensor_type=SensorType.pressure,
  sensor_id=sensor_id,
  stop_condition=MoveStopCondition.sync_line,
+ binding_flags=sensor_binding,
  )
  if sync_buffer_output:
  sensor_group = _fix_pass_step_for_buffer(
@@ -422,6 +445,7 @@ async def liquid_probe(
  sensor_type=SensorType.pressure,
  sensor_id=sensor_id,
  stop_condition=MoveStopCondition.sync_line,
+ binding_flags=sensor_binding,
  )
 
  raise_z_axis = create_step(