WIP: Fix/socket polling #101
@@ -939,11 +939,11 @@ class DigitalTwin(BECWidget, QWidget):
|
||||
Calculates bragg angle in rad
|
||||
"""
|
||||
xtal = self.input.mo1_xtal.currentText()
|
||||
if xtal == "Si(111)":
|
||||
if xtal == "Si111":
|
||||
d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=True)[
|
||||
"mo1_bragg_crystal_d_spacing_si111"
|
||||
]["value"]
|
||||
elif xtal == "Si(311)":
|
||||
elif xtal == "Si311":
|
||||
d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si311.read(cached=True)[
|
||||
"mo1_bragg_crystal_d_spacing_si311"
|
||||
]["value"]
|
||||
|
||||
@@ -110,7 +110,7 @@ class InputPanel(QWidget):
|
||||
|
||||
# Monochromator
|
||||
self.mo1_mode = ComboBox("mo1_mode", "Mode", ["Monochromatic", "Pinkbeam"])
|
||||
self.mo1_xtal = ComboBox("mo1_xtal", "Crystal", ["Si(111)", "Si(311)"])
|
||||
self.mo1_xtal = ComboBox("mo1_xtal", "Crystal", ["Si111", "Si311"])
|
||||
self.mo1_bragg_angle = NumberIndicator("Bragg Angle", "deg", decimals=1)
|
||||
self.mo1_eres = NumberIndicator("Energy Resolution", "eV", decimals=2)
|
||||
self.mo1_ass_group = Group(
|
||||
|
||||
@@ -77,6 +77,15 @@ class ScanControlXAS(ScanControl):
|
||||
self._update_d_spacing, MessageEndpoints.device_readback("mo1_bragg")
|
||||
)
|
||||
|
||||
# Read once manually to get a first value
|
||||
if "mo1_bragg" in self.dev:
|
||||
self.d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=True)[
|
||||
"mo1_bragg_crystal_d_spacing_si111"
|
||||
]["value"]
|
||||
self.xas_scans_helper_widget.update_plot(d_spacing=self.d_spacing)
|
||||
else:
|
||||
logger.warning("mo1_bragg not in config, widget will not plot anything!")
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def _update_d_spacing(self, msg: dict, _: dict):
|
||||
d_spacing = msg["signals"].get("mo1_bragg_crystal_current_d_spacing")["value"]
|
||||
@@ -293,6 +302,7 @@ class XASScansHelper(QWidget):
|
||||
if self.d_spacing is None:
|
||||
return
|
||||
if d_spacing == 0:
|
||||
logger.warning("self.d_spacing is 0")
|
||||
return
|
||||
|
||||
x_time = np.linspace(0, self.scan_parameters["scan_time"], PLOT_RESOLUTION)
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
"""
|
||||
ACS controller device exposing plain read/write variables (no motion).
|
||||
|
||||
Uses the same BEC building blocks as before:
|
||||
- debye_bec.devices.utils.term_trail_controller.TermTrailController
|
||||
-> shared TCP/IP communicator (vendored term/trail extension of
|
||||
ophyd_devices.utils.controller.Controller)
|
||||
- ophyd_devices.utils.socket.SocketIO -> raw socket helper
|
||||
- ophyd_devices.utils.socket.SocketSignal -> Signal base talking through it
|
||||
|
||||
Protocol:
|
||||
read: "?GETVAR(tag)" -> reply is the value
|
||||
write: "SETVAR(value,tag)"
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
|
||||
# import traceback
|
||||
from enum import Enum
|
||||
|
||||
import numpy as np
|
||||
from bec_lib.logger import bec_logger
|
||||
from ophyd_devices.utils.controller import threadlocked
|
||||
from ophyd_devices.utils.socket import SocketSignal
|
||||
|
||||
from debye_bec.devices.utils.term_trail_controller import TermTrailController
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class ACSController(TermTrailController):
|
||||
"""
|
||||
Shared TCP/IP communicator for one ACS controller.
|
||||
|
||||
Instantiating this class twice with the same (socket_host, socket_port)
|
||||
returns the same object (see `Controller.__new__`), so every variable
|
||||
signal below -- across however many devices -- shares one connection.
|
||||
"""
|
||||
|
||||
_axes_per_controller = 0 # not used for plain variables, no motion axes
|
||||
|
||||
def __init__(self, *, socket_cls, socket_host, socket_port, device_manager):
|
||||
super().__init__(
|
||||
socket_cls=socket_cls,
|
||||
socket_host=socket_host,
|
||||
socket_port=socket_port,
|
||||
device_manager=device_manager,
|
||||
term="\r",
|
||||
trail=["\r:\r", ":\r"],
|
||||
socket_timeout=0.1,
|
||||
)
|
||||
|
||||
@threadlocked
|
||||
def get_var(self, tag: int, prec: int, idx: int | None = None) -> float:
|
||||
if self.sock is None:
|
||||
self.on()
|
||||
|
||||
idx = f",{idx:0.0f}" if idx is not None else ""
|
||||
reply = self.socket_put_and_receive(f"?{{%0.{prec:0.0f}f}}GETVAR({tag}{idx})")
|
||||
|
||||
if reply.startswith("?"):
|
||||
error = self._query_error(reply)
|
||||
raise RuntimeError(f"ACS error {reply}: {error}")
|
||||
|
||||
return float(reply)
|
||||
|
||||
@threadlocked
|
||||
def _query_error(self, reply: str) -> str:
|
||||
# reply is like "?2002"
|
||||
return self.socket_put_and_receive(f"?{reply}")
|
||||
|
||||
@threadlocked
|
||||
def set_var(self, tag: int, value, prec: int, idx: int | None = None) -> None:
|
||||
if self.sock is None:
|
||||
self.on()
|
||||
idx = f",{idx:0.0f}" if idx is not None else ""
|
||||
reply = self.socket_put_and_receive(f"SETVAR({np.round(value, prec)},{tag}{idx})")
|
||||
|
||||
if reply.startswith("?"):
|
||||
error = self._query_error(reply)
|
||||
raise RuntimeError(f"ACS error {reply}: {error}")
|
||||
|
||||
|
||||
class AcsSignal(SocketSignal):
|
||||
"""Read/write ACS controller variable, identified by its tag number."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*args,
|
||||
tag: int,
|
||||
prec: int,
|
||||
num_el: int = 1,
|
||||
enum: Enum = None,
|
||||
max_poll: float = 0.05,
|
||||
**kwargs,
|
||||
):
|
||||
self.tag = tag
|
||||
self.prec = prec
|
||||
self.num_el = num_el
|
||||
self.enum = enum
|
||||
self.last_get = time.time()
|
||||
self._poll_time = max_poll
|
||||
self._poll_thread = None
|
||||
self._stop_event = threading.Event()
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
@property
|
||||
def controller(self) -> ACSController:
|
||||
return self.root.controller
|
||||
|
||||
def _socket_get(self):
|
||||
now = time.time()
|
||||
interval = now - self.last_get
|
||||
self.last_get = now
|
||||
# logger.info(f"Get signal with tag {self.tag}, time to last get: {interval*1e3} ms")
|
||||
# logger.info(f"socket_get called from: {traceback.format_stack()}")
|
||||
|
||||
def convert(val):
|
||||
return self.enum(val).name if self.enum is not None else val
|
||||
|
||||
if self.num_el <= 1:
|
||||
val = convert(self.controller.get_var(self.tag, self.prec))
|
||||
# logger.info(f"Got signal with tag {self.tag} and value {val}, time to last get: {interval*1e3} ms")
|
||||
return val
|
||||
return np.array(
|
||||
[convert(self.controller.get_var(self.tag, self.prec, i)) for i in range(self.num_el)]
|
||||
)
|
||||
|
||||
def _socket_set(self, val):
|
||||
def convert(v):
|
||||
if self.enum is None:
|
||||
return v
|
||||
if isinstance(v, str):
|
||||
return self.enum[v].value # e.g. "SI111" -> 0
|
||||
return self.enum(v).value # e.g. 0 or Xtal.SI111 -> 0
|
||||
|
||||
if self.num_el <= 1:
|
||||
self.controller.set_var(self.tag, convert(val), self.prec)
|
||||
else:
|
||||
if len(val) != self.num_el:
|
||||
raise ValueError(
|
||||
f"Length of val ({len(val)}) must be equal to specified length of variable ({self.num_el})"
|
||||
)
|
||||
|
||||
for i, v in enumerate(val):
|
||||
self.controller.set_var(self.tag, convert(v), self.prec, i)
|
||||
|
||||
def subscribe(self, callback, event_type=None, run=True):
|
||||
self._ensure_polling()
|
||||
if run:
|
||||
self._force_fresh_read()
|
||||
return super().subscribe(callback, event_type=event_type, run=run)
|
||||
|
||||
def _force_fresh_read(self):
|
||||
old_value = self._readback
|
||||
try:
|
||||
new_value = self._socket_get()
|
||||
except Exception as e:
|
||||
logger.warning(f"Fresh read failed for {self.name} during subscribe() with {e}")
|
||||
return
|
||||
self._readback = new_value
|
||||
self._run_subs(
|
||||
sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time()
|
||||
)
|
||||
|
||||
def clear_sub(self, cb, event_type=None):
|
||||
super().clear_sub(cb, event_type=event_type)
|
||||
if not any(self._callbacks.values()):
|
||||
self._stop_polling()
|
||||
|
||||
def _ensure_polling(self):
|
||||
if self._poll_thread is None or not self._poll_thread.is_alive():
|
||||
self._stop_event.clear()
|
||||
self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True)
|
||||
self._poll_thread.start()
|
||||
|
||||
def _stop_polling(self):
|
||||
self._stop_event.set()
|
||||
|
||||
def _poll_loop(self):
|
||||
while not self._stop_event.is_set():
|
||||
old_value = self._readback
|
||||
try:
|
||||
new_value = self._socket_get()
|
||||
except Exception:
|
||||
time.sleep(self._poll_time)
|
||||
continue
|
||||
self._readback = new_value
|
||||
self._run_subs(
|
||||
sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time()
|
||||
)
|
||||
time.sleep(self._poll_time)
|
||||
|
||||
|
||||
class AcsSignalRO(AcsSignal):
|
||||
"""Readonly ACS controller variable, identified by its tag number."""
|
||||
|
||||
def _socket_set(self, val):
|
||||
return
|
||||
@@ -13,6 +13,7 @@ from typing import Literal
|
||||
|
||||
from bec_lib.devicemanager import ScanInfo
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_server.device_server.devices.devicemanager import DeviceManagerDS
|
||||
from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
|
||||
from ophyd import Component as Cpt
|
||||
from ophyd import DeviceStatus, StatusBase
|
||||
@@ -59,7 +60,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
|
||||
USER_ACCESS = ["set_advanced_xas_settings", "set_xtal", "convert_angle_energy"]
|
||||
|
||||
def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs): # type: ignore
|
||||
def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, device_manager: DeviceManagerDS | None = None, **kwargs): # type: ignore
|
||||
"""
|
||||
Initialize the PSI Device Base class.
|
||||
|
||||
@@ -67,7 +68,9 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
name (str) : Name of the device
|
||||
scan_info (ScanInfo): The scan info to use.
|
||||
"""
|
||||
super().__init__(name=name, scan_info=scan_info, prefix=prefix, **kwargs)
|
||||
super().__init__(
|
||||
name=name, scan_info=scan_info, prefix=prefix, device_manager=device_manager, **kwargs
|
||||
)
|
||||
self.scan_parameters: ScanServerScanInfo = None
|
||||
self.timeout_for_pvwait = 7.5
|
||||
self.valid_scan_names = [
|
||||
@@ -78,6 +81,8 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
"nidaq_continuous_scan",
|
||||
]
|
||||
|
||||
self.stage_start = None
|
||||
|
||||
########################################
|
||||
# Beamline Specific Implementations #
|
||||
########################################
|
||||
@@ -103,6 +108,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
|
||||
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
|
||||
"""
|
||||
self.stage_start = time.time()
|
||||
self.scan_parameters = fetch_scan_info(self.scan_info)
|
||||
if self.scan_control.scan_msg.get() != ScanControlLoadMessage.PENDING:
|
||||
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
|
||||
@@ -287,9 +293,9 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
else:
|
||||
return
|
||||
# Setting scan duration seems to lag behind slightly in the backend, include small sleep
|
||||
logger.info(f"Sleeping for one second")
|
||||
time.sleep(1)
|
||||
logger.info(f"Device {self.name}, done sleeping")
|
||||
# logger.info(f"Sleeping for one second")
|
||||
# time.sleep(1)
|
||||
# logger.info(f"Device {self.name}, done sleeping")
|
||||
# Load the scan parameters to the controller
|
||||
status = CompareStatus(
|
||||
self.scan_control.scan_msg,
|
||||
@@ -300,6 +306,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
self.scan_control.scan_load.put(1)
|
||||
# Wait for params to be checked from controller
|
||||
status.wait(self.timeout_for_pvwait)
|
||||
# logger.info(f"Starting scan took {time.time() - self.stage_start} s")
|
||||
return None
|
||||
|
||||
def on_unstage(self) -> DeviceStatus | StatusBase | None:
|
||||
@@ -346,7 +353,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
self.cancel_on_stop(status)
|
||||
logger.info(f"Finished calling complete on {self.name} within {time.time()-time_started}s.")
|
||||
return status
|
||||
|
||||
|
||||
def _status_callback(self, status, **kwargs):
|
||||
logger.info(f"Complete finished on mo1bragg with {status.done} and {status.success}")
|
||||
|
||||
@@ -380,7 +387,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
if scan_parameters.scan_name in self.valid_scan_names:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _progress_update(self, value, old_value, **kwargs) -> None:
|
||||
"""Callback method to update the scan progress, runs a callback
|
||||
to SUB_PROGRESS subscribers, i.e. BEC.
|
||||
@@ -449,13 +456,13 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
|
||||
in_signal = self.calculator.calc_energy
|
||||
out_signal = self.calculator.calc_angle
|
||||
else:
|
||||
raise Mo1BraggError(f'Unknown mode {mode}')
|
||||
raise Mo1BraggError(f"Unknown mode {mode}")
|
||||
|
||||
in_signal.put(inp)
|
||||
status = CompareStatus(self.calculator.calc_done, 1)
|
||||
self.cancel_on_stop(status)
|
||||
status.wait(self.timeout_for_pvwait)
|
||||
status = CompareStatus(out_signal, 0, operation_success='>')
|
||||
status = CompareStatus(out_signal, 0, operation_success=">")
|
||||
self.cancel_on_stop(status)
|
||||
status.wait(self.timeout_for_pvwait)
|
||||
return out_signal.get()
|
||||
|
||||
@@ -1,18 +1,18 @@
|
||||
"""Positioner implementation with readback angle of the MO1 Bragg positioner."""
|
||||
|
||||
from ophyd import Component as Cpt
|
||||
from ophyd import EpicsSignalRO, EpicsSignalWithRBV
|
||||
|
||||
from debye_bec.devices.mo1_bragg.acs import AcsSignal, AcsSignalRO
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggPositioner
|
||||
|
||||
|
||||
class Mo1BraggAngle(Mo1BraggPositioner):
|
||||
"""Positioner implementation with readback angle of the MO1 Bragg positioner."""
|
||||
|
||||
readback = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True)
|
||||
setpoint = Cpt(EpicsSignalWithRBV, suffix="set_abs_pos_angle", kind="normal", auto_monitor=True)
|
||||
low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_angle_RBV", kind="config", auto_monitor=True)
|
||||
high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_angle_RBV", kind="config", auto_monitor=True)
|
||||
readback = Cpt(AcsSignalRO, tag=51503, prec=6, kind="omitted")
|
||||
setpoint = Cpt(AcsSignal, tag=51501, prec=6, kind="omitted")
|
||||
low_lim = Cpt(AcsSignalRO, tag=51505, prec=1, kind="omitted")
|
||||
high_lim = Cpt(AcsSignalRO, tag=51504, prec=1, kind="omitted")
|
||||
|
||||
@property
|
||||
def egu(self) -> str:
|
||||
|
||||
@@ -1,24 +1,22 @@
|
||||
"""Module for the Mo1 Bragg positioner"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from typing import Literal
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_server.device_server.devices.devicemanager import DeviceManagerDS
|
||||
from ophyd import Component as Cpt
|
||||
from ophyd import (
|
||||
Device,
|
||||
DeviceStatus,
|
||||
EpicsSignal,
|
||||
EpicsSignalRO,
|
||||
EpicsSignalWithRBV,
|
||||
PositionerBase,
|
||||
Signal,
|
||||
)
|
||||
from ophyd import Device, DeviceStatus, PositionerBase, Signal
|
||||
from ophyd.utils import LimitError
|
||||
from ophyd_devices.utils.socket import SocketIO
|
||||
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType
|
||||
# from debye_bec.devices.mo1_bragg.acs_controller import ACSSignal
|
||||
from debye_bec.devices.mo1_bragg.acs import ACSController, AcsSignal, AcsSignalRO
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType, Xtal
|
||||
|
||||
# Initialise logger
|
||||
logger = bec_logger.logger
|
||||
@@ -39,9 +37,9 @@ class MoveTypeSignal(Signal):
|
||||
# pylint: disable=arguments-differ
|
||||
def set(self, value: str | MoveType) -> None:
|
||||
"""Returns currently active move method
|
||||
|
||||
Args:
|
||||
value (str | MoveType) : Can be either 'energy' or 'angle'
|
||||
auto_monitor=True
|
||||
Args:
|
||||
value (str | MoveType) : Can be either 'energy' or 'angle'
|
||||
"""
|
||||
|
||||
value = MoveType(value.lower())
|
||||
@@ -54,169 +52,130 @@ class MoveTypeSignal(Signal):
|
||||
class Mo1BraggStatus(Device):
|
||||
"""Mo1 Bragg PVs for status monitoring"""
|
||||
|
||||
error_status = Cpt(EpicsSignalRO, suffix="error_status_RBV", kind="config", auto_monitor=True)
|
||||
brake_enabled = Cpt(EpicsSignalRO, suffix="brake_enabled_RBV", kind="config", auto_monitor=True)
|
||||
mot_commutated = Cpt(
|
||||
EpicsSignalRO, suffix="mot_commutated_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
axis_enabled = Cpt(EpicsSignalRO, suffix="axis_enabled_RBV", kind="config", auto_monitor=True)
|
||||
enc_initialized = Cpt(
|
||||
EpicsSignalRO, suffix="enc_initialized_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
heartbeat = Cpt(EpicsSignalRO, suffix="heartbeat_RBV", kind="config", auto_monitor=True)
|
||||
error_status = Cpt(AcsSignalRO, tag=10000, prec=0, kind="config")
|
||||
brake_enabled = Cpt(AcsSignalRO, tag=10001, prec=0, kind="omitted")
|
||||
mot_commutated = Cpt(AcsSignalRO, tag=10002, prec=0, kind="omitted")
|
||||
axis_enabled = Cpt(AcsSignalRO, tag=10003, prec=0, kind="omitted")
|
||||
enc_initialized = Cpt(AcsSignalRO, tag=10004, prec=0, kind="omitted")
|
||||
heartbeat = Cpt(AcsSignalRO, tag=10005, prec=0, kind="omitted")
|
||||
|
||||
|
||||
class Mo1BraggEncoder(Device):
|
||||
"""Mo1 Bragg PVs to communicate with the encoder"""
|
||||
|
||||
enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit", kind="config")
|
||||
enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config")
|
||||
enc_reinit = Cpt(AcsSignal, tag=11000, prec=0, kind="omitted")
|
||||
enc_reinit_done = Cpt(AcsSignalRO, tag=11001, prec=0, kind="config")
|
||||
|
||||
|
||||
class Mo1BraggCrystal(Device):
|
||||
"""Mo1 Bragg PVs to set the crystal parameters"""
|
||||
|
||||
bragg_off_si111 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si111", kind="config")
|
||||
bragg_off_si311 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si311", kind="config")
|
||||
phi_off_si111 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si111", kind="config")
|
||||
phi_off_si311 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si311", kind="config")
|
||||
azm_off_si111 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si111", kind="config")
|
||||
azm_off_si311 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si311", kind="config")
|
||||
miscut_si111 = Cpt(EpicsSignalWithRBV, suffix="miscut_si111", kind="config")
|
||||
miscut_si311 = Cpt(EpicsSignalWithRBV, suffix="miscut_si311", kind="config")
|
||||
xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config")
|
||||
d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config")
|
||||
d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config")
|
||||
set_offset = Cpt(EpicsSignal, suffix="set_offset", kind="config", put_complete=True)
|
||||
current_d_spacing = Cpt(
|
||||
EpicsSignalRO, suffix="current_d_spacing_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_bragg_off = Cpt(
|
||||
EpicsSignalRO, suffix="current_bragg_off_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_phi_off = Cpt(
|
||||
EpicsSignalRO, suffix="current_phi_off_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_azm_off = Cpt(
|
||||
EpicsSignalRO, suffix="current_azm_off_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_miscut = Cpt(
|
||||
EpicsSignalRO, suffix="current_miscut_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_xtal = Cpt(
|
||||
EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
current_xtal_string = Cpt(
|
||||
EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True
|
||||
)
|
||||
bragg_off_si111 = Cpt(AcsSignal, tag=50500, prec=12, kind="config")
|
||||
bragg_off_si311 = Cpt(AcsSignal, tag=50501, prec=12, kind="config")
|
||||
phi_off_si111 = Cpt(AcsSignal, tag=50507, prec=12, kind="config")
|
||||
phi_off_si311 = Cpt(AcsSignal, tag=50508, prec=12, kind="config")
|
||||
azm_off_si111 = Cpt(AcsSignal, tag=50510, prec=12, kind="config")
|
||||
azm_off_si311 = Cpt(AcsSignal, tag=50511, prec=12, kind="config")
|
||||
miscut_si111 = Cpt(AcsSignal, tag=50513, prec=12, kind="config")
|
||||
miscut_si311 = Cpt(AcsSignal, tag=50514, prec=12, kind="config")
|
||||
d_spacing_si111 = Cpt(AcsSignal, tag=50000, prec=12, kind="config")
|
||||
d_spacing_si311 = Cpt(AcsSignal, tag=50001, prec=12, kind="config")
|
||||
|
||||
current_d_spacing = Cpt(AcsSignalRO, tag=50503, prec=12, kind="normal")
|
||||
current_bragg_off = Cpt(AcsSignalRO, tag=50504, prec=12, kind="config")
|
||||
current_phi_off = Cpt(AcsSignalRO, tag=50509, prec=12, kind="config")
|
||||
current_azm_off = Cpt(AcsSignalRO, tag=50512, prec=12, kind="config")
|
||||
current_miscut = Cpt(AcsSignalRO, tag=50515, prec=12, kind="config")
|
||||
current_xtal = Cpt(AcsSignalRO, tag=10501, prec=0, kind="config")
|
||||
current_xtal_string = Cpt(AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal")
|
||||
|
||||
|
||||
class Mo1BraggScanSettings(Device):
|
||||
"""Mo1 Bragg PVs to set the scan setttings"""
|
||||
|
||||
# TRIG settings
|
||||
trig_select_ref_enum = Cpt(EpicsSignalWithRBV, suffix="trig_select_ref_ENUM", kind="config")
|
||||
trig_select_ref_enum = Cpt(AcsSignal, tag=12504, prec=0, kind="config")
|
||||
|
||||
trig_ena_hi_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_hi_ENUM", kind="config")
|
||||
trig_time_hi = Cpt(EpicsSignalWithRBV, suffix="trig_time_hi", kind="config")
|
||||
trig_every_n_hi = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_hi", kind="config")
|
||||
trig_ena_hi_enum = Cpt(AcsSignal, tag=12501, prec=0, kind="config")
|
||||
trig_time_hi = Cpt(AcsSignal, tag=52501, prec=3, kind="config")
|
||||
trig_every_n_hi = Cpt(AcsSignal, tag=12503, prec=0, kind="config")
|
||||
|
||||
trig_ena_lo_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_lo_ENUM", kind="config")
|
||||
trig_time_lo = Cpt(EpicsSignalWithRBV, suffix="trig_time_lo", kind="config")
|
||||
trig_every_n_lo = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_lo", kind="config")
|
||||
trig_ena_lo_enum = Cpt(AcsSignal, tag=12500, prec=0, kind="config")
|
||||
trig_time_lo = Cpt(AcsSignal, tag=52500, prec=3, kind="config")
|
||||
trig_every_n_lo = Cpt(AcsSignal, tag=12502, prec=0, kind="config")
|
||||
|
||||
# XAS simple scan settings
|
||||
s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config")
|
||||
s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config")
|
||||
s_scan_energy_lo = Cpt(
|
||||
EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True
|
||||
)
|
||||
s_scan_energy_hi = Cpt(
|
||||
EpicsSignalWithRBV, suffix="s_scan_energy_hi", kind="config", auto_monitor=True
|
||||
)
|
||||
s_scan_scantime = Cpt(
|
||||
EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True
|
||||
)
|
||||
s_scan_angle_hi = Cpt(AcsSignal, tag=53001, prec=6, kind="omitted")
|
||||
s_scan_angle_lo = Cpt(AcsSignal, tag=53000, prec=6, kind="omitted")
|
||||
|
||||
# XAS advanced scan settings
|
||||
a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=True)
|
||||
a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=True)
|
||||
a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=True)
|
||||
s_scan_energy_lo = Cpt(AcsSignal, tag=53003, prec=6, kind="config")
|
||||
s_scan_energy_hi = Cpt(AcsSignal, tag=53004, prec=6, kind="config")
|
||||
s_scan_scantime = Cpt(AcsSignal, tag=53002, prec=3, kind="config")
|
||||
|
||||
a_scan_pos = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="config")
|
||||
a_scan_vel = Cpt(AcsSignal, tag=53501, prec=6, num_el=41, kind="config")
|
||||
a_scan_time = Cpt(AcsSignal, tag=53502, prec=6, num_el=41, kind="config")
|
||||
|
||||
|
||||
class Mo1TriggerSettings(Device):
|
||||
"""Mo1 Trigger settings"""
|
||||
|
||||
settle_time = Cpt(EpicsSignalWithRBV, suffix="settle_time", kind="config")
|
||||
max_dev = Cpt(EpicsSignalWithRBV, suffix="max_dev", kind="config")
|
||||
settle_time = Cpt(AcsSignal, tag=55000, prec=3, kind="config")
|
||||
max_dev = Cpt(AcsSignal, tag=55001, prec=6, kind="config")
|
||||
|
||||
xrd_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_src_ENUM", kind="config")
|
||||
xrd_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_mode_ENUM", kind="config")
|
||||
xrd_trig_len = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_len", kind="config")
|
||||
xrd_trig_period = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_period", kind="config")
|
||||
xrd_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="xrd_n_of_trig", kind="config")
|
||||
xrd_trig_req = Cpt(EpicsSignal, suffix="xrd_trig_req", kind="config")
|
||||
xrd_trig_src_enum = Cpt(AcsSignal, tag=14501, prec=0, kind="config")
|
||||
xrd_trig_mode_enum = Cpt(AcsSignal, tag=14502, prec=0, kind="config")
|
||||
xrd_trig_len = Cpt(AcsSignal, tag=54500, prec=3, kind="config")
|
||||
xrd_trig_period = Cpt(AcsSignal, tag=54504, prec=3, kind="config")
|
||||
xrd_n_of_trig = Cpt(AcsSignal, tag=14512, prec=0, kind="config")
|
||||
xrd_trig_req = Cpt(AcsSignal, tag=14500, prec=0, kind="config")
|
||||
|
||||
falcon_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_src_ENUM", kind="config")
|
||||
falcon_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_mode_ENUM", kind="config")
|
||||
falcon_trig_len = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_len", kind="config")
|
||||
falcon_trig_period = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_period", kind="config")
|
||||
falcon_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="falcon_n_of_trig", kind="config")
|
||||
falcon_trig_req = Cpt(EpicsSignal, suffix="falcon_trig_req", kind="config")
|
||||
falcon_trig_src_enum = Cpt(AcsSignal, tag=14504, prec=0, kind="config")
|
||||
falcon_trig_mode_enum = Cpt(AcsSignal, tag=14505, prec=0, kind="config")
|
||||
falcon_trig_len = Cpt(AcsSignal, tag=54501, prec=3, kind="config")
|
||||
falcon_trig_period = Cpt(AcsSignal, tag=54505, prec=3, kind="config")
|
||||
falcon_n_of_trig = Cpt(AcsSignal, tag=14513, prec=0, kind="config")
|
||||
falcon_trig_req = Cpt(AcsSignal, tag=14503, prec=0, kind="config")
|
||||
|
||||
univ1_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_src_ENUM", kind="config")
|
||||
univ1_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_mode_ENUM", kind="config")
|
||||
univ1_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_len", kind="config")
|
||||
univ1_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_period", kind="config")
|
||||
univ1_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ1_n_of_trig", kind="config")
|
||||
univ1_trig_req = Cpt(EpicsSignal, suffix="univ1_trig_req", kind="config")
|
||||
univ1_trig_src_enum = Cpt(AcsSignal, tag=14507, prec=0, kind="config")
|
||||
univ1_trig_mode_enum = Cpt(AcsSignal, tag=14508, prec=0, kind="config")
|
||||
univ1_trig_len = Cpt(AcsSignal, tag=54502, prec=3, kind="config")
|
||||
univ1_trig_period = Cpt(AcsSignal, tag=54506, prec=3, kind="config")
|
||||
univ1_n_of_trig = Cpt(AcsSignal, tag=14514, prec=0, kind="config")
|
||||
univ1_trig_req = Cpt(AcsSignal, tag=14506, prec=0, kind="config")
|
||||
|
||||
univ2_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_src_ENUM", kind="config")
|
||||
univ2_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_mode_ENUM", kind="config")
|
||||
univ2_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_len", kind="config")
|
||||
univ2_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_period", kind="config")
|
||||
univ2_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ2_n_of_trig", kind="config")
|
||||
univ2_trig_req = Cpt(EpicsSignal, suffix="univ2_trig_req", kind="config")
|
||||
univ2_trig_src_enum = Cpt(AcsSignal, tag=14510, prec=0, kind="config")
|
||||
univ2_trig_mode_enum = Cpt(AcsSignal, tag=14511, prec=0, kind="config")
|
||||
univ2_trig_len = Cpt(AcsSignal, tag=54503, prec=3, kind="config")
|
||||
univ2_trig_period = Cpt(AcsSignal, tag=54507, prec=3, kind="config")
|
||||
univ2_n_of_trig = Cpt(AcsSignal, tag=14515, prec=0, kind="config")
|
||||
univ2_trig_req = Cpt(AcsSignal, tag=14509, prec=0, kind="config")
|
||||
|
||||
|
||||
class Mo1BraggCalculator(Device):
|
||||
"""Mo1 Bragg PVs to convert angle to energy or vice-versa."""
|
||||
|
||||
calc_reset = Cpt(EpicsSignalWithRBV, suffix="calc_reset", kind="config", put_complete=True)
|
||||
calc_done = Cpt(EpicsSignalRO, suffix="calc_done_RBV", kind="config")
|
||||
calc_energy = Cpt(EpicsSignalWithRBV, suffix="calc_energy", kind="config")
|
||||
calc_angle = Cpt(EpicsSignalWithRBV, suffix="calc_angle", kind="config")
|
||||
calc_reset = Cpt(AcsSignal, tag=14000, prec=0, kind="omitted")
|
||||
calc_done = Cpt(AcsSignalRO, tag=14001, prec=0, kind="omitted")
|
||||
calc_energy = Cpt(AcsSignal, tag=54000, prec=0, kind="omitted")
|
||||
calc_angle = Cpt(AcsSignal, tag=54001, prec=0, kind="omitted")
|
||||
|
||||
|
||||
class Mo1BraggScanControl(Device):
|
||||
"""Mo1 Bragg PVs to control the scan after setting the parameters."""
|
||||
|
||||
scan_mode_enum = Cpt(EpicsSignalWithRBV, suffix="scan_mode_ENUM", kind="config")
|
||||
scan_duration = Cpt(
|
||||
EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True
|
||||
)
|
||||
scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=True)
|
||||
scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="config", auto_monitor=True)
|
||||
scan_start_infinite = Cpt(
|
||||
EpicsSignal, suffix="scan_start_infinite", kind="config", put_complete=True
|
||||
)
|
||||
scan_start_timer = Cpt(EpicsSignal, suffix="scan_start_timer", kind="config", put_complete=True)
|
||||
scan_stop = Cpt(EpicsSignal, suffix="scan_stop", kind="config", put_complete=True)
|
||||
scan_status = Cpt(
|
||||
EpicsSignalRO, suffix="scan_status_ENUM_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
scan_time_left = Cpt(
|
||||
EpicsSignalRO, suffix="scan_time_left_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="config", auto_monitor=True)
|
||||
scan_val_reset = Cpt(EpicsSignal, suffix="scan_val_reset", kind="config", put_complete=True)
|
||||
scan_progress = Cpt(EpicsSignalRO, suffix="scan_progress_RBV", kind="config", auto_monitor=True)
|
||||
scan_spectra_done = Cpt(
|
||||
EpicsSignalRO, suffix="scan_n_osc_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
scan_spectra_left = Cpt(
|
||||
EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="config", auto_monitor=True
|
||||
)
|
||||
scan_mode_enum = Cpt(AcsSignal, tag=12000, prec=0, kind="config")
|
||||
scan_duration = Cpt(AcsSignal, tag=52000, prec=1, kind="config")
|
||||
scan_load = Cpt(AcsSignal, tag=12001, prec=0, kind="omitted")
|
||||
scan_msg = Cpt(AcsSignalRO, tag=12007, prec=0, kind="config")
|
||||
scan_start_infinite = Cpt(AcsSignal, tag=12003, prec=0, kind="omitted")
|
||||
scan_start_timer = Cpt(AcsSignal, tag=12006, prec=0, kind="omitted")
|
||||
scan_stop = Cpt(AcsSignal, tag=12004, prec=0, kind="omitted")
|
||||
scan_status = Cpt(AcsSignalRO, tag=12002, prec=0, kind="config")
|
||||
scan_time_left = Cpt(AcsSignalRO, tag=52001, prec=1, kind="omitted")
|
||||
scan_done = Cpt(AcsSignalRO, tag=12005, prec=0, kind="omitted")
|
||||
scan_val_reset = Cpt(AcsSignal, tag=52000, prec=0, kind="omitted")
|
||||
scan_progress = Cpt(AcsSignalRO, tag=12011, prec=1, max_poll=0.5, kind="omitted")
|
||||
scan_spectra_done = Cpt(AcsSignalRO, tag=12009, prec=0, kind="omitted")
|
||||
scan_spectra_left = Cpt(AcsSignalRO, tag=12010, prec=0, kind="omitted")
|
||||
|
||||
|
||||
class Mo1BraggPositioner(Device, PositionerBase):
|
||||
@@ -241,31 +200,24 @@ class Mo1BraggPositioner(Device, PositionerBase):
|
||||
|
||||
############# Energy PVs #############
|
||||
|
||||
readback = Cpt(
|
||||
EpicsSignalRO, suffix="feedback_pos_energy_RBV", kind="hinted", auto_monitor=True
|
||||
)
|
||||
setpoint = Cpt(
|
||||
EpicsSignalWithRBV, suffix="set_abs_pos_energy", kind="normal", auto_monitor=True
|
||||
)
|
||||
motor_is_moving = Cpt(
|
||||
EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True
|
||||
)
|
||||
low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_energy_RBV", kind="config", auto_monitor=True)
|
||||
high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_energy_RBV", kind="config", auto_monitor=True)
|
||||
velocity = Cpt(EpicsSignalWithRBV, suffix="move_velocity", kind="config", auto_monitor=True)
|
||||
|
||||
angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True)
|
||||
readback = Cpt(AcsSignalRO, tag=51508, prec=3, kind="hinted")
|
||||
setpoint = Cpt(AcsSignal, tag=51507, prec=3, kind="normal")
|
||||
motor_is_moving = Cpt(AcsSignalRO, tag=11504, prec=0, max_poll=0.5, kind="normal")
|
||||
low_lim = Cpt(AcsSignalRO, tag=51510, prec=3, kind="config")
|
||||
high_lim = Cpt(AcsSignalRO, tag=51509, prec=3, kind="config")
|
||||
velocity = Cpt(AcsSignal, tag=51502, prec=3, kind="config")
|
||||
angle = Cpt(AcsSignalRO, tag=51503, prec=6, kind="normal")
|
||||
|
||||
########## Move Command PVs ##########
|
||||
|
||||
move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True)
|
||||
move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config", put_complete=True)
|
||||
move_abs = Cpt(AcsSignal, tag=11503, prec=0, kind="omitted")
|
||||
move_stop = Cpt(AcsSignal, tag=11509, prec=0, kind="omitted")
|
||||
|
||||
SUB_READBACK = "readback"
|
||||
_default_sub = SUB_READBACK
|
||||
SUB_PROGRESS = "progress"
|
||||
|
||||
def __init__(self, prefix="", *, name: str, **kwargs):
|
||||
def __init__(self, prefix="", *, name: str, device_manager: DeviceManagerDS, **kwargs):
|
||||
"""Initialize the Mo1 Bragg positioner.
|
||||
|
||||
Args:
|
||||
@@ -273,6 +225,13 @@ class Mo1BraggPositioner(Device, PositionerBase):
|
||||
name (str): Name of the device
|
||||
kwargs: Additional keyword arguments
|
||||
"""
|
||||
|
||||
host = "129.129.123.32"
|
||||
port = 701
|
||||
self.controller = ACSController(
|
||||
socket_cls=SocketIO, socket_host=host, socket_port=port, device_manager=device_manager
|
||||
)
|
||||
|
||||
super().__init__(prefix, name=name, **kwargs)
|
||||
self._move_thread = None
|
||||
self._stopped = False
|
||||
|
||||
@@ -3,6 +3,27 @@
|
||||
import enum
|
||||
|
||||
|
||||
class Xtal(int, enum.Enum):
|
||||
"""Enum class for the xtal (crystal) of the Bragg positioner"""
|
||||
|
||||
Si111 = 0
|
||||
Si311 = 1
|
||||
|
||||
|
||||
class TriggerSelectReference(int, enum.Enum):
|
||||
"""Enum class for the trigger reference selectrion"""
|
||||
|
||||
ANGLE = 0
|
||||
ENERGY = 1
|
||||
|
||||
|
||||
class TriggerEnable(int, enum.Enum):
|
||||
"""Enum class to enable/disable trigger control"""
|
||||
|
||||
DISABLED = 0
|
||||
ENABLED = 1
|
||||
|
||||
|
||||
class TriggerControlSource(int, enum.Enum):
|
||||
"""Enum class for the trigger control source of the trigger generator"""
|
||||
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
"""Socket controller base with a configurable wire protocol.
|
||||
|
||||
On top of the stock ``Controller`` this adds:
|
||||
- configurable outgoing line termination (``term``) and reply terminators
|
||||
(``trail``, a single string or several alternatives),
|
||||
- a buffered ``socket_get`` that reassembles replies split across TCP reads and
|
||||
rejects runaway replies (``max_reply_length``),
|
||||
- a per-controller ``socket_timeout`` for individual send/recv operations,
|
||||
- reconnect-before-retry on communication errors, discarding stale replies.
|
||||
"""
|
||||
|
||||
import functools
|
||||
import time
|
||||
import traceback
|
||||
|
||||
from bec_lib import bec_logger
|
||||
from ophyd_devices.utils.controller import Controller, ControllerCommunicationError, threadlocked
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
def retry_once_reconnected(fcn):
|
||||
"""Decorator to rerun a function once if a communication error was raised.
|
||||
|
||||
Reconnects first to discard any stale/desynced reply left on the wire from the
|
||||
failed attempt -- without this, a late-arriving reply to the *first* attempt could
|
||||
be misread as the reply to the retry.
|
||||
"""
|
||||
|
||||
@functools.wraps(fcn)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
try:
|
||||
val = fcn(self, *args, **kwargs)
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.warning(
|
||||
f"Communication error occurred. Reconnecting and retrying the command. Traceback: {content}"
|
||||
)
|
||||
self._reconnect()
|
||||
val = fcn(self, *args, **kwargs)
|
||||
return val
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
class TermTrailController(Controller):
|
||||
"""Socket controller with configurable term/trail strings and buffered replies.
|
||||
|
||||
Args:
|
||||
term (str, optional): Termination string appended to each outgoing socket request.
|
||||
Defaults to the class attribute _term ("\\n").
|
||||
trail (str | list[str], optional): Termination string(s) marking the end of a
|
||||
socket reply. Defaults to the class attribute _trail ("\\r\\n").
|
||||
max_reply_length (int, optional): Max accepted length of a socket reply in bytes.
|
||||
Defaults to 1024 bytes.
|
||||
socket_timeout (int | float, optional): Timeout for each socket operation in
|
||||
seconds. Defaults to 2 seconds.
|
||||
|
||||
Subclasses that need a different wire protocol should override the _term and _trail
|
||||
class attributes. The constructor arguments only take effect on the first
|
||||
instantiation per host:port; the controller is a singleton and later conflicting
|
||||
values are ignored with a warning.
|
||||
"""
|
||||
|
||||
_term = "\n" # termination string appended to each outgoing request
|
||||
_trail = "\r\n" # termination string(s) stripped from the end of each reply
|
||||
_max_reply_length = 1024 # max accepted length of a socket reply in bytes
|
||||
_socket_timeout = 2 # timeout for each socket operation in seconds
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
socket_cls,
|
||||
socket_host,
|
||||
socket_port,
|
||||
device_manager,
|
||||
name: str = "",
|
||||
attr_name="",
|
||||
parent=None,
|
||||
labels=None,
|
||||
kind=None,
|
||||
term: str | None = None,
|
||||
trail: str | list[str] | tuple[str, ...] | None = None,
|
||||
max_reply_length: int | None = None,
|
||||
socket_timeout: int | float | None = None,
|
||||
):
|
||||
if term is not None and not isinstance(term, str):
|
||||
raise TypeError(f"term must be a string, got {type(term).__name__}")
|
||||
if trail is not None:
|
||||
if isinstance(trail, str):
|
||||
pass
|
||||
elif isinstance(trail, (list, tuple)) and all(isinstance(t, str) for t in trail):
|
||||
if not trail:
|
||||
raise ValueError("trail must not be an empty list/tuple.")
|
||||
else:
|
||||
raise TypeError(
|
||||
f"trail must be a string or a list/tuple of strings, got {type(trail).__name__}"
|
||||
)
|
||||
if max_reply_length is not None and not isinstance(max_reply_length, int):
|
||||
raise TypeError(
|
||||
f"max_reply_length must be an int, got {type(max_reply_length).__name__}"
|
||||
)
|
||||
if socket_timeout is not None and not isinstance(socket_timeout, (int, float)):
|
||||
raise TypeError(f"socket_timeout must be a number, got {type(socket_timeout).__name__}")
|
||||
|
||||
first_init = not self._initialized
|
||||
super().__init__(
|
||||
socket_cls=socket_cls,
|
||||
socket_host=socket_host,
|
||||
socket_port=socket_port,
|
||||
device_manager=device_manager,
|
||||
name=name,
|
||||
attr_name=attr_name,
|
||||
parent=parent,
|
||||
labels=labels,
|
||||
kind=kind,
|
||||
)
|
||||
if first_init:
|
||||
if term is not None:
|
||||
self._term = term
|
||||
if trail is not None:
|
||||
self._trail = trail
|
||||
self._trail_options: tuple[str, ...] = (
|
||||
(self._trail,) if isinstance(self._trail, str) else tuple(self._trail)
|
||||
)
|
||||
if max_reply_length is not None:
|
||||
self._max_reply_length = max_reply_length
|
||||
if socket_timeout is not None:
|
||||
self._socket_timeout = socket_timeout
|
||||
elif (
|
||||
(term is not None and term != self._term)
|
||||
or (trail is not None and trail != self._trail)
|
||||
or (max_reply_length is not None and max_reply_length != self._max_reply_length)
|
||||
or (socket_timeout is not None and socket_timeout != self._socket_timeout)
|
||||
):
|
||||
logger.warning(
|
||||
f"Controller {self._socket_host}:{self._socket_port} is already initialized with "
|
||||
f"term={self._term!r}, trail={self._trail!r}, max_reply_length={self._max_reply_length!r}, "
|
||||
f"socket_timeout={self._socket_timeout!r}; ignoring conflicting values "
|
||||
f"term={term!r}, trail={trail!r}, max_reply_length={max_reply_length!r}, "
|
||||
f"socket_timeout={socket_timeout!r}."
|
||||
)
|
||||
|
||||
@threadlocked
|
||||
def _reconnect(self):
|
||||
"""
|
||||
Close and reopen the socket connection.
|
||||
|
||||
Required after any communication error, in particular a recv() timeout:
|
||||
this protocol has no per-message IDs, so there is no way to know whether
|
||||
bytes that show up on the wire *after* a timeout belong to the request that
|
||||
timed out or to whatever is sent next. A stale reply arriving late would
|
||||
otherwise be read as the answer to a new command (or concatenated with it).
|
||||
Closing and reopening the TCP connection discards any such reply-in-flight,
|
||||
so the next command starts from a guaranteed-clean slate.
|
||||
"""
|
||||
try:
|
||||
if self.sock is not None:
|
||||
self.sock.close()
|
||||
except Exception:
|
||||
logger.warning("Error closing socket during reconnect.", exc_info=True)
|
||||
finally:
|
||||
self.sock = None
|
||||
self.connected = False
|
||||
self.on()
|
||||
|
||||
@threadlocked
|
||||
def socket_put(self, val: str):
|
||||
"""
|
||||
Send a command to the controller through the socket.
|
||||
|
||||
Args:
|
||||
val (str): Command to send
|
||||
"""
|
||||
self.command_history.append(f"[PUT]: time:{time.time()}, cmd:{val + self._term}")
|
||||
self.sock.put(f"{val}{self._term}".encode())
|
||||
|
||||
@threadlocked
|
||||
def socket_get(self):
|
||||
"""
|
||||
Receive a single, complete reply from the controller.
|
||||
|
||||
Loops on `recv()` until any one of `self._trail_options` is seen, since a
|
||||
reply can arrive split across multiple TCP reads, and different commands on
|
||||
the same controller (e.g. ACS SETVAR vs GETVAR) can use different terminators.
|
||||
`self._max_reply_length` guards against a malformed/runaway reply with no
|
||||
matching trail. Does not protect against stale replies from a prior
|
||||
timed-out request; that's handled by reconnecting the socket on
|
||||
communication errors (see `_reconnect`).
|
||||
|
||||
Returns:
|
||||
str: The decoded reply, including its trailing terminator.
|
||||
|
||||
Raises:
|
||||
ControllerCommunicationError: If the connection closes while waiting
|
||||
for a reply, or the reply exceeds `self._max_reply_length`.
|
||||
"""
|
||||
buf = b""
|
||||
while True:
|
||||
for trail in self._trail_options:
|
||||
if trail.encode() in buf:
|
||||
response = buf.decode()
|
||||
self.command_history.append(f"[GET]: time:{time.time()}, rep:{response}")
|
||||
return response
|
||||
|
||||
chunk = self.sock.receive()
|
||||
if not chunk:
|
||||
raise ControllerCommunicationError(
|
||||
"Socket connection closed by remote host while waiting for a reply."
|
||||
)
|
||||
buf += chunk
|
||||
if len(buf) > self._max_reply_length:
|
||||
raise ControllerCommunicationError(
|
||||
f"Reply exceeded max_reply_length ({self._max_reply_length} bytes): {buf!r}"
|
||||
)
|
||||
|
||||
@retry_once_reconnected
|
||||
@threadlocked
|
||||
def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str:
|
||||
"""
|
||||
Send a command to the controller and receive the response.
|
||||
Override this method in the derived class if necessary, especially if the response
|
||||
needs to be parsed differently.
|
||||
"""
|
||||
try:
|
||||
self.socket_put(val)
|
||||
if remove_trailing_chars:
|
||||
return self._remove_trailing_characters(self.socket_get())
|
||||
return self.socket_get()
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
f"Error in socket_put_and_receive: {exc}. Command history: {list(self.command_history)}"
|
||||
)
|
||||
raise ControllerCommunicationError(
|
||||
f"Failed to communicate with the controller. The last {self._command_history_length} commands were: "
|
||||
f"{list(self.command_history)}"
|
||||
) from exc
|
||||
|
||||
def _remove_trailing_characters(self, var) -> str:
|
||||
"""Strip whichever configured trail terminator is present at the end of a
|
||||
reply; mid-reply occurrences are kept."""
|
||||
for trail in self._trail_options:
|
||||
if var.endswith(trail):
|
||||
return var.removesuffix(trail)
|
||||
return var
|
||||
|
||||
def on(self, timeout: int = 10) -> None:
|
||||
"""
|
||||
Open a new socket connection to the controller
|
||||
|
||||
Args:
|
||||
timeout (int): Time in seconds to wait for the connection itself to
|
||||
be established (passed to `SocketIO.open`). This is separate from
|
||||
`self._socket_timeout`, which governs how long each individual
|
||||
send/recv call is allowed to take once connected.
|
||||
"""
|
||||
if not self.connected or self.sock is None:
|
||||
try:
|
||||
self.sock = self._socket_cls(
|
||||
host=self._socket_host,
|
||||
port=self._socket_port,
|
||||
socket_timeout=self._socket_timeout,
|
||||
)
|
||||
except TypeError:
|
||||
# socket classes without a socket_timeout parameter (e.g. test mocks)
|
||||
self.sock = self._socket_cls(host=self._socket_host, port=self._socket_port)
|
||||
self.sock.open(timeout=timeout)
|
||||
self.connected = True
|
||||
else:
|
||||
logger.info("The connection has already been established.")
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Tests for the AcsSignal config-read cache and change-only value callbacks."""
|
||||
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from ophyd import Component as Cpt
|
||||
from ophyd import Device
|
||||
|
||||
from debye_bec.devices.mo1_bragg.acs import AcsSignal
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
|
||||
class _AcsDevice(Device):
|
||||
"""Minimal host device providing the controller attribute AcsSignal expects."""
|
||||
|
||||
cached = Cpt(AcsSignal, tag=100, prec=3, kind="config", cache_ttl=30.0)
|
||||
uncached = Cpt(AcsSignal, tag=200, prec=3, kind="config")
|
||||
array = Cpt(AcsSignal, tag=300, prec=3, num_el=3, kind="config", cache_ttl=30.0)
|
||||
|
||||
def __init__(self, *args, controller=None, **kwargs):
|
||||
self.controller = controller
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def acs_device():
|
||||
controller = mock.MagicMock()
|
||||
controller.get_var.return_value = 1.5
|
||||
yield _AcsDevice(name="acs", controller=controller)
|
||||
|
||||
|
||||
def test_cached_get_reads_hardware_once(acs_device):
|
||||
assert acs_device.cached.get() == 1.5
|
||||
assert acs_device.cached.get() == 1.5
|
||||
acs_device.controller.get_var.assert_called_once_with(100, 3)
|
||||
|
||||
|
||||
def test_uncached_get_reads_hardware_every_time(acs_device):
|
||||
acs_device.uncached.get()
|
||||
acs_device.uncached.get()
|
||||
assert acs_device.controller.get_var.call_count == 2
|
||||
|
||||
|
||||
def test_cache_expires_after_ttl(acs_device):
|
||||
acs_device.cached.cache_ttl = 0.05
|
||||
acs_device.cached.get()
|
||||
time.sleep(0.06)
|
||||
acs_device.cached.get()
|
||||
assert acs_device.controller.get_var.call_count == 2
|
||||
|
||||
|
||||
def test_put_invalidates_cache(acs_device):
|
||||
acs_device.cached.get()
|
||||
acs_device.cached.put(2.0)
|
||||
acs_device.controller.set_var.assert_called_once_with(100, 2.0, 3)
|
||||
acs_device.controller.get_var.return_value = 2.0
|
||||
assert acs_device.cached.get() == 2.0
|
||||
assert acs_device.controller.get_var.call_count == 2
|
||||
|
||||
|
||||
def test_invalidate_cache_forces_fresh_read(acs_device):
|
||||
acs_device.cached.get()
|
||||
acs_device.cached.invalidate_cache()
|
||||
acs_device.cached.get()
|
||||
assert acs_device.controller.get_var.call_count == 2
|
||||
|
||||
|
||||
def test_value_callbacks_fire_only_on_change(acs_device):
|
||||
events = []
|
||||
acs_device.cached.subscribe(lambda **kwargs: events.append(kwargs), run=False)
|
||||
|
||||
acs_device.cached.get() # None -> 1.5: fires
|
||||
assert len(events) == 1
|
||||
|
||||
acs_device.cached.invalidate_cache()
|
||||
acs_device.cached.get() # 1.5 -> 1.5: hardware read, but no callback
|
||||
assert acs_device.controller.get_var.call_count == 2
|
||||
assert len(events) == 1
|
||||
|
||||
acs_device.controller.get_var.return_value = 2.5
|
||||
acs_device.cached.invalidate_cache()
|
||||
acs_device.cached.get() # 1.5 -> 2.5: fires
|
||||
assert len(events) == 2
|
||||
assert events[-1]["value"] == 2.5
|
||||
|
||||
|
||||
def test_array_signal_caches_and_compares_by_content(acs_device):
|
||||
acs_device.controller.get_var.side_effect = lambda tag, prec, idx=None: float(idx)
|
||||
|
||||
events = []
|
||||
acs_device.array.subscribe(lambda **kwargs: events.append(kwargs), run=False)
|
||||
|
||||
assert np.array_equal(acs_device.array.get(), np.array([0.0, 1.0, 2.0]))
|
||||
assert acs_device.controller.get_var.call_count == 3 # one call per element
|
||||
assert len(events) == 1
|
||||
|
||||
# cache hit: no additional hardware reads
|
||||
acs_device.array.get()
|
||||
assert acs_device.controller.get_var.call_count == 3
|
||||
|
||||
# fresh read with identical content: no callback
|
||||
acs_device.array.invalidate_cache()
|
||||
acs_device.array.get()
|
||||
assert acs_device.controller.get_var.call_count == 6
|
||||
assert len(events) == 1
|
||||
@@ -8,6 +8,7 @@ from unittest import mock
|
||||
import ophyd
|
||||
import pytest
|
||||
from bec_lib.messages import ScanQueueMessage, ScanStatusMessage
|
||||
from bec_server.device_server.tests.utils import DMMock
|
||||
from bec_server.scan_server.scan_assembler import ScanAssembler
|
||||
from bec_server.scan_server.scan_queue import RequestBlock
|
||||
from bec_server.scan_server.scan_worker import ScanWorker
|
||||
@@ -15,7 +16,7 @@ from bec_server.scan_server.tests.fixtures import scan_server_mock
|
||||
from ophyd.utils import LimitError
|
||||
from ophyd_devices.tests.utils import MockPV
|
||||
|
||||
# from bec_server.device_server.tests.utils import DMMock
|
||||
from debye_bec.devices.mo1_bragg.acs import ACSController
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg import (
|
||||
Mo1Bragg,
|
||||
Mo1BraggError,
|
||||
@@ -40,12 +41,28 @@ def scan_worker_mock(scan_server_mock):
|
||||
def mock_bragg():
|
||||
name = "bragg"
|
||||
prefix = "X01DA-OP-MO1:BRAGG:"
|
||||
with mock.patch.object(ophyd, "cl") as mock_cl:
|
||||
# dict-backed stand-in for the ACS controller so AcsSignal set/get round-trips
|
||||
# work without a socket
|
||||
acs_store = {}
|
||||
|
||||
def _set_var(tag, value, prec, idx=None):
|
||||
acs_store[(tag, idx)] = value
|
||||
|
||||
def _get_var(tag, prec, idx=None):
|
||||
return acs_store.get((tag, idx), 0.0)
|
||||
|
||||
ACSController._reset_controller()
|
||||
with (
|
||||
mock.patch.object(ACSController, "get_var", side_effect=_get_var),
|
||||
mock.patch.object(ACSController, "set_var", side_effect=_set_var),
|
||||
mock.patch.object(ophyd, "cl") as mock_cl,
|
||||
):
|
||||
mock_cl.get_pv = MockPV
|
||||
mock_cl.thread_class = threading.Thread
|
||||
dev = Mo1Bragg(name=name, prefix=prefix)
|
||||
dev = Mo1Bragg(name=name, prefix=prefix, device_manager=DMMock())
|
||||
patch_dual_pvs(dev)
|
||||
yield dev
|
||||
ACSController._reset_controller()
|
||||
|
||||
|
||||
def test_init(mock_bragg):
|
||||
@@ -119,6 +136,34 @@ def test_set_xtal(mock_bragg):
|
||||
assert dev.crystal.xtal_enum.get() == 1
|
||||
|
||||
|
||||
def test_read_configuration_never_touches_acs_and_omits_status_pvs(mock_bragg):
|
||||
"""The device server re-reads the full configuration whenever any auto-monitored
|
||||
signal updates; neither the ticking status PVs nor the ACS socket signals may be
|
||||
part of it, so no passive read path ever produces a GETVAR."""
|
||||
dev = mock_bragg
|
||||
config = dev.read_configuration()
|
||||
# ACS scan settings are omitted: config reads must not touch the socket at all
|
||||
assert "bragg_scan_settings_s_scan_energy_lo" not in config
|
||||
assert "bragg_scan_settings_s_scan_energy_hi" not in config
|
||||
assert "bragg_scan_settings_s_scan_scantime" not in config
|
||||
assert dev.controller.get_var.call_count == 0
|
||||
# live status/progress PVs are omitted from the configuration
|
||||
assert "bragg_status_heartbeat" not in config
|
||||
assert "bragg_scan_control_scan_msg" not in config
|
||||
assert "bragg_scan_control_scan_status" not in config
|
||||
assert "bragg_scan_control_scan_progress" not in config
|
||||
assert "bragg_scan_control_scan_time_left" not in config
|
||||
assert "bragg_scan_control_scan_done" not in config
|
||||
|
||||
# explicit reads still work, are cached, and refresh after a put
|
||||
assert dev.scan_settings.s_scan_energy_lo.get() == 0.0
|
||||
dev.scan_settings.s_scan_energy_lo.get()
|
||||
assert dev.controller.get_var.call_count == 1 # second read served from cache
|
||||
dev.scan_settings.s_scan_energy_lo.put(7000.0)
|
||||
assert dev.scan_settings.s_scan_energy_lo.get() == 7000.0
|
||||
assert dev.controller.get_var.call_count == 2 # put dropped the cache
|
||||
|
||||
|
||||
def test_set_xas_settings(mock_bragg):
|
||||
dev = mock_bragg
|
||||
dev.set_xas_settings(low=0.5, high=1, scan_time=0.1)
|
||||
|
||||
@@ -5,8 +5,10 @@ from unittest import mock
|
||||
|
||||
import ophyd
|
||||
import pytest
|
||||
from bec_server.device_server.tests.utils import DMMock
|
||||
from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
|
||||
|
||||
from debye_bec.devices.mo1_bragg.acs import ACSController
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg_angle import Mo1BraggAngle
|
||||
from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggStoppedError
|
||||
|
||||
@@ -18,12 +20,14 @@ def mock_bragg() -> Mo1BraggAngle:
|
||||
"""Fixture for the Mo1BraggAngle device."""
|
||||
name = "bragg"
|
||||
prefix = "X01DA-OP-MO1:BRAGG:"
|
||||
ACSController._reset_controller()
|
||||
with mock.patch.object(ophyd, "cl") as mock_cl:
|
||||
mock_cl.get_pv = MockPV
|
||||
mock_cl.thread_class = threading.Thread
|
||||
dev = Mo1BraggAngle(name=name, prefix=prefix)
|
||||
dev = Mo1BraggAngle(name=name, prefix=prefix, device_manager=DMMock())
|
||||
patch_dual_pvs(dev)
|
||||
yield dev
|
||||
ACSController._reset_controller()
|
||||
|
||||
|
||||
def test_mo1_bragg_angle_init(mock_bragg):
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Tests for the vendored TermTrailController (term/trail wire-protocol support)."""
|
||||
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_server.device_server.tests.utils import DMMock
|
||||
from ophyd_devices.tests.utils import SocketMock
|
||||
from ophyd_devices.utils.controller import Controller, ControllerCommunicationError
|
||||
|
||||
from debye_bec.devices.utils.term_trail_controller import TermTrailController
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def make_controller():
|
||||
"""Factory fixture to build a fresh controller singleton with custom term/trail settings."""
|
||||
|
||||
def _make(controller_cls=TermTrailController, **kwargs):
|
||||
Controller._reset_controller()
|
||||
controller = controller_cls(
|
||||
name="controller",
|
||||
socket_cls=SocketMock,
|
||||
socket_host="localhost",
|
||||
socket_port=8080,
|
||||
device_manager=DMMock(),
|
||||
**kwargs,
|
||||
)
|
||||
controller.on()
|
||||
return controller
|
||||
|
||||
yield _make
|
||||
Controller._reset_controller()
|
||||
|
||||
|
||||
def test_socket_put_appends_custom_term(make_controller):
|
||||
controller = make_controller(term="\r")
|
||||
controller.socket_put("get")
|
||||
assert controller.sock.buffer_put == [b"get\r"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
["trail", "reply", "expected"],
|
||||
[
|
||||
(None, "value\r\n", "value"),
|
||||
(None, "\r\n", ""),
|
||||
(None, "line1\r\nline2\r\n", "line1\r\nline2"),
|
||||
("\n", "value\n", "value"),
|
||||
("\n", "\n", ""),
|
||||
("\n", "line1\nline2\n", "line1\nline2"),
|
||||
("", "value\r\n", "value\r\n"),
|
||||
],
|
||||
)
|
||||
def test_remove_trailing_characters_strips_suffix_only(make_controller, trail, reply, expected):
|
||||
"""The trail is stripped from the end of a reply only; mid-reply occurrences are preserved,
|
||||
a bare terminator reduces to an empty payload, and an empty trail strips nothing."""
|
||||
kwargs = {} if trail is None else {"trail": trail}
|
||||
controller = make_controller(**kwargs)
|
||||
assert controller._remove_trailing_characters(reply) == expected
|
||||
|
||||
|
||||
def test_multiple_trail_options_strip_whichever_matches(make_controller):
|
||||
"""With several trail alternatives (e.g. ACS GETVAR vs SETVAR replies), the one
|
||||
that terminates the reply is stripped."""
|
||||
controller = make_controller(term="\r", trail=["\r:\r", ":\r"])
|
||||
controller.sock.buffer_recv = [b"8000.5\r:\r"]
|
||||
assert controller.socket_put_and_receive("?GETVAR(1)") == "8000.5"
|
||||
controller.sock.buffer_recv = [b":\r"]
|
||||
assert controller.socket_put_and_receive("SETVAR(1,1)") == ""
|
||||
|
||||
|
||||
def test_socket_get_reassembles_chunked_reply(make_controller):
|
||||
controller = make_controller()
|
||||
controller.sock.buffer_recv = [b"val", b"ue\r", b"\n"]
|
||||
assert controller.socket_put_and_receive("get") == "value"
|
||||
|
||||
|
||||
def test_socket_get_raises_on_runaway_reply(make_controller):
|
||||
controller = make_controller(max_reply_length=8)
|
||||
controller.sock.buffer_recv = [b"0123456789abcdef"]
|
||||
with pytest.raises(ControllerCommunicationError):
|
||||
controller.socket_put_and_receive("get")
|
||||
|
||||
|
||||
def test_put_and_receive_reconnects_and_retries(make_controller):
|
||||
"""A communication error triggers exactly one reconnect, then the retry succeeds."""
|
||||
controller = make_controller()
|
||||
controller.sock.buffer_recv = [b""] # remote closed the connection
|
||||
|
||||
def _restock():
|
||||
controller.sock.buffer_recv = [b"value\r\n"]
|
||||
|
||||
with mock.patch.object(controller, "_reconnect", side_effect=_restock) as mock_reconnect:
|
||||
assert controller.socket_put_and_receive("get") == "value"
|
||||
mock_reconnect.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"kwargs", [{"term": b"\n"}, {"trail": b"\r\n"}, {"term": 13}, {"trail": 0}]
|
||||
)
|
||||
def test_term_and_trail_must_be_strings(make_controller, kwargs):
|
||||
with pytest.raises(TypeError):
|
||||
make_controller(**kwargs)
|
||||
|
||||
|
||||
def test_second_construction_with_conflicting_term_trail_warns(make_controller):
|
||||
"""A second construction for the same host:port keeps the first term/trail and warns."""
|
||||
controller = make_controller(term="\r", trail="\n")
|
||||
with mock.patch("debye_bec.devices.utils.term_trail_controller.logger") as mock_logger:
|
||||
second = TermTrailController(
|
||||
name="controller",
|
||||
socket_cls=SocketMock,
|
||||
socket_host="localhost",
|
||||
socket_port=8080,
|
||||
device_manager=DMMock(),
|
||||
term="\n",
|
||||
)
|
||||
assert second is controller
|
||||
mock_logger.warning.assert_called_once()
|
||||
assert controller._term == "\r"
|
||||
assert controller._trail == "\n"
|
||||
|
||||
|
||||
def test_second_construction_without_conflict_does_not_warn(make_controller):
|
||||
controller = make_controller(term="\r")
|
||||
with mock.patch("debye_bec.devices.utils.term_trail_controller.logger") as mock_logger:
|
||||
for kwargs in ({"term": "\r"}, {}):
|
||||
second = TermTrailController(
|
||||
name="controller",
|
||||
socket_cls=SocketMock,
|
||||
socket_host="localhost",
|
||||
socket_port=8080,
|
||||
device_manager=DMMock(),
|
||||
**kwargs,
|
||||
)
|
||||
assert second is controller
|
||||
mock_logger.warning.assert_not_called()
|
||||
|
||||
|
||||
def test_subclass_overrides_term_trail_as_class_attributes(make_controller):
|
||||
"""Subclasses can set the wire protocol with class attributes, without any constructor plumbing."""
|
||||
|
||||
class CRTermController(TermTrailController):
|
||||
_term = "\r"
|
||||
_trail = "\n"
|
||||
|
||||
controller = make_controller(controller_cls=CRTermController)
|
||||
controller.sock.buffer_recv = [b"value\n"]
|
||||
assert controller.socket_put_and_receive("get") == "value"
|
||||
assert controller.sock.buffer_put == [b"get\r"]
|
||||
Reference in New Issue
Block a user