mx/AareDAQ
0
1
Fork 0
Code Issues 52 Pull Requests Actions Packages Projects Releases 72 Wiki Activity

TELL: major refactor, removing redundant code, added a simulated tell client, updated docstrings and comments and converted wait functions to use inbuilt pshell functionality.

Browse Source
This commit is contained in:
appleb_m
2026-02-19 17:12:01 +01:00
parent c77ddad56b
commit 9dffeec1e5
1 changed files with 210 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/devices/tell_client.py
+210 -337
View File
@@ -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)