TELL: added http exception hadnling so we don't swallow tell errors. Added pin blower control functions.

This commit is contained in:
2026-04-22 10:19:52 +02:00
parent 2920a38647
commit c05969a02b
2 changed files with 31 additions and 3 deletions
+19 -1
View File
@@ -2,6 +2,7 @@ import json
import re
import time
from typing import Any, Callable, Protocol
from requests.exceptions import HTTPError
from urllib.parse import urlparse
import requests
@@ -149,8 +150,25 @@ class PShellTellBackend:
def wait_events(self, events: dict[str, Any], timeout: float):
return self._pshell.wait_events(events, timeout=timeout)
def _handle_pshell_error(self, e: HTTPError, operation: str, endpoint: str):
"""Extract meaningful error message from HTTPError response."""
msg = "Unknown server error"
if e.response is not None and e.response.text:
msg = e.response.text.strip()
raise TellCommunicationError(
msg,
base_url=self._url,
endpoint=endpoint,
operation=operation,
) from e
def eval(self, expr: str):
return self._pshell.eval(expr)
try:
return self._pshell.eval(expr)
except HTTPError as e:
self._handle_pshell_error(e, "EVAL", f"eval/{expr}")
except Exception:
raise
def start_eval(self, expr: str) -> int:
return self._pshell.start_eval(expr)
+12 -2
View File
@@ -263,6 +263,12 @@ class TellClient:
def get_setting(self, key: str) -> str:
return self.backend.eval(f"get_setting('{key}')&")
def blower_on(self):
self.backend.eval("blower_on()&")
def blower_off(self):
self.backend.eval("blower_off()&")
def get_mounted_sample(self) -> SampleDewarAddress | None:
ret = self.get_setting("mounted_sample_position").strip()
if not ret:
@@ -327,7 +333,8 @@ class TellClient:
return self.get_robot_status()["powered"]
def check_enable_motion(self):
return True
if not self.is_powered():
self.backend.eval("enable_motion()&")
def is_in_cold(self):
return self.is_position(POSITION_COLD)
@@ -398,6 +405,9 @@ def make_tell_client(bl: MXBeamline) -> TellClient:
if __name__ == "__main__":
from aare.common.beamline import mx_beamline
import time
bl = mx_beamline()
tell_client = make_tell_client(bl)
tell_client.state()
tell_client.blower_on()
# time.sleep(5)
tell_client.blower_off()