diff --git a/src/aare/devices/tell_client.py b/src/aare/devices/tell_client.py index 87d21a15..2883c4f2 100755 --- a/src/aare/devices/tell_client.py +++ b/src/aare/devices/tell_client.py @@ -10,10 +10,8 @@ import requests from aare.common.logger_config import setup_logger from aare.common.models import ( PuckLoadedInfo, - SampleShortInfo, DewarAddress, SampleDewarAddress, - #PuckInfo, ) from aareDB import PuckWithTellPosition @@ -48,6 +46,10 @@ class TellConnectionException(Exception): VALID_DEWAR_POSITIONS = [f"{p}{n}" for n in "12345" for p in "ABCDEFX"] +def is_valid_dewar_position(position): + """check if argument is a valid dewar position""" + return position in VALID_DEWAR_POSITIONS + POSITION_PARK = "pPark" POSITION_COLD = "pCold" POSITION_AUX = "pAux" @@ -58,39 +60,38 @@ POSITION_HEATER = "pHeatB" #Nov 26 13:36:00 mx-x06da-queue-01.psi.ch AareDAQ[2944444]: 2025-11-26 13:36:00,388 - aareDAQ - ERROR - Error getting status: ('Connection aborted.', ConnectionResetError(104, 'Connection reset by peer')) class TellClient: + """High-level Tell robot API using PShellClient""" def __init__(self, bl: MXBeamline): self.__url = None beamline = bl.value.lower() - if bl == MXBeamline.SIMULATED: - self.__simulation = True - elif bl == MXBeamline.X06DA: - self.__simulation = False + self.__beamline = bl + if bl == MXBeamline.X06DA: self.__url = f"http://{beamline}-tell.psi.ch:22222" + elif bl == MXBeamline.X10SA: - self.__simulation = False self.__url = f"http://PC17488:22222" - if self.__simulation: - print("TELL P-Shell in SIMULATION mode") + elif bl == MXBeamline.X06SA: + self.__url = f"" + raise NotImplemented(f"TellClient not implemente for {beamline}") + elif bl == MXBeamline.SIMULATED: + raise NotImplemented(f"Use SimClient, generate tell client using" + f"make_tell_client(beamline)") else: - print(f"Connecting TELL p-shell service at {self.__url} ...", end="") - hostname = urlparse(self.__url).hostname - try: - requests.get(f"{self.__url}/history/0", timeout=1.0) - except ConnectionError: - print(f"...connection to {hostname} failed") - raise - except requests.ReadTimeout: - print(f"...PShell service {hostname} is down") - raise - self.pshell = PShellClient(self.__url) - self._simulated_samples_info = {} - self._simulated_detected_pucks = [] - self._simulated_mounted_sample = "" - self._simulated_current = 30.0 - self._simulated_suppress = True - self._simulated_state = "Ready" - self._simulated_offset = 0.0 + raise ValueError(f"Unknown beamline {beamline}") + + print(f"Connecting TELL p-shell service at {self.__url} ...", end="") + hostname = urlparse(self.__url).hostname + try: + requests.get(f"{self.__url}/history/0", timeout=1.0) + except ConnectionError: + print(f"...connection to {hostname} failed") + raise + except requests.ReadTimeout: + print(f"...PShell service {hostname} is down") + raise + self.pshell = PShellClient(self.__url) + self._aborted = False self.state = self.get_state() self.debug = False @@ -98,129 +99,44 @@ class TellClient: @property def url(self): + """returns the configured base url for the Tell robot""" return self.__url - def smc_get_current(self) -> float: - if self.__simulation: - return 30.0 - return json.loads(self.pshell.eval("smart_magnet.get_current()&")) - - def smc_set_current(self, current: float): - if self.__simulation: - return - self.pshell.eval(f"smart_magnet.set_current({current:.1f})&") - - def smc_get_suppress(self) -> bool: - if self.__simulation: - return self._simulated_suppress - return json.loads(self.pshell.eval("smart_magnet.get_supress()&")) - - def smc_set_suppress(self, state: bool): - if self.__simulation: - self._simulated_suppress = state - return - self.pshell.eval(f"smart_magnet.set_supress({state})&") - - def check_smc(self): - self.pshell.eval("smart_magnet.set_supress(False)&") - self.pshell.eval("smart_magnet.set_resting_current()&") - self.pshell.eval("smart_magnet.check_mounted(idle_time=1.0, timeout=1.0") - - def magnet_blower(self, state): - """turn on/off blower to the magnet for de-icying purposes - - state = false (blower is off) - state = true (blower is on) - """ - if X06DA: - self.pshell.eval(f"set_pin_cleaner({state})&") - else: - cmd = "true" if state else "false" - self.pshell.eval(f'robot.evaluate("doFOut1={cmd}")&') - def get_state(self): - if self.__simulation: - return self._simulated_state + """returns the current state of the robot""" self.state = self.pshell.get_state() return self.state def get_result(self, command_id=-1): - if self.__simulation: - return { - "id": self._last_cmd_id, - "status": "completed", - "exception": "", - "return": (True, "PINCODE6546"), - } + """returns the result of the last command issued to the robot""" return self.pshell.get_result(command_id) - def wait_ready(self): - if self.__simulation: - logger.info("simulated wait_ready") - time.sleep(3.0) - self._simulated_state = "Ready" - return - # Monitors event but polls every second just n case an event is missed - logger.debug(f"Waiting for robot to be ready. Current state: {self.state} Get state: {self.get_state()}") - while True: - if self.state != "Busy": - logger.debug(f"Robot is now ready...? {self.state} Get:{self.get_state()}") - break - time.sleep(0.2) - self.get_state() - if self.state != "Ready": - if self.state == "Initializing": - raise Exception("Tell reconnecting") - elif self.state == "Closing": - raise Exception("Tell is disconnecting") - raise Exception("Invalid state: " + str(self.state)) + def wait_ready(self, timeout: float = 360.0): + """waits until the robot is ready to accept commands returns None if simulation + and raises an exception if the robot is not ready""" + self.pshell.wait_state("Ready", timeout=timeout) + + def wait_not_busy(self, timeout: float = 360.0): + """waits until the robot is not busy and returns None if simulation + and raises an exception if the robot is busy""" + self.pshell.wait_state_not("Busy", timeout=timeout) + state = self.get_state() + if state != "Ready": + if state == "Initializing": + raise TellConnectionException("Tell reconnecting") + elif state == "Closing": + raise TellConnectionException("Tell is disconnecting") + raise Exception("Invalid state: " + str(state)) def set_in_mount_position(self, value): - if self.__simulation: - return + """tells the robot that the beamlien is safe and to set the in mount position flag allowing mounting + :param value """ self.pshell.eval("in_mount_position = " + str(value) + "&") - def is_in_mount_position(self): - if self.__simulation: - return True + def is_in_mount_position(self) -> bool: + """checks to see if the robot is in the mount position and returns a boolean""" return self.pshell.eval("in_mount_position&").lower() == "true" - def set_simulated_mounted_sample(self, info): - self._simulated_mounted_sample = info - - def set_simulated_samples_info(self, samples): - self._simulated_samples_info = samples - - def set_simulated_detected_pucks(self, pucks): - self._simulated_detected_pucks = pucks - - def get_samples_info(self) -> List[SampleShortInfo]: - if self.__simulation: - j = self._simulated_samples_info # FIXME - else: - j = json.loads(self.pshell.eval("get_samples_info()&")) - - output: List[SampleShortInfo] = [] - for i in j: - if len(i["puckAddress"]) == 2: - dewar_location = DewarAddress( - segment=i["puckAddress"][0], pos=i["puckAddress"][1] - ) - else: - dewar_location = None - - output.append( - SampleShortInfo( - puck_name=i["puckBarcode"], - dewar_name=i["dewarName"], - sample_name=i["sampleName"], - pin=i["samplePosition"], - user=i["userName"], - location=dewar_location, - ) - ) - return output - def set_samples_info(self, info: List[PuckWithTellPosition]): """sets the samples in the robot dewar based on the given list of PuckWithTellPosition objects and runs set_sample_info in the background""" @@ -256,33 +172,6 @@ class TellClient: self.get_state() return ret - def wait_cmd(self, cmd): - if self.__simulation: - return {True, "BARCODE_BIGUS"} - self.wait_ready() - result = self.get_result(cmd) - # print (result) - if result["exception"] is not None: - raise Exception(result["exception"]) - return result["return"] - - def is_cmd_completed(self, cmd): - if self.__simulation: - return True - return self.get_result(cmd)["status"] != "running" - - def wait_mount_complete(self, timeout: float = 360): - if self.__simulation: - time.sleep(1.0) - return - logger.debug(f"Waiting for mount to complete. Is busy? {self.is_busy()}") - timeisup = timeout + time.time() - while time.time() < timeisup: - if not self.is_busy(): - logger.debug(f"Finished waiting for mount to complete. Is busy? {self.is_busy()}") - break - time.sleep(0.2) - def check_command_ok(self, timeout: float = 360.0, msg: str = ""): """checks to see if the last command issued to the robot was completed and returns the result Returns an exception if the command result doesnt return completed or removed""" @@ -290,13 +179,16 @@ class TellClient: result = self.get_result(self._last_cmd_id) logger.debug(f"{msg} {result}") status = result["status"] - if "completed" != status: + if "completed" != status: #FIXME this is very limiting and depends on tell reporting statuses if "removed" != status: raise TellMountFailedException(f"{msg} {result}") else: return f"{msg} {result}" def estimate_mounting_time(self, segment) -> int: + """Adds additional time if cooling/drying is expected based on requested segment, + current sample segment and gripper position. + :param segment: any - however valid segment ABCDEFX """ try: current_mounted = self.get_mounted_sample() gripper_in_cold = self.is_in_cold() @@ -327,24 +219,21 @@ class TellClient: wait: bool = False, # blocking operation timeout: float = 600.0, ): + """send api request to mount sample from dewer after validating dewer address returns None or repsonse. + If the robot is busy, mount will raise an exception. + :param address: SampleDewarAddress + :param force: bool + :param read_dm: bool + :param auto_unmount: bool + :param wait: bool + :param timeout: float + """ SampleDewarAddress.model_validate(address) segment = address.puck.segment puck = address.puck.pos sample = address.pin - if self.__simulation: - self._last_cmd_id = random.randint(1000, 9999) - logger.info( - f"simulated mount({segment}, {puck}, {sample}) -> {self._last_cmd_id}" - ) - self._simulated_mounted_sample = f"{segment}{puck}{sample}" - if random.random() < 0.1: - logger.info("simulated failed mount") - raise TellMountFailedException(f"mount failed for {segment}{puck}") - - return self._last_cmd_id - if self.is_busy(): raise TellCommandWhileBusyException("mount received while robot is busy") @@ -386,10 +275,10 @@ class TellClient: return None def unmount(self, force=False, wait=False, timeout=360.0): - # Force has a meaning, will unmount even if smart magnet is not detecting sample - if self.__simulation: - print("simulated unmount") - return + """send api request to unmount sample from dewer returns None or repsonse. + :param force: bool Force has a meaning, will unmount even if smart magnet is not detecting sample + :param wait: bool If true will wait until unmount is completed + :timeout: float""" if self.is_busy(): raise TellCommandWhileBusyException("mount received while robot is busy") @@ -401,72 +290,51 @@ class TellClient: return self._last_cmd_id - def scan_pin(self, segment, puck, sample, force=False): - return self.start_cmd("scan_pin", segment, puck, sample, force) - - def scan_puck(self, segment, puck, force=False): - return self.start_cmd("scan_puck", segment, puck, force) - def dry(self, heat_time=None, speed=None, wait_cold=None, wait=False): - if self.__simulation: - time.sleep(5.0) + """send api request to dry tell gripper. + :param: heat_time float if None Tell will use default for drying time + :param: speed float if None Tell will use default for drying speed + :param: wait_cold bool if -1 to go to park after dry. if None Tell will use default time to wait_cold. + :param wait: bool If true will wait until drying is completed + """ self.pshell.wait_state("Ready", timeout=30.0) self._last_cmd_id = self.start_cmd("dry", heat_time, speed, wait_cold) if wait: self.check_command_ok(timeout=360.0, msg=f"Dry failed") def move_park(self, wait=False): - if self.__simulation: - return + """send api request to move robot to park position""" self._last_cmd_id = self.start_cmd("move_park") if wait: self.check_command_ok(timeout=360.0, msg=f"Move to park failed") def move_cold(self, reset_timestamp=False, wait=False): - if self.__simulation: - return + """send api request to move robot to cold position""" self._last_cmd_id = self.start_cmd("move_cold", reset_timestamp) if wait: self.check_command_ok(timeout=360.0, msg=f"Move to cold failed") - def trash(self): - if self.__simulation: - return - return self.start_cmd("trash_sample") - def abort_cmd(self): - if self.__simulation: - print("simulated abort") - return + """sends an abort pshell requesst and a robot stop task command""" self.pshell.abort() self.pshell.eval("robot.stop_task()&") - def set_gonio_mount_position(self, homing=False): - if self.__simulation: - return - if homing: - self.pshell.eval("home_fast_table()") - self.pshell.eval("set_mount_position()") - def set_setting(self, key: str, value: str): + """wrapper for pshell eval set_setting command + :param key str, name of a setting in tell + :param value str, the new value of the setting as a string""" self.pshell.eval(f"set_setting('{key}', '{value}')&") def get_setting(self, key: str) -> str: + """wrapper for pshell eval get_setting command, returns the current value for key as a string + :param key str, name of a setting in tell""" return self.pshell.eval(f"get_setting('{key}')&") - def enable_room_temperature(self): - self.set_setting("room_temperature_enabled", "true") - - def disable_room_temperature(self): - self.set_setting("room_temperature_enabled", "false") - def get_mounted_sample(self) -> SampleDewarAddress | None: - if self.__simulation: - ret = self._simulated_mounted_sample - else: - ret = self.pshell.eval("get_setting('mounted_sample_position')&").strip() + """get the current mounted sample and return a SampleDewarAddress object or None if no sample is mounted""" + ret = self.get_setting('mounted_sample_position').strip() if not ret or len(ret) == 0: return None @@ -481,47 +349,21 @@ class TellClient: return None def get_system_check(self): - if self.__simulation: - if random.random() < 0.1: - raise RuntimeError("get_system_check_failed") - return "Ok" + """returns the current system check status""" return self.pshell.eval("system_check_msg()&") def get_robot_state(self): - if self.__simulation: - return "Ready" + """returns the current robot state""" return self.pshell.eval("robot.state&") def get_robot_status(self): - if self.__simulation: - return { - "powered": True, - "settled": True, - "speed": 100, - "empty": True, - "mode": "remote", - "task": None, - "pos": "pCold", - "open": True, - "status": "move", - } - + """returns the current robot status""" status = self.pshell.eval("robot.take()&") return eval(status) # FIXME ALL functions must return a valid JSON object - def get_speed(self) -> float: - if self.__simulation: - if random.random() < 0.1: - return random.choice([1, 5, 25, 50, 75, 90]) - return 100.0 - speed = self.get_robot_status()["speed"] - return float(speed) - def get_detected_pucks(self) -> List[PuckLoadedInfo]: - if self.__simulation: - j = self._simulated_detected_pucks - else: - j = json.loads(self.pshell.eval("get_pucks_info()&")) + """returns a list of detected pucks as PuckLoadedInfo objects""" + j = json.loads(self.pshell.eval("get_pucks_info()&")) output = [] @@ -539,17 +381,8 @@ class TellClient: ) return output - def set_pin_offset(self, value): - if self.__simulation: - print(f"simulated set_pin_offset {value}") - self._simulated_offset = value - return - self.pshell.eval("set_pin_offset(" + str(value) + ")&") - def get_pin_offset(self): - if self.__simulation: - print(f"simulated get_pin_offset -> {self._simulated_offset}") - return self._simulated_offset + """get the pin offset for the smart magnet, returns offset as a float""" try: offset = float(self.pshell.eval("get_pin_offset()&")) except Exception: @@ -557,112 +390,79 @@ class TellClient: return offset def get_current(self): - if self.__simulation: - return self._simulated_current + """get the current drawn by the smart magnet, returns current as a float in mA""" current = self.pshell.eval("smart_magnet.get_current_rb()&") return float(current) - def set_current(self, current): - if self.__simulation: - self._simulated_current = current - return + def set_current(self, current: float) -> float: + """set the current drawn by the smart magnet, returns current as a float in mA""" self.pshell.eval("smart_magnet.set_current({:.1f})&".format(current)) current = self.pshell.eval("smart_magnet.get_current_rb()&") return float(current) - def print_info(self): - print("State: " + str(self.get_state())) - print("Mounted sample: " + str(self.get_mounted_sample())) - print("System check: " + str(self.get_system_check())) - print("Robot state: " + str(self.get_robot_state())) - print("Robot status: ") - status = self.get_robot_status() - status = status - for k, v in status.items(): - print(f"{k:>10s}: {v}") - print("Pin offset: " + str(self.get_pin_offset())) - print("Mount position: " + str(self.is_in_mount_position())) - print("") - def is_powered(self): - if self.__simulation: - return True + """returns True if the robot is powered on""" return self.get_robot_status()["powered"] def check_enable_motion(self): - if self.__simulation: - if random.random() < 0.1: - raise RuntimeError("check_enable_motion failed") + """check if the robot is powered on and enable motion if not""" if not self.is_powered(): self.pshell.eval("enable_motion()&") - def is_in_park(self): - if self.__simulation: - return True - return self.is_position(POSITION_PARK) - - def is_in_home(self): - if self.__simulation: - return False - return self.is_position(POSITION_HOME) - def is_in_cold(self): - if self.__simulation: - return False + """Compare current robot position to the set cold position. Returns True if in cold position, False otherwise.""" return self.is_position(POSITION_COLD) def is_position(self, position: str) -> bool: + """Compare current robot position to a given position. Returns True if in position, False otherwise.""" return position == self.get_robot_status()["pos"] - def get_task(self): - """ - robot_status = { - 'powered': False, - 'settled': True, - 'speed': 10, - 'empty': True, - 'mode': 'remote', - 'task': None, - 'pos': 'pPark', - 'open': True, - 'status': 'hold' - } - :return: - """ - status = self.get_robot_status() - return status["task"] - def is_ready(self): + """returns True if the robot is ready to receive commands""" return "ready" == self.get_state().lower() def is_busy(self): + """returns True if the robot is busy""" return "busy" == self.get_state().lower() def check_smart_magnet_mounted(self, timeout: float = 10.0, idle_time: float = 1.0, interval: float = 0.1): + """Reads smart_magent state and tries to infer if a sample is present + Handles: PAUSED, Fault, Busy and Ready states. + Raises a ManualMountException is the amgnet indicates a sample is present but get_mounted_sample is None. + Raises a SmartMagnetFaultException if the magnet detects no sample but the robot thinks a sample is mounted""" + #TODO tidy up initial_state = self.pshell.eval("smart_magnet.state&") + logger.debug(f"checking smart magnet_initial state: {initial_state}") + if initial_state == "Paused": self.pshell.eval("smart_magnet.set_supress(False)&") self.pshell.eval("smart_magnet.set_resting_current()&") + elif initial_state == "Fault": logger.error(f"tell smart magnet is in unknown state {initial_state}") raise SmartMagnetFaultException - #time.sleep(1.0) + state = self.pshell.eval("smart_magnet.state&") try: - # sample_present = bool(self.pshell.eval( - # f"smart_magnet.check_mounted(idle_time={str(idle_time)}, timeout={str(timeout)}, interval={str(interval)})")) - #logger.debug(f"sample present: {sample_present} of type {type(sample_present)}") - #time.sleep(1.0) if state == "Busy": logger.debug('state busy') self.pshell.eval("smart_magnet.set_supress(True)&") self.pshell.eval("smart_magnet.state&") sample_present = True + if self.get_mounted_sample() is None: + logger.warning("Check mount: A manually mounted sample is detected.") + logger.warning("Remove before mounting with the robot.") + raise ManualMountException + return True elif state == "Ready": logger.debug('No sample detected, ready to mount') sample_present = False + if self.get_mounted_sample(): + logger.error("Check mount: No sample detected, but robot thinks is mounted") + raise SmartMagnetFaultException + return False elif state == "Paused": logger.debug("Smart magnet detection is paused") return None @@ -670,32 +470,105 @@ class TellClient: self.pshell.eval("smart_magnet.set_supress(True)&") logger.error(f"Tell smart magnet is in unknown state {state}") raise SmartMagnetFaultException - if sample_present: - print(self.get_mounted_sample()) - if self.get_mounted_sample() is None: - logger.warning("Check mount: A manually mounted sample is detected.") - logger.warning("Remove before mounting with the robot.") - raise ManualMountException - return True - elif self.get_mounted_sample(): - logger.error("Check mount: No sample detected, but robot thinks is mounted") - raise SmartMagnetFaultException - return False + except Exception as e: logger.error(f"check_smart_magnet_mounted failed: {e}") raise e - #sample_present = False +class SimTellClient: + """ + Simulation-only Tell client. -def is_true(value): - """check if argument is semantically true""" - value = str(value).lower() - return value != "0" or value in ("true", "yes", "on", "enabled") + Keeps behavior deterministic-ish and stateful without needing PShellClient. + Implement more methods as your callers need them. + """ + def __init__(self): + self._state = "Ready" + self._last_cmd_id = 1000 + self._mounted_sample: str = "" + self._simulated_samples_info = {} + self._simulated_detected_pucks = [] + self._simulated_current = 30.0 + self._simulated_suppress = True + self._simulated_offset = 0.0 + @property + def url(self): + return None -def is_false(value): - return not is_true(value) + def _next_cmd_id(self) -> int: + self._last_cmd_id += 1 + return self._last_cmd_id + def get_state(self) -> str: + return self._state -def is_valid_dewar_position(position): - return position in VALID_DEWAR_POSITIONS + def is_ready(self) -> bool: + return self._state.lower() == "ready" + + def is_busy(self) -> bool: + return self._state.lower() == "busy" + + def wait_ready(self, timeout: float = 360.0): + # Keep it simple: flip to Ready quickly. + time.sleep(0.05) + self._state = "Ready" + + def mount( + self, + address: SampleDewarAddress, + force: bool = False, + read_dm: bool = False, + auto_unmount: bool = False, + wait: bool = False, + timeout: float = 600.0, + ): + SampleDewarAddress.model_validate(address) + if self.is_busy(): + raise TellCommandWhileBusyException("mount received while robot is busy") + + cmd_id = self._next_cmd_id() + self._state = "Busy" + + segment = address.puck.segment + puck = address.puck.pos + sample = address.pin + self._mounted_sample = f"{segment}{puck}{sample}" + + if wait: + self.wait_ready(timeout=timeout) + else: + # quickly become ready anyway, but asynchronously-ish + time.sleep(0.01) + self._state = "Ready" + + return cmd_id + + def unmount(self, force: bool = False, wait: bool = False, timeout: float = 360.0): + if self.is_busy(): + raise TellCommandWhileBusyException("unmount received while robot is busy") + + cmd_id = self._next_cmd_id() + self._state = "Busy" + self._mounted_sample = "" + if wait: + self.wait_ready(timeout=timeout) + else: + time.sleep(0.01) + self._state = "Ready" + return cmd_id + + def get_mounted_sample(self) -> SampleDewarAddress | None: + ret = self._mounted_sample + if not ret: + return None + match = re.match(r"([A-Z])(\d)(\d{1,2})", ret) + if not match: + return None + segment, puck, sample = match.groups() + return SampleDewarAddress(puck=DewarAddress(segment=segment, pos=int(puck)), pin=int(sample)) + +def make_tell_client(bl: MXBeamline) -> TellClient | SimTellClient: + if bl == MXBeamline.SIMULATED: + return SimTellClient() + return TellClient(bl) \ No newline at end of file