diff --git a/pylabrobot/hamilton/star/driver/features/pipettes.py b/pylabrobot/hamilton/star/driver/features/pipettes.py index 5498ecfe527..5126fd62013 100644 --- a/pylabrobot/hamilton/star/driver/features/pipettes.py +++ b/pylabrobot/hamilton/star/driver/features/pipettes.py @@ -1226,6 +1226,42 @@ async def _temporary_y_drive_profile( async with self._temporary_drive_profile({"yv": speed, "yr": acceleration_level}, channels): yield + async def _set_y_drive_profiles( + self, channels: List[int], speed: float, acceleration_level: int + ) -> None: + """Give channels one Y speed and acceleration level, writing only where a drive differs. + + `C0 JY` moves each channel at what its drive holds, and a channel command (`Px YS`, `Px YL`) + leaves its own values there; neighbours at different speeds can run into each other. + + Args: + channels: which channels, 0-indexed from the back. + speed: in mm/s. + acceleration_level: 1 (gentlest) to 4. + """ + # A channel module answers one command at a time: every round sends each at most one. + held_speeds = await asyncio.gather( + *(self.request_y_speed(channel=channel) for channel in channels) + ) + held_levels = await asyncio.gather( + *(self.request_y_acceleration_level(channel=channel) for channel in channels) + ) + target = self._drive_parameter_to_increments(parameter="yv", value=speed) + await asyncio.gather( + *( + self._set_y_speed(channel=channel, speed=speed) + for channel, held in zip(channels, held_speeds) + if self._drive_parameter_to_increments(parameter="yv", value=held) != target + ) + ) + await asyncio.gather( + *( + self._set_y_acceleration_level(channel=channel, level=acceleration_level) + for channel, held in zip(channels, held_levels) + if held != acceleration_level + ) + ) + # ---- z ----------------------------------------------------------------------------------------- async def request_z_speed(self, channel: int) -> float: @@ -1291,40 +1327,6 @@ async def _temporary_z_drive_profile( async with self._temporary_drive_profile({"zv": speed, "zr": acceleration}, channels): yield - async def _set_y_drive_profiles( - self, channels: List[int], speed: float, acceleration_level: int - ) -> None: - """Give channels one Y speed and acceleration level, writing only where a drive differs. - - `C0 JY` moves each channel at what its drive holds, and a channel command (`Px YS`, `Px YL`) - leaves its own values there; neighbours at different speeds can run into each other. - - Args: - channels: which channels, 0-indexed from the back. - speed: in mm/s. - acceleration_level: 1 (gentlest) to 4. - """ - # A channel module answers one command at a time: every round sends each at most one. - held_speeds = await asyncio.gather(*(self.request_y_speed(channel) for channel in channels)) - held_levels = await asyncio.gather( - *(self.request_y_acceleration_level(channel) for channel in channels) - ) - target = self._drive_parameter_to_increments("yv", speed) - await asyncio.gather( - *( - self._set_y_speed(channel, speed) - for channel, held in zip(channels, held_speeds) - if self._drive_parameter_to_increments("yv", held) != target - ) - ) - await asyncio.gather( - *( - self._set_y_acceleration_level(channel, acceleration_level) - for channel, held in zip(channels, held_levels) - if held != acceleration_level - ) - ) - async def _set_default_drive_parameters(self) -> None: """Give every channel the driver's Y and Z defaults (`Px AA`), whatever a session left. @@ -1332,7 +1334,7 @@ async def _set_default_drive_parameters(self) -> None: what they hold is never taken as a default; Y is written only where it differs. """ await self._set_y_drive_profiles( - list(range(self.num_channels)), + channels=list(range(self.num_channels)), speed=self.default_y_speed, acceleration_level=self.default_y_acceleration_level, ) @@ -1487,7 +1489,9 @@ async def _move_to_planned_y_positions( speed = self.default_y_speed if speed is None else speed if acceleration_level is None: acceleration_level = self.default_y_acceleration_level - await self._set_y_drive_profiles(channels, speed=speed, acceleration_level=acceleration_level) + await self._set_y_drive_profiles( + channels=channels, speed=speed, acceleration_level=acceleration_level + ) yp = " ".join([f"{round(y * 10):04}" for y in channel_locations.values()]) try: resp = await self._driver.send_command( @@ -1502,14 +1506,17 @@ async def _move_to_planned_y_positions( # A speed given for this move only leaves the defaults in the drives, as every other move; # one round per value, so no channel module is sent two commands at once. rounds = [] - if self._drive_parameter_to_increments("yv", speed) != self._drive_parameter_to_increments( - "yv", self.default_y_speed - ): - rounds.append([self._set_y_speed(channel, self.default_y_speed) for channel in channels]) + default_speed = self._drive_parameter_to_increments( + parameter="yv", value=self.default_y_speed + ) + if self._drive_parameter_to_increments(parameter="yv", value=speed) != default_speed: + rounds.append( + [self._set_y_speed(channel=channel, speed=self.default_y_speed) for channel in channels] + ) if acceleration_level != self.default_y_acceleration_level: rounds.append( [ - self._set_y_acceleration_level(channel, self.default_y_acceleration_level) + self._set_y_acceleration_level(channel=channel, level=self.default_y_acceleration_level) for channel in channels ] ) @@ -2391,8 +2398,7 @@ async def _unchecked_fw_probe_x_using_clld( """Move the arm along X until a channel's cLLD triggers, or to the end, as given. `C0 XL`. Args: - channel: the channel whose cLLD stops the search, 0-indexed from the back (`pn`). Left out, - the firmware watches the last channel a command named, or the first. + channel: the channel whose cLLD stops the search, 0-indexed from the back (`pn`). end_position: where the search ends, in mm, sent in 0.1 mm (`xs`). read_timeout: how long to wait for the answer, in s. """ @@ -2682,7 +2688,7 @@ async def probe_y_using_clld( try: try: await self._unchecked_fw_probe_y_using_clld( - channel, + channel=channel, end_position=end_increments, detection_edge=detection_edge, search_speed=speed_increments, @@ -2693,15 +2699,15 @@ async def probe_y_using_clld( if not self._found_nothing(error, self.channel_id(channel)): raise found = False - detected = await self.request_y_position(channel) + detected = await self.request_y_position(channel=channel) read = True finally: # A search that failed, or whose read failed, left the channel where nothing records it. if not read: - await self._record_where_they_stopped("y", [channel]) + await self._record_where_they_stopped(axis="y", channels=[channel]) # The post-detection move, along the path just searched, moving this channel alone. - distance = c.y_drive_mm_to_increments(min(post_detection_distance, abs(detected - here))) + distance = c.y_drive_mm_to_increments(mm=min(post_detection_distance, abs(detected - here))) if distance > 0: try: await self._unchecked_fw_move_y_relative( @@ -2713,15 +2719,17 @@ async def probe_y_using_clld( ) finally: # Whether the move arrived or stopped part way, this read is how either is recorded. - await self._record_where_they_stopped("y", [channel]) + await self._record_where_they_stopped(axis="y", channels=[channel]) finally: # The search and the post-detection move leave their speed in the drive: put the defaults # back, so the probe leaves nothing behind. try: - if speed_increments != c.y_drive_mm_to_increments(self.default_y_speed): - await self._set_y_speed(channel, self.default_y_speed) + if speed_increments != c.y_drive_mm_to_increments(mm=self.default_y_speed): + await self._set_y_speed(channel=channel, speed=self.default_y_speed) if acceleration_level != self.default_y_acceleration_level: - await self._set_y_acceleration_level(channel, self.default_y_acceleration_level) + await self._set_y_acceleration_level( + channel=channel, level=self.default_y_acceleration_level + ) except Exception: logger.warning("could not put channel %s's Y profile back to the default", channel) surface = detected + diameter / 2 if direction == "backward" else detected - diameter / 2 diff --git a/pylabrobot/hamilton/star/driver/features/pipettes_tests.py b/pylabrobot/hamilton/star/driver/features/pipettes_tests.py index 86a46f3889e..e3d7cb9df0a 100644 --- a/pylabrobot/hamilton/star/driver/features/pipettes_tests.py +++ b/pylabrobot/hamilton/star/driver/features/pipettes_tests.py @@ -421,7 +421,7 @@ async def recorded(module: str, command: str, **kwargs: Any): async def test_setup_does_not_take_what_every_channel_holds_as_the_default(self): pipettes = await simulated_channels() for channel in range(pipettes.num_channels): - await pipettes._set_y_speed(channel, 350.0) + await pipettes._set_y_speed(channel=channel, speed=350.0) await pipettes._driver.setup() for channel in range(pipettes.num_channels): self.assertEqual(await pipettes.request_y_speed(channel), 274.99) @@ -450,19 +450,19 @@ async def recorded(module: str, command: str, **kwargs: Any): async def test_a_y_move_first_gives_a_channel_left_at_another_speed_the_default(self): pipettes = await simulated_channels() - await pipettes._set_y_speed(7, 10.0) - await pipettes._set_y_acceleration_level(7, 1) + await pipettes._set_y_speed(channel=7, speed=10.0) + await pipettes._set_y_acceleration_level(channel=7, level=1) sent = await self.recorded_y_moves(pipettes) ys = await pipettes.request_y_positions() - await pipettes.move_to_y_positions({7: ys[7] - 5.0}) + await pipettes.move_to_y_positions(ys={7: ys[7] - 5.0}) self.assertEqual(sent[:2], ["P8AAyv5939", "P8AAyr4"]) self.assertTrue(sent[2].startswith("C0JY")) self.assertEqual(len(sent), 3) async def test_a_y_move_never_sends_one_channel_two_commands_at_once(self): pipettes = await simulated_channels() - await pipettes._set_y_speed(7, 10.0) - await pipettes._set_y_acceleration_level(7, 1) + await pipettes._set_y_speed(channel=7, speed=10.0) + await pipettes._set_y_acceleration_level(channel=7, level=1) answer = pipettes._driver.send_command in_flight: Set[str] = set() overlaps: List[str] = [] @@ -479,14 +479,14 @@ async def one_at_a_time(module: str, command: str, **kwargs: Any): pipettes._driver.send_command = one_at_a_time # type: ignore[assignment] ys = await pipettes.request_y_positions() - await pipettes.move_to_y_position(7, ys[7] - 5.0, speed=100.0, acceleration_level=1) + await pipettes.move_to_y_position(channel=7, y=ys[7] - 5.0, speed=100.0, acceleration_level=1) self.assertEqual(overlaps, []) async def test_a_y_move_given_a_speed_holds_it_for_that_move_only(self): pipettes = await simulated_channels() sent = await self.recorded_y_moves(pipettes) ys = await pipettes.request_y_positions() - await pipettes.move_to_y_position(7, ys[7] - 5.0, speed=100.0, acceleration_level=1) + await pipettes.move_to_y_position(channel=7, y=ys[7] - 5.0, speed=100.0, acceleration_level=1) move = next(index for index, command in enumerate(sent) if command.startswith("C0JY")) channels = range(1, pipettes.num_channels + 1) given = {f"P{n}AAyv2160" for n in channels} | {f"P{n}AAyr1" for n in channels} @@ -511,7 +511,7 @@ async def failing(module: str, command: str, **kwargs: Any): pipettes._driver.send_command = failing # type: ignore[assignment] ys = await pipettes.request_y_positions() with self.assertRaises(RuntimeError): - await pipettes.move_to_y_position(7, ys[7] - 5.0, speed=100.0) + await pipettes.move_to_y_position(channel=7, y=ys[7] - 5.0, speed=100.0) move = next(index for index, command in enumerate(sent) if command.startswith("C0JY")) self.assertEqual( set(sent[move + 1 :]), {f"P{n}AAyv5939" for n in range(1, pipettes.num_channels + 1)} @@ -698,7 +698,7 @@ async def test_x_probe_searches_backs_away_and_corrects_for_the_tip(self): async def test_x_probe_with_no_post_detection_distance_sends_only_the_search(self): self._stand_at_x(300.0, 250.04) x = await self.pipettes.probe_x_using_clld( - 0, "left", search_end_position=200.0, post_detection_distance=0 + channel=0, direction="left", search_end_position=200.0, post_detection_distance=0 ) self.assertEqual(self.sent, ["C0XLpn01xs02000"]) self.moves.assert_not_awaited() @@ -754,7 +754,9 @@ async def test_y_probe_with_no_post_detection_distance_sends_only_the_search(sel after = list(self.ys) after[1] = 250.0 self._stand_at_y(self.ys, self.ys, after) - y = await self.pipettes.probe_y_using_clld(1, "forward", post_detection_distance=0) + y = await self.pipettes.probe_y_using_clld( + channel=1, direction="forward", post_detection_distance=0 + ) self.assertEqual([command[2:4] for command in self.sent], ["YL", "AA"]) self.moves.assert_not_awaited() self.assertEqual(y, 249.4) @@ -796,7 +798,7 @@ async def test_x_probe_that_finds_nothing_backs_away_and_answers_none(self): self.assertIsNone(await self.pipettes.probe_x_using_clld(0, "left", search_end_position=200.0)) self.moves.assert_awaited_once_with(202.0) - async def test_y_probe_that_finds_nothing_backs_away_and_answers_none(self): + async def test_y_probe_that_finds_nothing_makes_its_post_detection_move_and_answers_none(self): for trace in (70, 73): self.moves.reset_mock() self._answer_the_search_with("YL", f"P2YLid0001er{trace}") diff --git a/pylabrobot/hamilton/star/driver/simulator.py b/pylabrobot/hamilton/star/driver/simulator.py index bd25b2ea7b2..6b959398d6b 100644 --- a/pylabrobot/hamilton/star/driver/simulator.py +++ b/pylabrobot/hamilton/star/driver/simulator.py @@ -289,9 +289,9 @@ def update_location_by_reference_point( 0.0 if y is None else y - self._modelled_y(channel), # the speed the channel's simulated drive holds, as `C0 JY` moves each at its own self.configuration.y_drive_increments_to_mm( - self.device.channel_drive_parameters.get(channel, SIMULATED_CHANNEL_DRIVE_PARAMETERS)[ - "yv" - ] + increments=self.device.channel_drive_parameters.get( + channel, SIMULATED_CHANNEL_DRIVE_PARAMETERS + )["yv"] ), None, ),