Skip to content

v1 STAR pipettes: tidy the Y profile and probe code #1613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BioCam merged 1 commit into from
Nov 11, 2026
Merged

v1 STAR pipettes: tidy the Y profile and probe code #1613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 59 additions & 51 deletions pylabrobot/hamilton/star/driver/features/pipettes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -1291,48 +1327,14 @@ 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.

The drives keep what was last sent until a power cycle, by this or any other program, so
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,
)
Expand Down Expand Up @@ -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(
Expand All @@ -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
]
)
Expand Down Expand Up @@ -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.
"""
Expand Down Expand Up @@ -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,
Expand All @@ -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(
Expand All @@ -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
Expand Down
26 changes: 14 additions & 12 deletions pylabrobot/hamilton/star/driver/features/pipettes_tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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] = []
Expand All @@ -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}
Expand All @@ -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)}
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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}")
Expand Down
6 changes: 3 additions & 3 deletions pylabrobot/hamilton/star/driver/simulator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
),
Expand Down
Loading