"""
Implementation of RoboInterface extending the PFonRail to the specific uppsala working cell.
"""
from .movement_profiles import slow_profile, smoothen_profile, straighten_profile
from .pf_on_rail import PFonRail
import time
import logging
from collections.abc import Callable
from functools import partial
from pathlib import Path
from ...control.graph_manager import JointState
from datetime import datetime
[docs]
def _lab_adaption_root() -> Path | None:
"""Root directory of the installed lab_adaption package, or None if it is not installed."""
try:
import lab_adaption
except ImportError:
logging.warning(f"{datetime.now()}: lab_adaption is not installed; using the default graph directory instead.")
return None
return Path(lab_adaption.__file__).resolve().parent
#some custom naming of the positions
ECHO_SOURCE = 0
ECHO_DESTINATION = 1
# robot drives there to have barcode scanned
bc_reader_position = "bc_reader"
lid_pick_offset = 7
lid_place_offset = 10
lid_off = 14
lid_alone_offset = - 3
# (un-)lidding commands are encoded '[un]lid_[DEVICE]_[POSITION]'
LID_ACTION = "lid_"
UNLID_ACTION = "unlid_"
[docs]
def lidding_action(intermediate_actions: list[str], prefix: str) -> str | None:
"""
Picks the (un-)lidding command out of the intermediate actions. There should be at most one.
:param intermediate_actions:
:param prefix: LID_ACTION or UNLID_ACTION
:return: the command or None
"""
return next((cmd for cmd in intermediate_actions if cmd.startswith(prefix)), None)
[docs]
def approach_position(device: str) -> str:
approach_pos = f"{device}_approach"
logging.debug(f"{datetime.now()}: Approach position to device {device} is {approach_pos}.")
return approach_pos
[docs]
def device_from_identifier(identifier: str) -> str:
device_name = identifier.split('_')[0]
logging.debug(f"{datetime.now()}: {identifier} belongs to device {device_name}.")
return device_name
[docs]
class UppsalaPFonRail(PFonRail):
# store/load the position graph in the installed lab_adaption package directory
graph_dir: Path | None = _lab_adaption_root()
# where the lid currently held by the gripper was picked up, if it was fetched in advance
_prefetched_lid_position: str | None = None
GRIPPER_OPEN: float = 90
@property
def name(self) -> str:
# overwrite it because its nicer as server name
return "PFonRail"
[docs]
@classmethod
def get_name(cls) -> str:
return "UppsalaPFonRail"
[docs]
def site_to_position_identifier(self, device: str, slot: int) -> str:
# handle special case of echo
if device == "Echo":
if slot == ECHO_SOURCE:
return "Echo_source"
elif slot == ECHO_DESTINATION:
return "Echo_destination"
# devices with just one handover position are tagged like this
pos_identifier = f"{device}_nest"
# devices with multiple handover positions have their number attached (starting at 0)
if not self.graph_manager.position_known(pos_identifier):
pos_identifier += str(slot)
if not self.graph_manager.position_known(pos_identifier):
logging.error(f"{datetime.now()}: neither {device}_nest nor {pos_identifier} are known positions")
return pos_identifier
[docs]
def move_to_safe_pos(self, identifier: str, offset: dict[str, float] | None = None):
"""
Moves to the approach position belonging to the specified position
:param offset:
:param identifier:
:return:
"""
device_name = device_from_identifier(identifier)
safe_pos = approach_position(device_name)
self.move_to_position(safe_pos, offset=offset)
[docs]
def pick_at_position(self, identifier: str, offset: dict[str, float] | None=None):
# picking opens the gripper before moving in, which would drop a lid held since
# prepare_for_input(), so it goes back to its storage first
if self.holding_lid:
self.return_prefetched_lid()
super().pick_at_position(identifier, offset = offset)
self.move_to_safe_pos(identifier, offset=offset)
[docs]
def place_at_position(self, identifier: str, offset: dict[str, float] | None = None):
offset = offset or {}
super().place_at_position(identifier, offset=offset)
self.move_to_safe_pos(identifier, offset=offset)
[docs]
def prepare_for_output(self, internal_pos: int, device: str, position: int) -> bool:
safe_pos = approach_position(device)
self.move_to_position(safe_pos)
return True
[docs]
def cancel_transfer(self):
# the transfer the lid was fetched for will not happen, so put it back where it came from
if self.holding_lid:
logging.info(f"{datetime.now()}: transfer cancelled, returning the lid")
self.return_prefetched_lid()
super().cancel_transfer()
# InteractiveTransferInterface
@property
def available_intermediate_actions(self) -> list[str]:
return ["read_barcode", "unlid", "lid"]
@property
def preparable_intermediate_actions(self) -> list[str]:
# unlidding and barcode reading need the plate, only the lid can be fetched in advance
return ["lid"]
@property
def holding_lid(self) -> bool:
"""
Whether the gripper holds a lid that was fetched in advance by prefetch_lid(), i.e.
something the arm put there on purpose and still knows about.
:return:
"""
return self._prefetched_lid_position is not None
@property
def currently_gripping_plate(self) -> bool:
# a lid is not a plate, and it is never dropped by a following pick: pick_at_position()
# puts it back first. So a lid held on purpose must not look like labware stuck in the
# gripper, which would make PrepareForInput refuse the transfer it was fetched for.
return not self.holding_lid and super().currently_gripping_plate
[docs]
def lid_site_of(self, lidding_cmd: str) -> str:
"""
Resolves the position of the lid an (un-)lidding command refers to.
:param lidding_cmd: a command encoded '[un]lid_[DEVICE]_[POSITION]'
:return: the position identifier of the lid
"""
# the device name may contain underscores itself, so only the prefix and the trailing
# slot number are split off
_, device_and_position = lidding_cmd.split("_", 1)
lid_device, lid_position = device_and_position.rsplit("_", 1)
return self.site_to_position_identifier(lid_device, int(lid_position))
[docs]
def prefetch_lid(self, lid_position: str):
"""
Picks up a lid before the plate it belongs to is needed, so that put_lid_on() only has to
place it. Failing here fails the preparation, the lid is never silently left behind.
:param lid_position:
:return:
"""
logging.info(f"{datetime.now()}: fetching the lid at {lid_position} in advance")
print(f"{datetime.now()}: fetching lid at {lid_position}")
self.pick_at_position(lid_position, offset={"z": lid_alone_offset})
self._prefetched_lid_position = lid_position
[docs]
def return_prefetched_lid(self):
"""
Puts a lid fetched in advance back where it was taken from. The arm is only considered
free of it once it is actually placed, so the position is cleared afterwards.
:return:
"""
lid_position = self._prefetched_lid_position
logging.info(f"{datetime.now()}: returning the lid held in advance to {lid_position}")
print(f"{datetime.now()}: putting lid to {lid_position}")
self.place_at_position(lid_position, offset={"z": lid_alone_offset})
self._prefetched_lid_position = None
[docs]
def remove_lid(self, plate_position: str, lid_deposit_position: str):
logging.info(f"{datetime.now()}: removing lid at {plate_position} and putting it to {lid_deposit_position}")
print(f"{datetime.now()}: unlidding at {plate_position}")
self.move_to_position(plate_position, offset={"z": lid_pick_offset})
self.grip_close()
self.move_to_position(plate_position, offset={"z": lid_off})
self.move_to_safe_pos(plate_position, offset={"z": lid_off})
self.store_lid(lid_deposit_position)
[docs]
def remove_lid_deferred(self, plate_position: str, lid_deposit_position: str) -> Callable[[], None]:
"""
Takes the lid off the plate and retreats to the safe position with the lid gripped.
Storing the lid away is not executed but returned as a tail task, so that the caller can
hand it to RoboInterface.defer(): the command finishes as soon as the robot is clear of
the device, while the robot stays reserved until the lid is stored.
:param plate_position: where the lidded plate stands
:param lid_deposit_position: where the lid is to be stored
:return: the work to be done once the current command returned
"""
logging.info(f"{datetime.now()}: removing lid at {plate_position}, storing it to {lid_deposit_position} later")
print(f"{datetime.now()}: unlidding at {plate_position}")
self.move_to_position(plate_position, offset={"z": lid_pick_offset})
self.grip_close()
self.move_to_position(plate_position, offset={"z": lid_off})
self.move_to_safe_pos(plate_position, offset={"z": lid_off})
# only reached with the lid actually gripped: a failure above propagates to the caller
# instead of becoming a tail task nobody is waiting for
return partial(self.store_lid, lid_deposit_position)
[docs]
def store_lid(self, lid_deposit_position: str):
"""
Places a lid currently held by the gripper into its storage position.
:param lid_deposit_position:
:return:
"""
print(f"{datetime.now()}: putting lid to {lid_deposit_position}")
self.place_at_position(lid_deposit_position, offset={"z": lid_alone_offset})
[docs]
def put_lid_on(self, plate_position: str, lid_position: str):
logging.info(f"{datetime.now()}: putting lid from {lid_position} on plate at {plate_position}")
if self._prefetched_lid_position == lid_position:
print(f"{datetime.now()}: lidding at {plate_position}")
# already in the gripper from prepare_for_input(), so no pick and no opening here
self._prefetched_lid_position = None
else:
# any lid held for another position is put back by pick_at_position()
print(f"{datetime.now()}: fetching lid at {lid_position}")
self.pick_at_position(lid_position, offset={"z":lid_alone_offset})
print(f"{datetime.now()}: lidding at {plate_position}")
self.move_to_position(plate_position, offset={"z":lid_off})
self.move_to_position(plate_position, offset={"z":lid_place_offset})
self.grip_open()
self.move_to_safe_pos(plate_position, offset={"z":lid_place_offset})
[docs]
def get_barcode_scanned(self):
if self.graph_manager.position_known(bc_reader_position):
self.move_to_position(bc_reader_position)
time.sleep(.5)
[docs]
def put_labware(self, intermediate_actions: list[str], device: str, position: int):
position_identifier = self.site_to_position_identifier(device, position)
self.place_at_position(position_identifier)
unlid_cmd = lidding_action(intermediate_actions, UNLID_ACTION)
if unlid_cmd:
# the command finishes once the robot left the device holding the lid, so the device
# can start working on the plate while the lid is being stored away
self.defer(self.remove_lid_deferred(position_identifier, self.lid_site_of(unlid_cmd)))
[docs]
def get_labware(self, intermediate_actions: list[str], device: str, position: int):
position_identifier = self.site_to_position_identifier(device, position)
lid_cmd = lidding_action(intermediate_actions, LID_ACTION)
if lid_cmd:
# put_lid_on() picks the lid up itself unless prepare_for_input() already did
self.put_lid_on(position_identifier, self.lid_site_of(lid_cmd))
self.pick_at_position(position_identifier)
if "read_barcode" in intermediate_actions:
self.get_barcode_scanned()
[docs]
def move_to_coordinates(self, coords: JointState, **kwargs) -> None:
# go slower if the next or last position is a nest or the current position is unknown
profile = self.move_profile
# if specified, switch to a smooth and/or linear movement
if "corners" in kwargs and kwargs["corners"] == "smooth":
profile = smoothen_profile(profile)
if "mode" in kwargs and kwargs["mode"] == "straight":
profile = straighten_profile(profile)
if self.speed > 10:
keywords = ("nest", "Echo_source", "Echo_destination")
if not self.current_position or any(
keyword in self.next_positions[0] or keyword in self.current_position
for keyword in keywords
):
profile = slow_profile # slow down around nests and echo positions
with self.use_move_profile(profile):
super().move_to_coordinates(coords, **kwargs)
