"""Static PVs for KDAECTRL."""
import logging
import time
from queue import PriorityQueue
from p4p.nt import NTEnum, NTScalar
from p4p.server import ServerOperation, StaticProvider
from p4p.server.thread import SharedPV
from kafka_dae_control.data import Data
from kafka_dae_control.defaults import NOT_SETTABLE_PERIOD_MODES, FrameSyncSelect, PeriodMode
from kafka_dae_control.event_with_error import EventWithError
from kafka_dae_control.queue_utils import QueueItem, QueuePriority
from kafka_dae_control.worker_event_types import (
BeginEvent,
CancelAbortEvent,
CurrentPeriodSetEvent,
EndEvent,
FrameSyncDelaySetpointEvent,
FrameSyncSelectChangeEvent,
NumberOfPeriodsSetEvent,
PauseResumeEvent,
PeriodModeSetEvent,
VetoesUpdateEvent,
)
logger = logging.getLogger(__name__)
# Event time-of-flights must fit into an i32 number of nanoseconds
# That puts the upper limit of ToF at 2.147 seconds.
# Put the limit here at 2 seconds, to also account for an 100ms frame length.
MAX_FRAME_SYNC_DELAY_US = 2000000
[docs]
class StaticPVs:
"""Static PVs for KDAECTRL."""
def __init__(self, data: "Data", queue: PriorityQueue[QueueItem]) -> None: # ruff:ignore[too-many-statements]
"""Set up static PVs for KDAECTRL.
Args:
data: the data class containing the state of the program.
queue: the worker event queue.
"""
self.hw_running = SharedPV(
nt=NTScalar(display=True, form=True),
initial={
"value": data.running,
},
)
self.frame_sync_select_rbv = SharedPV(
nt=NTEnum(),
initial={
"choices": [x.name for x in FrameSyncSelect],
"index": data.frame_sync_select_rbv,
},
)
self.frame_sync_select_sp = SharedPV(
nt=NTEnum(),
initial={
"choices": [x.name for x in FrameSyncSelect if x != FrameSyncSelect.UNKNOWN],
"index": data.frame_sync_select_sp,
},
)
self.begin_run_ex = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.begin_run = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.end = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.pause = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.resume = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.abort = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.cancel_abort = SharedPV(nt=NTScalar(display=True, form=True), initial={"value": False})
self.run_number = SharedPV(
nt=NTScalar("s", display=True, form=True), initial={"value": str(data.run_number)}
)
self.i_run_number = SharedPV(
nt=NTScalar(display=True, form=True), initial={"value": data.run_number}
)
self.veto_names_array = SharedPV(
nt=NTScalar("as", display=True, form=True),
initial={
"value": data.veto_names_array,
},
)
self.vetoes = SharedPV(
nt=NTScalar("al", display=True, form=True),
initial={
"value": data.vetoes,
"display.units": "",
"display.precision": 0,
},
)
self.hard_vetoes_rbv = SharedPV(
nt=NTScalar("l", display=True, form=True),
initial={
"value": data.hard_vetoes_rbv,
"display.units": "",
"display.precision": 0,
},
)
self.num_periods_sp = SharedPV(
nt=NTScalar(display=True, form=True), initial={"value": data.num_periods_sp}
)
self.num_periods_rbv = SharedPV(
nt=NTScalar(display=True, form=True), initial={"value": data.num_periods_sp}
)
self.period_sp = SharedPV(
nt=NTScalar(display=True, form=True), initial={"value": data.current_period_sp}
)
self.period_rbv = SharedPV(
nt=NTScalar(display=True, form=True), initial={"value": data.current_period_sp}
)
self.period_type_rbv = SharedPV(
nt=NTEnum(),
initial={
"choices": [x.name for x in PeriodMode],
"index": data.period_mode_rbv,
},
)
self.period_type_sp = SharedPV(
nt=NTEnum(),
initial={
"choices": [x.name for x in PeriodMode if x not in NOT_SETTABLE_PERIOD_MODES],
"index": data.period_mode_sp,
},
)
self.frame_sync_delay_rbv = SharedPV(
nt=NTScalar(display=True, form=True),
initial={
"value": data.frame_sync_delay_rbv,
"display.units": "us",
"display.precision": 0,
},
)
self.frame_sync_delay_sp = SharedPV(
nt=NTScalar(display=True, form=True),
initial={
"value": data.frame_sync_delay_rbv,
"display.units": "us",
"display.precision": 0,
},
)
@self.begin_run_ex.put # pragma: no cover
def begin_run_ex_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("beginrunex with value %s", value)
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, BeginEvent(value=value, done_event=ev)))
try:
ev.wait()
pv.post(0, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to begin: {e}")
@self.begin_run.put # pragma: no cover
def begin_run_put(pv: SharedPV, op: ServerOperation) -> None:
logger.info("beginrun")
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, BeginEvent(value=False, done_event=ev)))
try:
ev.wait()
pv.post(0, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to begin: {e}")
@self.end.put # pragma: no cover
def end_put(pv: SharedPV, op: ServerOperation) -> None:
logger.info("end")
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, EndEvent(value=False, done_event=ev)))
try:
ev.wait()
pv.post(0, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to end: {e}")
@self.abort.put
def abort_put(pv: SharedPV, op: ServerOperation) -> None:
logger.info("abort")
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, EndEvent(value=True, done_event=ev)))
try:
ev.wait()
pv.post(0)
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to abort: {e}")
@self.cancel_abort.put
def cancel_abort_put(pv: SharedPV, op: ServerOperation) -> None:
logger.info("cancel abort")
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, CancelAbortEvent(done_event=ev)))
try:
ev.wait()
pv.post(0)
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to cancel abort: {e}")
@self.frame_sync_select_sp.put
def frame_sync_select_sp_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("put with %s to frame_sync_select_sp", value)
ev = EventWithError()
queue.put(
QueueItem(
QueuePriority.HIGH,
FrameSyncSelectChangeEvent(value=FrameSyncSelect[str(value)], done_event=ev),
)
)
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set frame_sync_select_sp: {e}")
@self.vetoes.put
def vetoes_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("put with %s to vetoes sp", value)
ev = EventWithError()
queue.put(QueueItem(QueuePriority.HIGH, VetoesUpdateEvent(value=value, done_event=ev)))
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set soft_vetoes: {e}")
@self.num_periods_sp.put
def num_periods_sp_put(pv: SharedPV, op: ServerOperation) -> None:
value = int(op.value())
logger.info("put with %s to num_periods_sp", value)
ev = EventWithError()
queue.put(
QueueItem(QueuePriority.HIGH, NumberOfPeriodsSetEvent(value=value, done_event=ev))
)
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set num_periods_sp: {e}")
@self.period_sp.put
def period_sp_put(pv: SharedPV, op: ServerOperation) -> None:
value = int(op.value())
logger.info("put with %s to period_sp", value)
if value < 1:
op.done(error="Period must be greater than 0")
return
ev = EventWithError()
queue.put(
QueueItem(QueuePriority.HIGH, CurrentPeriodSetEvent(value=value, done_event=ev))
)
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set period_sp: {e}")
@self.period_type_sp.put
def period_type_sp_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("put with %s to period_sp", value)
ev = EventWithError()
queue.put(
QueueItem(
QueuePriority.HIGH,
PeriodModeSetEvent(value=PeriodMode[str(value)], done_event=ev),
)
)
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set period_sp: {e}")
@self.pause.put
def pause_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("put with %s to pause", value)
ev = EventWithError()
queue.put(
QueueItem(
QueuePriority.HIGH,
PauseResumeEvent(value=True, done_event=ev),
)
)
try:
ev.wait()
pv.post(0, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to pause: {e}")
@self.resume.put
def resume_put(pv: SharedPV, op: ServerOperation) -> None:
value = op.value()
logger.info("put with %s to resume", value)
ev = EventWithError()
queue.put(
QueueItem(
QueuePriority.HIGH,
PauseResumeEvent(value=False, done_event=ev),
)
)
try:
ev.wait()
pv.post(0, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to resume: {e}")
@self.frame_sync_delay_sp.put
def frame_sync_delay_sp_put(pv: SharedPV, op: ServerOperation) -> None:
value = int(op.value())
logger.info("put with %s to frame_sync_delay_sp", value)
if value < 0:
# The hardware won't accept a negative delay number, and even if it did
# it would be meaningless...
logger.warning(
"Invalid frame_sync_delay value %s attempted; cannot delay by "
"a negative amount. Not setting frame_sync_delay.",
value,
)
op.done(error="frame_sync_delay must be greater or equal to 0")
return
elif value > MAX_FRAME_SYNC_DELAY_US:
logger.warning(
"Invalid frame_sync_delay value %s attempted; may cause event times "
"to exceed i32 number of nanoseconds. Not setting frame_sync_delay.",
value,
)
op.done(error="frame_sync_delay must be less than or equal to 200000 us")
return
ev = EventWithError()
queue.put(
QueueItem(
QueuePriority.HIGH,
FrameSyncDelaySetpointEvent(value=value, done_event=ev),
)
)
try:
ev.wait()
pv.post(value, timestamp=time.time())
op.done()
except Exception as e: # ruff:ignore[blind-except]
op.done(error=f"Failed to set frame sync delay: {e}")
[docs]
def update_all(self, data: Data) -> None:
"""Post updates to all PVs using the data class values.
Args:
data: the data class containing the state of the program.
"""
kwargs = {
"timestamp": time.time(),
"severity": 0 if data.last_hardware_update_ok else 3,
"message": "NO_ALARM" if data.last_hardware_update_ok else "COMM_ALARM",
}
self.run_number.post(str(data.run_number), **kwargs)
self.i_run_number.post(data.run_number, **kwargs)
self.hw_running.post(data.running, **kwargs)
self.frame_sync_select_rbv.post(data.frame_sync_select_rbv.name, **kwargs)
self.hard_vetoes_rbv.post(data.hard_vetoes_rbv, **kwargs)
self.num_periods_rbv.post(data.num_periods_rbv, **kwargs)
self.period_rbv.post(data.current_period_rbv, **kwargs)
self.period_type_rbv.post(data.period_mode_rbv.name, **kwargs)
self.frame_sync_delay_rbv.post(data.frame_sync_delay_rbv, **kwargs)
[docs]
def static_pv_provider(
pv_prefix: str, data: "Data", queue: PriorityQueue[QueueItem]
) -> tuple[StaticPVs, StaticProvider]:
"""Generate a static pv provider containing all the static PVs.
This also sets up basic post hooks for observable dataclass items.
Args:
pv_prefix: the PV prefix.
data: The data class containing the state of the program.
queue: the worker event queue.
Returns: A static pv provider containing static PVs.
"""
static_pvs = StaticPVs(data, queue)
static_provider = StaticProvider()
static_provider.add(f"{pv_prefix}HWRUNNING", static_pvs.hw_running)
static_provider.add(f"{pv_prefix}BEGINRUNEX", static_pvs.begin_run_ex)
static_provider.add(f"{pv_prefix}BEGINRUN", static_pvs.begin_run)
static_provider.add(f"{pv_prefix}ENDRUN", static_pvs.end)
static_provider.add(f"{pv_prefix}PAUSERUN", static_pvs.pause)
static_provider.add(f"{pv_prefix}RESUMERUN", static_pvs.resume)
static_provider.add(f"{pv_prefix}ABORTRUN", static_pvs.abort)
static_provider.add(f"{pv_prefix}RECOVERRUN", static_pvs.cancel_abort)
static_provider.add(f"{pv_prefix}RUNNUMBER", static_pvs.run_number)
static_provider.add(f"{pv_prefix}IRUNNUMBER", static_pvs.i_run_number)
static_provider.add(f"{pv_prefix}DAETIMINGSOURCE", static_pvs.frame_sync_select_rbv)
static_provider.add(f"{pv_prefix}DAETIMINGSOURCE:SP", static_pvs.frame_sync_select_sp)
static_provider.add(f"{pv_prefix}VETO:NAMES", static_pvs.veto_names_array)
static_provider.add(f"{pv_prefix}VETO:SP", static_pvs.vetoes)
static_provider.add(f"{pv_prefix}VETO:HARD", static_pvs.hard_vetoes_rbv)
static_provider.add(f"{pv_prefix}NUMPERIODS:MAX", static_pvs.num_periods_sp)
static_provider.add(f"{pv_prefix}NUMPERIODS:SP", static_pvs.num_periods_sp)
static_provider.add(f"{pv_prefix}NUMPERIODS", static_pvs.num_periods_rbv)
static_provider.add(f"{pv_prefix}PERIOD", static_pvs.period_rbv)
static_provider.add(f"{pv_prefix}PERIOD:SP", static_pvs.period_sp)
static_provider.add(f"{pv_prefix}PERIODTYPE", static_pvs.period_type_rbv)
static_provider.add(f"{pv_prefix}PERIODTYPE:SP", static_pvs.period_type_sp)
static_provider.add(f"{pv_prefix}FRAME_SYNC_DELAY", static_pvs.frame_sync_delay_rbv)
static_provider.add(f"{pv_prefix}FRAME_SYNC_DELAY:SP", static_pvs.frame_sync_delay_sp)
return static_pvs, static_provider