wip socket signals
CI for debye_bec / test (pull_request) Failing after 1m58s
CI for debye_bec / test (push) Failing after 3m19s

This commit is contained in:
x01da
2026-08-05 10:09:25 +02:00
parent 32325a9433
commit ff85ff08e0
2 changed files with 31 additions and 15 deletions
+13 -1
View File
@@ -13,6 +13,8 @@ Protocol:
from __future__ import annotations
import time
import traceback
from enum import Enum
import numpy as np
@@ -43,6 +45,7 @@ class ACSController(Controller):
device_manager=device_manager,
term="\r",
trail=["\r:\r", ":\r"],
socket_timeout=0.1,
)
@threadlocked
@@ -85,6 +88,7 @@ class AcsSignal(SocketSignal):
self.prec = prec
self.num_el = num_el
self.enum = enum
self.last_get = time.time()
super().__init__(*args, **kwargs)
@property
@@ -92,11 +96,19 @@ class AcsSignal(SocketSignal):
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:
return convert(self.controller.get_var(self.tag, self.prec))
val = convert(self.controller.get_var(self.tag, self.prec))
logger.info(f"Get signal with tag {self.tag}, 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)]
)
@@ -45,9 +45,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())
@@ -112,14 +112,14 @@ class Mo1BraggCrystal(Device):
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
# )
current_xtal_string = Cpt(
AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal", auto_monitor=True
EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True
)
# current_xtal_string = Cpt(
# AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal", auto_monitor=True
# )
class Mo1BraggScanSettings(Device):
"""Mo1 Bragg PVs to set the scan setttings"""
@@ -148,14 +148,18 @@ class Mo1BraggScanSettings(Device):
# EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True
# )
s_scan_energy_lo = Cpt(AcsSignal, tag=53003, prec=6, kind="config", auto_monitor=True)
s_scan_energy_hi = Cpt(AcsSignal, tag=53004, prec=6, kind="config", auto_monitor=True)
s_scan_scantime = Cpt(AcsSignal, tag=53002, prec=3, kind="config", auto_monitor=True)
s_scan_energy_lo = Cpt(AcsSignal, tag=53003, prec=6, kind="config", auto_monitor=False)
s_scan_energy_hi = Cpt(AcsSignal, tag=53004, prec=6, kind="config", auto_monitor=False)
s_scan_scantime = Cpt(AcsSignal, tag=53002, prec=3, kind="config", auto_monitor=False)
# 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)
a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False)
a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=False)
a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=False)
# a_scan_pos = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="omitted")
# a_scan_vel = Cpt(AcsSignal, tag=53501, prec=6, num_el=41, kind="omitted")
# a_scan_time = Cpt(AcsSignal, tag=53502, prec=6, num_el=41, kind="omitted")
class Mo1TriggerSettings(Device):