702 lines
25 KiB
Python
Executable File
702 lines
25 KiB
Python
Executable File
import json
|
|
import random
|
|
import re
|
|
import time
|
|
from typing import List
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
from aare.common.logger_config import setup_logger
|
|
from aare.common.models import (
|
|
PuckLoadedInfo,
|
|
SampleShortInfo,
|
|
DewarAddress,
|
|
SampleDewarAddress,
|
|
#PuckInfo,
|
|
)
|
|
from aareDB import PuckWithTellPosition
|
|
|
|
from aare.common.beamline import MXBeamline # noqa: F401
|
|
from pshell import PShellClient
|
|
|
|
logger = setup_logger("aareDAQ")
|
|
|
|
class ManualMountException(Exception):
|
|
"""Custom exception for manual mounting"""
|
|
pass
|
|
|
|
|
|
class SmartMagnetFaultException(Exception):
|
|
"""Custom exception for smart magnet fault"""
|
|
pass
|
|
|
|
|
|
class TellMountFailedException(Exception):
|
|
"""Custom exception for mount failure"""
|
|
pass
|
|
|
|
|
|
class TellCommandWhileBusyException(Exception):
|
|
"""Custom exception for trying to move Tell when it is busy"""
|
|
pass
|
|
|
|
|
|
class TellConnectionException(Exception):
|
|
"""Custom exception for connection problems"""
|
|
pass
|
|
|
|
VALID_DEWAR_POSITIONS = [f"{p}{n}" for n in "12345" for p in "ABCDEFX"]
|
|
|
|
POSITION_PARK = "pPark"
|
|
POSITION_COLD = "pCold"
|
|
POSITION_AUX = "pAux"
|
|
POSITION_DEWAR = "pDewar"
|
|
POSITION_HOME = "pHome"
|
|
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:
|
|
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.__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")
|
|
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
|
|
self._aborted = False
|
|
self.state = self.get_state()
|
|
self.debug = False
|
|
self._last_cmd_id = -1
|
|
|
|
@property
|
|
def url(self):
|
|
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
|
|
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"),
|
|
}
|
|
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 set_in_mount_position(self, value):
|
|
if self.__simulation:
|
|
return
|
|
self.pshell.eval("in_mount_position = " + str(value) + "&")
|
|
|
|
def is_in_mount_position(self):
|
|
if self.__simulation:
|
|
return True
|
|
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"""
|
|
|
|
j = []
|
|
for x in info:
|
|
j.append(
|
|
{
|
|
"userName": x.pgroup,
|
|
"dewarName": x.dewar_name or "",
|
|
"puckName": x.puck_name,
|
|
"puckType": "Unipuck", # could use x.puck_type
|
|
"puckAddress": x.tell_position or "",
|
|
"puckBarcode": x.puck_name,
|
|
"sampleBarcode": "",
|
|
"sampleMountCount": 0,
|
|
"sampleName": "",
|
|
"samplePosition": 1,
|
|
"sampleStatus": "",
|
|
}
|
|
)
|
|
|
|
self.pshell.run("data/set_samples_info", pars=[json.dumps(j)], background=True)
|
|
# self.pshell.eval("set_samples_info(" + json.dumps(info) + ")&")
|
|
|
|
def start_cmd(self, cmd, *argv):
|
|
"""starts a command on the robot and returns the command id"""
|
|
cmd = cmd + "("
|
|
for a in argv:
|
|
cmd = cmd + (("'" + a + "'") if type(a) is str else str(a)) + ", "
|
|
cmd = cmd + ")"
|
|
ret = self.pshell.start_eval(cmd)
|
|
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"""
|
|
self.wait_not_busy(timeout)
|
|
result = self.get_result(self._last_cmd_id)
|
|
logger.debug(f"{msg} {result}")
|
|
status = result["status"]
|
|
if "completed" != status:
|
|
if "removed" != status:
|
|
raise TellMountFailedException(f"{msg} {result}")
|
|
else:
|
|
return f"{msg} {result}"
|
|
|
|
def estimate_mounting_time(self, segment) -> int:
|
|
try:
|
|
current_mounted = self.get_mounted_sample()
|
|
gripper_in_cold = self.is_in_cold()
|
|
|
|
if current_mounted is None:
|
|
unmount_needs_drying = 0 # might not have anything
|
|
unmount_needs_cooling = 0
|
|
else:
|
|
segment_in_cold = current_mounted.puck.segment in "ABCDEF"
|
|
unmount_needs_drying = int(gripper_in_cold and not segment_in_cold)
|
|
unmount_needs_cooling = int(not gripper_in_cold and segment_in_cold)
|
|
|
|
mount_needs_cooling = int(segment in "ABCDEF" and not gripper_in_cold)
|
|
mount_needs_drying = int(segment not in "ABCDEF" and gripper_in_cold)
|
|
|
|
needs_cooling = mount_needs_cooling + unmount_needs_cooling
|
|
needs_drying = mount_needs_drying + unmount_needs_drying
|
|
return needs_cooling * 30 + needs_drying * 120
|
|
except:
|
|
return 0
|
|
|
|
def mount(
|
|
self,
|
|
address: SampleDewarAddress,
|
|
force: bool = False, # kept for future
|
|
read_dm: bool = False, # read data matrix
|
|
auto_unmount: bool = False, # single command, if False it will raise exception
|
|
wait: bool = False, # blocking operation
|
|
timeout: float = 600.0,
|
|
):
|
|
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")
|
|
|
|
logger.info(f"loading sample {sample} from segment {segment} - {puck}")
|
|
|
|
self._last_cmd_id = self.start_cmd(
|
|
"mount", segment, puck, sample, force, read_dm, auto_unmount
|
|
)
|
|
|
|
wait_timeout = timeout + self.estimate_mounting_time(segment)
|
|
logger.info("waiting for mount to complete")
|
|
if wait and segment in "ABCDEF":
|
|
event, value = self.pshell.wait_events({"state": None, "motion_task": "dry", "gripper_detection" : "No Pin in Gripper"}, timeout=wait_timeout)
|
|
if event is None or event == "state":
|
|
logger.info(f"event: {event} occurred with value: {value}, checking command completed okay")
|
|
self.check_command_ok(
|
|
timeout=wait_timeout, msg=f"Mount {segment}{puck}-{sample}: "
|
|
)
|
|
return value
|
|
elif event == "gripper_detection":
|
|
logger.info(f"gripper detection: {event} occurred with value: {value}")
|
|
return value
|
|
elif event == "motion_task" and value == "dry":
|
|
logger.info(f"event: {event} occurred with value: {value}")
|
|
logger.info(" Drying occurring, releasing interface to user")
|
|
return value
|
|
else:
|
|
logger.info(f"Unexpected event: {event} occurred with value: {value}")
|
|
logger.info("Checking command completed okay anyway")
|
|
self.check_command_ok(
|
|
timeout=wait_timeout, msg=f"Mount {segment}{puck}-{sample}: "
|
|
)
|
|
elif wait and segment == "X":
|
|
logger.info("Loading an auxiliary puck")
|
|
self.check_command_ok(
|
|
timeout=wait_timeout, msg=f"Mount {segment}{puck}-{sample}: "
|
|
)
|
|
logger.info("post waiting")
|
|
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
|
|
|
|
if self.is_busy():
|
|
raise TellCommandWhileBusyException("mount received while robot is busy")
|
|
|
|
self._last_cmd_id = self.start_cmd("unmount", None, None, None, force)
|
|
|
|
if wait:
|
|
self.check_command_ok(timeout=timeout, msg="Unmount failed: ")
|
|
|
|
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)
|
|
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
|
|
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
|
|
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
|
|
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):
|
|
self.pshell.eval(f"set_setting('{key}', '{value}')&")
|
|
|
|
def get_setting(self, key: str) -> str:
|
|
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()
|
|
if not ret or len(ret) == 0:
|
|
return None
|
|
|
|
match = re.match(r"([A-Z])(\d)(\d{1,2})", ret)
|
|
|
|
if match:
|
|
segment, puck, sample = match.groups()
|
|
dewar_location = DewarAddress(segment=segment, pos=int(puck))
|
|
return SampleDewarAddress(puck=dewar_location, pin=int(sample))
|
|
else:
|
|
logger.warning(f"Failed to decode mounted sample position: {ret}")
|
|
return None
|
|
|
|
def get_system_check(self):
|
|
if self.__simulation:
|
|
if random.random() < 0.1:
|
|
raise RuntimeError("get_system_check_failed")
|
|
return "Ok"
|
|
return self.pshell.eval("system_check_msg()&")
|
|
|
|
def get_robot_state(self):
|
|
if self.__simulation:
|
|
return "Ready"
|
|
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",
|
|
}
|
|
|
|
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()&"))
|
|
|
|
output = []
|
|
|
|
for i in j:
|
|
if i["puckState"] == "Present":
|
|
puck_address = i["puckAddress"]
|
|
if len(puck_address) == 2:
|
|
output.append(
|
|
PuckLoadedInfo(
|
|
puck_name=i["puckBarcode"],
|
|
location=DewarAddress(
|
|
segment=puck_address[0], pos=int(puck_address[1])
|
|
),
|
|
),
|
|
)
|
|
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
|
|
try:
|
|
offset = float(self.pshell.eval("get_pin_offset()&"))
|
|
except Exception:
|
|
offset = 0.0
|
|
return offset
|
|
|
|
def get_current(self):
|
|
if self.__simulation:
|
|
return self._simulated_current
|
|
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
|
|
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
|
|
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")
|
|
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
|
|
return self.is_position(POSITION_COLD)
|
|
|
|
def is_position(self, position: str) -> bool:
|
|
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):
|
|
return "ready" == self.get_state().lower()
|
|
|
|
def is_busy(self):
|
|
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):
|
|
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
|
|
elif state == "Ready":
|
|
logger.debug('No sample detected, ready to mount')
|
|
sample_present = False
|
|
elif state == "Paused":
|
|
logger.debug("Smart magnet detection is paused")
|
|
return None
|
|
else:
|
|
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
|
|
|
|
|
|
def is_true(value):
|
|
"""check if argument is semantically true"""
|
|
value = str(value).lower()
|
|
return value != "0" or value in ("true", "yes", "on", "enabled")
|
|
|
|
|
|
def is_false(value):
|
|
return not is_true(value)
|
|
|
|
|
|
def is_valid_dewar_position(position):
|
|
return position in VALID_DEWAR_POSITIONS
|