From 3cde8f350eab8e734960dfcd8a684c087c9e67e1 Mon Sep 17 00:00:00 2001 From: appel_c Date: Wed, 11 Feb 2026 15:06:00 +0100 Subject: [PATCH] refactor(eps): Refactoring of EPS device from cSAXS --- csaxs_bec/devices/epics/eps.py | 1182 ++++++++++++++++++++++++-------- 1 file changed, 880 insertions(+), 302 deletions(-) diff --git a/csaxs_bec/devices/epics/eps.py b/csaxs_bec/devices/epics/eps.py index 071b258..499c98d 100644 --- a/csaxs_bec/devices/epics/eps.py +++ b/csaxs_bec/devices/epics/eps.py @@ -1,151 +1,887 @@ +"""EPS module for cSAXS beamline: defines the EPS device with its components and methods.""" + +from __future__ import annotations -from ophyd import Device, Component as Cpt, EpicsSignal import time +from bec_lib.logger import bec_logger +from ophyd import Component as Cpt +from ophyd import Device, EpicsSignal, EpicsSignalRO, Kind +from ophyd_devices import PSIDeviceBase + +logger = bec_logger.logger + # --------------------------- # Registry: sections/channels # --------------------------- -CHANNELS = { - "EPS Alarms": [ - { - "attr": "EPSAlarmCnt", - "label": "X12SA EPS Alarm count", - "pv": "X12SA-EPS-PLC:AlarmCnt_EPS", - "kind": "alarmcount", - }, - { - "attr": "MISAlarmCnt", - "label": "FrontEnd MIS Alarm count", - "pv": "ARS00-MIS-PLC-01:AlarmCnt_Frontends", - "kind": "alarmcount", - }, - ], - "Valves Frontend": [ - {"attr": "FEVVPG0000", "label": "FE-VVPG-0000", - "pv": "X12SA-FE-VVPG-0000:PLC_OPEN", "kind": "valve"}, - {"attr": "FEVVPG1010", "label": "FE-VVPG-1010", - "pv": "X12SA-FE-VVPG-1010:PLC_OPEN", "kind": "valve"}, - {"attr": "FEVVFV2010", "label": "FE-VVFV-2010", - "pv": "X12SA-FE-VVFV-2010:PLC_OPEN", "kind": "valve"}, - {"attr": "FEVVPG2010", "label": "FE-VVPG-2010", - "pv": "X12SA-FE-VVPG-2010:PLC_OPEN", "kind": "valve"}, - ], +class EPSSubDevices(Device): + """Base class for EPS sub-device components (e.g. alarms, valves, shutters). with common methods if needed.""" - "Valves Optics Hutch": [ - {"attr": "OPVVPG1010", "label": "OP-VVPG-1010", - "pv": "X12SA-OP-VVPG-1010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG2010", "label": "OP-VVPG-2010", - "pv": "X12SA-OP-VVPG-2010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG3010", "label": "OP-VVPG-3010", - "pv": "X12SA-OP-VVPG-3010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG3020", "label": "OP-VVPG-3020", - "pv": "X12SA-OP-VVPG-3020:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG4010", "label": "OP-VVPG-4010", - "pv": "X12SA-OP-VVPG-4010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG5010", "label": "OP-VVPG-5010", - "pv": "X12SA-OP-VVPG-5010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG6010", "label": "OP-VVPG-6010", - "pv": "X12SA-OP-VVPG-6010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPVVPG7010", "label": "OP-VVPG-7010", - "pv": "X12SA-OP-VVPG-7010:PLC_OPEN", "kind": "valve"}, - ], + def describe(self) -> dict: + desc = super().describe() + for walk in self.walk_signals(): + if walk.item.attr_name not in desc: + desc[walk.item.attr_name] = walk.item.describe() + return desc - "Valves ES Hutch": [ - {"attr": "ESVVPG1010", "label": "ES-VVPG-1010", - "pv": "X12SA-ES-VVPG-1010:PLC_OPEN", "kind": "valve"}, - ], - "Shutters Frontend": [ - {"attr": "FEPSH1", "label": "FE-PSH1-EMLS-0010", - "pv": "X12SA-FE-PSH1-EMLS-0010:OPEN", "kind": "shutter"}, - {"attr": "FESTO1", "label": "FE-STO1-EMLS-0010", - "pv": "X12SA-FE-STO1-EMLS-0010:OPEN", "kind": "shutter"}, - ], +class EPSAlarms(EPSSubDevices): + """EPS alarms at the cSAXS beamline.""" - "Shutters Endstation": [ - {"attr": "ESPSH17010", "label": "OP-PSH1-EMLS-7010", - "pv": "X12SA-OP-PSH1-EMLS-7010:OPEN", "kind": "shutter"}, - ], + eps_alarm_cnt = Cpt( + EpicsSignalRO, + read_pv="X12SA-EPS-PLC:AlarmCnt_EPS", + add_prefix=("",), + name="eps_alarm_cnt", + kind=Kind.omitted, + doc="X12SA EPS Alarm count", + auto_monitor=True, + labels={"alarm"}, + ) + mis_alarm_cnt = Cpt( + EpicsSignalRO, + read_pv="ARS00-MIS-PLC-01:AlarmCnt_Frontends", + add_prefix=("",), + name="mis_alarm_cnt", + kind=Kind.omitted, + doc="FrontEnd MIS Alarm count", + auto_monitor=True, + labels={"alarm"}, + ) - "DMM Monochromator": [ - {"attr": "DMM_ETTC_3010", "label": "DMM Temp Surface 1", - "pv": "X12SA-OP-DMM-ETTC-3010:TEMP", "kind": "temp"}, - {"attr": "DMM_ETTC_3020", "label": "DMM Temp Surface 2", - "pv": "X12SA-OP-DMM-ETTC-3020:TEMP", "kind": "temp"}, - {"attr": "DMM_ETTC_3030", "label": "DMM Temp Shield 1 (disaster)", - "pv": "X12SA-OP-DMM-ETTC-3030:TEMP", "kind": "temp"}, - {"attr": "DMM_ETTC_3040", "label": "DMM Temp Shield 2 (disaster)", - "pv": "X12SA-OP-DMM-ETTC-3040:TEMP", "kind": "temp"}, - {"attr": "DMM_EMLS_3010", "label": "DMM Translation ThruPos", - "pv": "X12SA-OP-DMM-EMLS-3010:THRU", "kind": "switch"}, - {"attr": "DMM_EMLS_3020", "label": "DMM Translation InPos", - "pv": "X12SA-OP-DMM-EMLS-3020:IN", "kind": "switch"}, - {"attr": "DMM_EMLS_3030", "label": "DMM Bragg ThruPos", - "pv": "X12SA-OP-DMM-EMLS-3030:THRU", "kind": "switch"}, - {"attr": "DMM_EMLS_3040", "label": "DMM Bragg InPos", - "pv": "X12SA-OP-DMM-EMLS-3040:IN", "kind": "switch"}, +class Valves(EPSSubDevices): + """Valves at the cSAXS beamline.""" - {"attr": "DMM_EMSW_3050_SWITCH", "label": "DMM Heater Fault XTAL 1", - "pv": "X12SA-OP-DMM-EMSW-3050:SWITCH", "kind": "fault"}, - {"attr": "DMM_EMSW_3060_SWITCH", "label": "DMM Heater Fault XTAL 2", - "pv": "X12SA-OP-DMM-EMSW-3060:SWITCH", "kind": "fault"}, - {"attr": "DMM_EMSW_3070_SWITCH", "label": "DMM Heater Fault Support 1", - "pv": "X12SA-OP-DMM-EMSW-3070:SWITCH", "kind": "fault"}, + ################ + ### Frontend ### + ################ + fe_vvpg_0000 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-VVPG-0000:PLC_OPEN", + add_prefix=("",), + name="fevvpg0000", + kind=Kind.omitted, + doc="Frontend valve FE-VVPG-0000", + auto_monitor=True, + labels={"valve"}, + ) + fe_vvpg_1010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-VVPG-1010:PLC_OPEN", + add_prefix=("",), + name="fevvpg1010", + kind=Kind.omitted, + doc="Frontend valve FE-VVPG-1010", + auto_monitor=True, + labels={"valve"}, + ) + fe_vvfv_2010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-VVFV-2010:PLC_OPEN", + add_prefix=("",), + name="fevvfv2010", + kind=Kind.omitted, + doc="Frontend valve FE-VVFV-2010", + auto_monitor=True, + labels={"valve"}, + ) + fe_vvpg_2010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-VVPG-2010:PLC_OPEN", + add_prefix=("",), + name="fevvpg2010", + kind=Kind.omitted, + doc="Frontend valve FE-VVPG-2010", + auto_monitor=True, + labels={"valve"}, + ) - {"attr": "DMM1_ENERGY_GET", "label": "DMM Energy", - "pv": "X12SA-OP-DMM1:ENERGY-GET", "kind": "energy"}, - {"attr": "DMM1_POSITION", "label": "DMM Position", - "pv": "X12SA-OP-DMM1:POSITION", "kind": "string"}, - {"attr": "DMM1_STRIPE", "label": "DMM Stripe", - "pv": "X12SA-OP-DMM1:STRIPE", "kind": "string"}, - ], + ################ + ### Optics ### + ################ - "CCM Monochromator": [ - {"attr": "CCM_ETTC_4010", "label": "CCM Temp Crystal", - "pv": "X12SA-OP-CCM-ETTC-4010:TEMP", "kind": "temp"}, - {"attr": "CCM_ETTC_4020", "label": "CCM Temp Shield (disaster)", - "pv": "X12SA-OP-CCM-ETTC-4020:TEMP", "kind": "temp"}, + op_vvpg_1010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-1010:PLC_OPEN", + add_prefix=("",), + name="opvvpg1010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-1010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_2010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-2010:PLC_OPEN", + add_prefix=("",), + name="opvvpg2010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-2010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_3010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-3010:PLC_OPEN", + add_prefix=("",), + name="opvvpg3010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-3010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_3020 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-3020:PLC_OPEN", + add_prefix=("",), + name="opvvpg3020", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-3020", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_4010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-4010:PLC_OPEN", + add_prefix=("",), + name="opvvpg4010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-4010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_5010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-5010:PLC_OPEN", + add_prefix=("",), + name="opvvpg5010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-5010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_6010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-6010:PLC_OPEN", + add_prefix=("",), + name="opvvpg6010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-6010", + auto_monitor=True, + labels={"valve"}, + ) + op_vvpg_7010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-VVPG-7010:PLC_OPEN", + add_prefix=("",), + name="opvvpg7010", + kind=Kind.omitted, + doc="Optics valve OP-VVPG-7010", + auto_monitor=True, + labels={"valve"}, + ) - {"attr": "CCM_EMSW_4010_SWITCH", "label": "CCM Heater Fault 1", - "pv": "X12SA-OP-CCM-EMSW-4010:SWITCH", "kind": "fault"}, - {"attr": "CCM_EMSW_4020_SWITCH", "label": "CCM Heater Fault 2", - "pv": "X12SA-OP-CCM-EMSW-4020:SWITCH", "kind": "fault"}, - {"attr": "CCM_EMSW_4030_SWITCH", "label": "CCM Heater Fault 3", - "pv": "X12SA-OP-CCM-EMSW-4030:SWITCH", "kind": "fault"}, + ################## + ### Endstation ### + ################## - {"attr": "CCM1_ENERGY_GET", "label": "CCM Energy", - "pv": "X12SA-OP-CCM1:ENERGY-GET", "kind": "energy"}, - {"attr": "CCM1_POSITION", "label": "CCM Position", - "pv": "X12SA-OP-CCM1:POSITION", "kind": "string"}, - ], + es_vvpg_1010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-ES-VVPG-1010:PLC_OPEN", + add_prefix=("",), + name="esvvpg1010", + kind=Kind.omitted, + doc="Endstation valve ES-VVPG-1010", + auto_monitor=True, + labels={"valve"}, + ) - "Cooling Water": [ - {"attr": "OPSL1EFSW2010", "label": "OP-SL1-EFSW-2010", - "pv": "X12SA-OP-SL1-EFSW-2010:FLOW", "kind": "flow"}, - {"attr": "OPSL2EFSW2010", "label": "OP-SL2-EFSW-2010", - "pv": "X12SA-OP-SL2-EFSW-2010:FLOW", "kind": "flow"}, - {"attr": "OPEB1EFSW5010", "label": "OP-EB1-EFSW-5010", - "pv": "X12SA-OP-EB1-EFSW-5010:FLOW", "kind": "flow"}, - {"attr": "OPEB1EFSW5020", "label": "OP-EB1-EFSW-5020", - "pv": "X12SA-OP-EB1-EFSW-5020:FLOW", "kind": "flow"}, - {"attr": "OPSL3EFSW5010", "label": "OP-SL3-EFSW-5010", - "pv": "X12SA-OP-SL3-EFSW-5010:FLOW", "kind": "flow"}, - {"attr": "OPKBEFSW6010", "label": "OP-KB-EFSW-6010", - "pv": "X12SA-OP-KB-EFSW-6010:FLOW", "kind": "flow"}, - {"attr": "OPPSH1EFSW7010", "label": "OP-PSH1-EFSW-7010", - "pv": "X12SA-OP-PSH1-EFSW-7010:FLOW", "kind": "flow"}, - {"attr": "ESEB2EFSW1010", "label": "ES-EB2-EFSW-1010", - "pv": "X12SA-ES-EB2-EFSW-1010:FLOW", "kind": "flow"}, - {"attr": "OPCSECVW0010", "label": "OP-CS-ECVW-0010", - "pv": "X12SA-OP-CS-ECVW-0010:PLC_OPEN", "kind": "valve"}, - {"attr": "OPCSECVW0020", "label": "OP-CS-ECVW-0020", - "pv": "X12SA-OP-CS-ECVW-0020:PLC_OPEN", "kind": "valve"}, - ], -} +class Shutters(EPSSubDevices): + """Shutters at the cSAXS beamline.""" + + fe_psh1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-PSH1-EMLS-0010:OPEN", + add_prefix=("",), + name="fepsh1", + kind=Kind.omitted, + doc="Frontend shutter FE-PSH1-EMLS-0010", + auto_monitor=True, + labels={"shutter"}, + ) + fe_sto1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-FE-STO1-EMLS-0010:OPEN", + add_prefix=("",), + name="festo1", + kind=Kind.omitted, + doc="Frontend shutter FE-STO1-EMLS-0010", + auto_monitor=True, + labels={"shutter"}, + ) + + es_psh17010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-PSH1-EMLS-7010:OPEN", + add_prefix=("",), + name="espsh17010", + kind=Kind.omitted, + doc="Endstation shutter OP-PSH1-EMLS-7010", + auto_monitor=True, + labels={"shutter"}, + ) + + +class DMMMonochromator(EPSSubDevices): + """DMM monochromator signals at the cSAXS beamline.""" + + dmm_temp_surface_1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-ETTC-3010:TEMP", + add_prefix=("",), + name="dmm_temp_surface_1", + kind=Kind.omitted, + doc="DMM Temp Surface 1", + auto_monitor=True, + labels={"temp"}, + ) + dmm_temp_surface_2 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-ETTC-3020:TEMP", + add_prefix=("",), + name="dmm_temp_surface_2", + kind=Kind.omitted, + doc="DMM Temp Surface 2", + auto_monitor=True, + labels={"temp"}, + ) + dmm_temp_shield_1_disaster = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-ETTC-3030:TEMP", + add_prefix=("",), + name="dmm_temp_shield_1_disaster", + kind=Kind.omitted, + doc="DMM Temp Shield 1 (disaster)", + auto_monitor=True, + labels={"temp"}, + ) + dmm_temp_shield_2_disaster = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-ETTC-3040:TEMP", + add_prefix=("",), + name="dmm_temp_shield_2_disaster", + kind=Kind.omitted, + doc="DMM Temp Shield 2 (disaster)", + auto_monitor=True, + labels={"temp"}, + ) + + dmm_translation_thru = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMLS-3010:THRU", + add_prefix=("",), + name="dmm_translation_thru", + kind=Kind.omitted, + doc="DMM Translation ThruPos", + auto_monitor=True, + labels={"switch"}, + ) + dmm_translation_in = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMLS-3020:IN", + add_prefix=("",), + name="dmm_translation_in", + kind=Kind.omitted, + doc="DMM Translation InPos", + auto_monitor=True, + labels={"switch"}, + ) + dmm_bragg_thru = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMLS-3030:THRU", + add_prefix=("",), + name="dmm_bragg_thru", + kind=Kind.omitted, + doc="DMM Bragg ThruPos", + auto_monitor=True, + labels={"switch"}, + ) + dmm_bragg_in = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMLS-3040:IN", + add_prefix=("",), + name="dmm_bragg_in", + kind=Kind.omitted, + doc="DMM Bragg InPos", + auto_monitor=True, + labels={"switch"}, + ) + + dmm_heater_fault_xtal_1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMSW-3050:SWITCH", + add_prefix=("",), + name="dmm_heater_fault_xtal_1", + kind=Kind.omitted, + doc="DMM Heater Fault XTAL 1", + auto_monitor=True, + labels={"fault"}, + ) + dmm_heater_fault_xtal_2 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMSW-3060:SWITCH", + add_prefix=("",), + name="dmm_heater_fault_xtal_2", + kind=Kind.omitted, + doc="DMM Heater Fault XTAL 2", + auto_monitor=True, + labels={"fault"}, + ) + dmm_heater_fault_support_1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM-EMSW-3070:SWITCH", + add_prefix=("",), + name="dmm_heater_fault_support_1", + kind=Kind.omitted, + doc="DMM Heater Fault Support 1", + auto_monitor=True, + labels={"fault"}, + ) + + dmm_energy = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM1:ENERGY-GET", + add_prefix=("",), + name="dmm_energy", + kind=Kind.omitted, + doc="DMM monochromator energy", + auto_monitor=True, + labels={"energy"}, + ) + dmm_position = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM1:POSITION", + add_prefix=("",), + name="dmm_position", + kind=Kind.omitted, + doc="DMM Position", + auto_monitor=True, + labels={"string"}, + ) + dmm_stripe = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-DMM1:STRIPE", + add_prefix=("",), + name="dmm_stripe", + kind=Kind.omitted, + doc="DMM Stripe", + auto_monitor=True, + labels={"string"}, + ) + + +class CCMMonochromator(EPSSubDevices): + """CCM monochromator signals at the cSAXS beamline.""" + + ccm_temp_crystal = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM-ETTC-4010:TEMP", + add_prefix=("",), + name="ccm_temp_crystal", + kind=Kind.omitted, + doc="CCM Temp Crystal", + auto_monitor=True, + labels={"temp"}, + ) + ccm_temp_shield_disaster = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM-ETTC-4020:TEMP", + add_prefix=("",), + name="ccm_temp_shield_disaster", + kind=Kind.omitted, + doc="CCM Temp Shield (disaster)", + auto_monitor=True, + labels={"temp"}, + ) + + ccm_heater_fault_1 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM-EMSW-4010:SWITCH", + add_prefix=("",), + name="ccm_heater_fault_1", + kind=Kind.omitted, + doc="CCM Heater Fault 1", + auto_monitor=True, + labels={"fault"}, + ) + ccm_heater_fault_2 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM-EMSW-4020:SWITCH", + add_prefix=("",), + name="ccm_heater_fault_2", + kind=Kind.omitted, + doc="CCM Heater Fault 2", + auto_monitor=True, + labels={"fault"}, + ) + ccm_heater_fault_3 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM-EMSW-4030:SWITCH", + add_prefix=("",), + name="ccm_heater_fault_3", + kind=Kind.omitted, + doc="CCM Heater Fault 3", + auto_monitor=True, + labels={"fault"}, + ) + + ccm_energy = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM1:ENERGY-GET", + add_prefix=("",), + name="ccm_energy", + kind=Kind.omitted, + doc="CCM monochromator energy", + auto_monitor=True, + labels={"energy"}, + ) + ccm_position = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CCM1:POSITION", + add_prefix=("",), + name="ccm_position", + kind=Kind.omitted, + doc="CCM Position", + auto_monitor=True, + labels={"string"}, + ) + + +class CoolingWater(EPSSubDevices): + """Cooling water signals at the cSAXS beamline.""" + + op_sl1_efsw_2010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-SL1-EFSW-2010:FLOW", + add_prefix=("",), + name="op_sl1_efsw_2010_flow", + kind=Kind.omitted, + doc="OP-SL1-EFSW-2010", + auto_monitor=True, + labels={"flow"}, + ) + op_sl2_efsw_2010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-SL2-EFSW-2010:FLOW", + add_prefix=("",), + name="op_sl2_efsw_2010_flow", + kind=Kind.omitted, + doc="OP-SL2-EFSW-2010", + auto_monitor=True, + labels={"flow"}, + ) + op_eb1_efsw_5010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-EB1-EFSW-5010:FLOW", + add_prefix=("",), + name="op_eb1_efsw_5010_flow", + kind=Kind.omitted, + doc="OP-EB1-EFSW-5010", + auto_monitor=True, + labels={"flow"}, + ) + op_eb1_efsw_5020_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-EB1-EFSW-5020:FLOW", + add_prefix=("",), + name="op_eb1_efsw_5020_flow", + kind=Kind.omitted, + doc="OP-EB1-EFSW-5020", + auto_monitor=True, + labels={"flow"}, + ) + op_sl3_efsw_5010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-SL3-EFSW-5010:FLOW", + add_prefix=("",), + name="op_sl3_efsw_5010_flow", + kind=Kind.omitted, + doc="OP-SL3-EFSW-5010", + auto_monitor=True, + labels={"flow"}, + ) + op_kb_efsw_6010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-KB-EFSW-6010:FLOW", + add_prefix=("",), + name="op_kb_efsw_6010_flow", + kind=Kind.omitted, + doc="OP-KB-EFSW-6010", + auto_monitor=True, + labels={"flow"}, + ) + op_psh1_efsw_7010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-PSH1-EFSW-7010:FLOW", + add_prefix=("",), + name="op_psh1_efsw_7010_flow", + kind=Kind.omitted, + doc="OP-PSH1-EFSW-7010", + auto_monitor=True, + labels={"flow"}, + ) + es_eb2_efsw_1010_flow = Cpt( + EpicsSignalRO, + read_pv="X12SA-ES-EB2-EFSW-1010:FLOW", + add_prefix=("",), + name="es_eb2_efsw_1010_flow", + kind=Kind.omitted, + doc="ES-EB2-EFSW-1010", + auto_monitor=True, + labels={"flow"}, + ) + + op_cs_ecvw_0010 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CS-ECVW-0010:PLC_OPEN", + add_prefix=("",), + name="op_cs_ecvw_0010", + kind=Kind.omitted, + doc="OP-CS-ECVW-0010", + auto_monitor=True, + labels={"valve"}, + ) + op_cs_ecvw_0020 = Cpt( + EpicsSignalRO, + read_pv="X12SA-OP-CS-ECVW-0020:PLC_OPEN", + add_prefix=("",), + name="op_cs_ecvw_0020", + kind=Kind.omitted, + doc="OP-CS-ECVW-0020", + auto_monitor=True, + labels={"valve"}, + ) + + +class EPS(PSIDeviceBase): + + alarms = Cpt(EPSAlarms, name="alarms") + valves = Cpt(Valves, name="valves") + shutters = Cpt(Shutters, name="shutters") + dmm_monochromator = Cpt(DMMMonochromator, name="dmm_monochromator") + ccm_monochromator = Cpt(CCMMonochromator, name="ccm_monochromator") + cooling_water = Cpt(CoolingWater, name="cooling_water") + + # Acknowledgment signals for PLC communication (if needed for future use) + ackerr = EpicsSignal("X12SA-EPS-PLC:ACKERR-REQUEST") + request = EpicsSignal("X12SA-OP-CS-ECVW:PLC_REQUEST") + + def _notify(self, msg: str, show_as_client_msg: bool = True): + """Utility method to print a message, and optionally send it to the client UI if it should be shown also as a client message.""" + try: + if show_as_client_msg: + self.device_manager.connector.send_client_info(msg, scope="", show_asap=True) + else: + print(msg) + except Exception: + logger.error(f"Failed to send client message, falling back to print: {msg}") + print(str(msg)) + + # ---------------------------------------------------------- + # Water cooling operation + # ---------------------------------------------------------- + + def safe_get(self, sig, default=None): + """Helper method to safely get a signal value, returning a default if there's an error.""" + try: + return sig.get() + except Exception as ex: + logger.warning(f"Failed to get signal {sig.pvname}: {ex}") + return default + + def water_cooling_op(self): + """ + Open ECVW valves, reset EPS alarms, monitor for 20s, + then ensure stability (valves remain open) for 10s. + All messages sent to client. + """ + + POLL_PERIOD = 2 + TIMEOUT = 20 + STABILITY = 15 + + self._notify("=== Water Cooling Operation ===") + + # --- Signals --- + eps_alarm_sig = self.alarms.eps_alarm_cnt + ackerr = self.ackerr + request = self.request + + valves = [self.cooling_water.op_cs_ecvw_0010, self.cooling_water.op_cs_ecvw_0020] + + # Flow channels list extracted from CHANNELS + flow_items = [ + walk.item + for walk in self.cooling_water.walk_signals() + if "flow" in walk.item._ophyd_labels_ + ] + + # --- Step 1: EPS alarm reset --- + alarm_value = self.safe_get(eps_alarm_sig, 0) + if alarm_value and alarm_value > 0: + self._notify(f"[WaterCooling] EPS alarms present ({alarm_value}) → resetting…") + try: + ackerr.put(1) + except Exception as ex: + self._notify(f"[WaterCooling] WARNING: ACKERR write failed: {ex}") + time.sleep(0.3) + else: + self._notify("[WaterCooling] No EPS alarms detected.") + + # --- Step 2: Issue open request --- + self._notify("[WaterCooling] Sending cooling-valve OPEN request…") + try: + request.put(1) + except Exception as ex: + self._notify(f"[WaterCooling] ERROR: Failed to send OPEN request: {ex}") + return False + + # --- Step 3: Monitoring loop (clean client table output) --- + start = time.time() + end = start + TIMEOUT + stable_until = None + + # Print (server-side) header once + print("Monitoring valves and flow sensors...") + print(f" Valves: {valves[0].attr_name[-4:]}, {valves[1].attr_name[-4:]}") + print(f" Note: stability requires valves to remain OPEN for {STABILITY} seconds.") + + # One table header to the client (via device manager) + # Fixed-width columns for alignment in monospaced UI + table_header = f"{'Time':>6} | {'Valves':<21} | {'Flows (OK/FAIL/N/A)':<20}" + self._notify(table_header) + + def snapshot(): + # Valve snapshot + v_states = [self.safe_get(v, None) for v in valves] + v1 = f"{valves[0].attr_name[-4:]}=" + ( + "OPEN " + if v_states[0] is True or v_states[0] == 1 + else "CLOSED" if v_states[0] is False or v_states[0] == 0 else "N/A " + ) + v2 = f"{valves[1].attr_name[-4:]}=" + ( + "OPEN " + if v_states[1] is True or v_states[1] == 1 + else "CLOSED" if v_states[1] is False or v_states[1] == 0 else "N/A " + ) + # 2 valves with a single space between => width ~ 21 + valve_str = f"{v1} {v2}" + + # Flow summary: OK/FAIL/N/A counts (compact) + flow_states = [] + for fsig in flow_items: + fval = self.safe_get(fsig, None) + flow_states.append( + True if fval in (1, True) else False if fval in (0, False) else None + ) + + ok = sum(1 for f in flow_states if f is True) + fail = sum(1 for f in flow_states if f is False) + na = sum(1 for f in flow_states if f is None) + flow_summary = f"{ok:>2} / {fail:>2} / {na:>2}" + + return v_states, valve_str, flow_summary + + while True: + # TODO Consider adding a timeout to avoid infinite loop. + now = time.time() + elapsed = int(now - start) + + if now > end: + # One last line to client + v_states, valves_s, flows_s = snapshot() + self._notify(f"{elapsed:>6}s | {valves_s:<21} | {flows_s:<20}") + print("→ TIMEOUT: Cooling valves failed to remain OPEN.") + return False + + # Live snapshot + v_states, valves_s, flows_s = snapshot() + # Exactly one concise line to client per cycle + self._notify(f"{elapsed:>6}s | {valves_s:<21} | {flows_s:<20}") + + both_open = all(s is not None and bool(s) for s in v_states) + + if both_open: + if stable_until is None: + stable_until = now + STABILITY + print( + f"[WaterCooling] Both valves OPEN → starting {STABILITY}s stability window…" + ) + else: + if now >= stable_until: + print("→ SUCCESS: Valves remained OPEN during stability window.") + return True + else: + if stable_until is not None: + print("[WaterCooling] Valve closed again → restarting stability window.") + stable_until = None + + time.sleep(POLL_PERIOD) + + def show_all(self): + red = "\x1b[91m" + green = "\x1b[92m" + white = "\x1b[0m" + bold = "\x1b[1m" + cyan = "\x1b[96m" + + # ---- New: enum maps for numeric -> string rendering ---- + POSITION_ENUM = {0: "out of beam", 1: "in beam"} + STRIPE_ENUM = {0: "Stripe 1 W/B4C", 1: "Stripe 2 NiV/B4C"} + POSITION_ATTRS = { + self.dmm_monochromator.dmm_position.attr_name, + self.ccm_monochromator.ccm_position.attr_name, + } + STRIPE_ATTRS = {self.dmm_monochromator.dmm_stripe.attr_name} + + def is_bool_like(v): + return isinstance(v, (bool, int)) and v in (0, 1, True, False) + + # ---- Changed: accept attr in formatter so we can apply enum mapping ---- + def fmt_value(value: any, signal: EpicsSignalRO): + if value is None: + return f"{red}MISSING{white}" + + attr = signal.attr_name + + # ---------- Explicit enum mappings by attribute ---------- + if attr in POSITION_ATTRS: + # Position comes as numeric 0/1 + try: + iv = int(value) + return POSITION_ENUM.get(iv, f"{iv}") + except Exception: + # Fallback if it’s already a string or unexpected + return f"{value}" + + if attr in STRIPE_ATTRS: + # Stripe comes as numeric 0/1 + try: + iv = int(value) + return STRIPE_ENUM.get(iv, f"{iv}") + except Exception: + return f"{value}" + + # ------------------- TEMPERATURE ------------------- + if "temp" in signal._ophyd_labels_ and isinstance(value, (int, float)): + return f"{value:.1f}" + + # ------------------- ENERGY ------------------------ + if "energy" in signal._ophyd_labels_ and isinstance(value, (int, float)): + return f"{value:.4f}" + + # ------------------- STRINGS ----------------------- + if "string" in signal._ophyd_labels_ or "position" in signal._ophyd_labels_: + # For other strings, just echo the value + return f"{value}" + + # ------------------- SWITCH (ACTIVE/INACTIVE) ------ + if "switch" in signal._ophyd_labels_ and is_bool_like(value): + return f"{green+'ACTIVE'+white if value else red+'INACTIVE'+white}" + + # ------------------- FAULT (OK/FAULT) -------------- + if "fault" in signal._ophyd_labels_ and is_bool_like(value): + return f"{green+'OK'+white if not value else red+'FAULT'+white}" + + # ------------------- VALVE/SHUTTER ----------------- + if ( + "valve" in signal._ophyd_labels_ or "shutter" in signal._ophyd_labels_ + ) and is_bool_like(value): + return f"{green+'OPEN'+white if value else red+'CLOSED'+white}" + + # ------------------- FLOW (OK/FAIL) ---------------- + if "flow" in signal._ophyd_labels_ and is_bool_like(value): + return f"{green}OK{white}" if bool(value) else f"{red}FAIL{white}" + + # ------------------- FALLBACK ----------------------- + return f"{value}" + + # ------------------- PRINT START --------------------- + print(f"{bold}X12SA EPS status{white}") + + for sub_device in self.walk_components(): + print(f"\n{bold}{sub_device.name}{white}") + + rows = [] + for walk in sub_device.walk_components(): + cpt: Cpt = walk.ancestors[-1] + it: EpicsSignalRO = walk.item + val = self.safe_get(it) + rows.append((cpt.doc, val, it._read_pv.pvname, it._ophyd_labels_[0], it.attr_name)) + + label_width = max(32, *(len(label) for (label, _, _, _, _) in rows)) + + for label, value, pv, kind, _attr in rows: + fv = fmt_value(value, it) # <-- pass attr to formatter + print(f" - {label:<{label_width}} {fv}") + + if sub_device.name == "cooling_water": + v1 = self.safe_get(self.cooling_water.op_cs_ecvw_0010) + v2 = self.safe_get(self.cooling_water.op_cs_ecvw_0020) + + def closed(v): + return is_bool_like(v) and not bool(v) + + if closed(v1) and closed(v2): + print( + f"\n{cyan}Hint:{white} Both water cooling valves are CLOSED.\n" + f"You can open them using: {bold}dev.x12saEPS.water_cooling_op(){white}" + ) + + # ---------------------------------------------------------- + # Consistency report + # ---------------------------------------------------------- + # def consistency_report(self, *, verbose=True): + # missing = [] + # dupes = [] + # seen = {} + + # for sub_device in self.walk_components(): + # section = sub_device.name + # for walk in sub_device.walk_components(): + # cpt: Cpt = walk.ancestors[-1] + # it: EpicsSignalRO = walk.item + # if not hasattr(self, it["attr"]): + # missing.append((section, it["attr"], it["label"], it["pv"])) + + # pv = it["pv"] + # if pv in seen: + # dupes.append((pv, seen[pv], (section, it["attr"], it["label"]))) + # else: + # seen[pv] = (section, it["attr"], it["label"]) + + # if verbose: + # print("=== Consistency Report ===") + + # if missing: + # print("\nMissing attributes:") + # for sec, a, lbl, pv in missing: + # print(f" - [{sec}] {a} {lbl} pv={pv}") + # else: + # print("\nNo missing attributes.") + + # if dupes: + # print("\nDuplicate PVs:") + # for pv, f1, f2 in dupes: + # print(f" {pv} → {f1} AND {f2}") + # else: + # print("\nNo duplicate PVs.") + + # return {"missing_attrs": missing, "duplicate_pvs": dupes} # ============================================================================== @@ -154,19 +890,13 @@ CHANNELS = { def create_dynamic_eps_class(): class_attrs = dict( - USER_ACCESS=["show_all", "water_cooling_op"], - SUB_VALUE="value", - _default_sub="value", + USER_ACCESS=["show_all", "water_cooling_op"], SUB_VALUE="value", _default_sub="value" ) # Dynamically define Components for section, items in CHANNELS.items(): for it in items: - class_attrs[it["attr"]] = Cpt( - EpicsSignal, - name=it["label"], - read_pv=it["pv"] - ) + class_attrs[it["attr"]] = Cpt(EpicsSignal, name=it["label"], read_pv=it["pv"]) # ---------------------------------------------------------- # Methods @@ -260,25 +990,21 @@ def create_dynamic_eps_class(): # One table header to the client (via device manager) # Fixed-width columns for alignment in monospaced UI - table_header = ( - f"{'Time':>6} | {'Valves':<21} | {'Flows (OK/FAIL/N/A)':<20}" - ) + table_header = f"{'Time':>6} | {'Valves':<21} | {'Flows (OK/FAIL/N/A)':<20}" self._notify(table_header) def snapshot(): # Valve snapshot v_states = [safe_get(v, None) for v in valves] - v1 = ( - f"{valve_attrs[0][-4:]}=" - + ("OPEN " if v_states[0] is True or v_states[0] == 1 else - "CLOSED" if v_states[0] is False or v_states[0] == 0 else - "N/A ") + v1 = f"{valve_attrs[0][-4:]}=" + ( + "OPEN " + if v_states[0] is True or v_states[0] == 1 + else "CLOSED" if v_states[0] is False or v_states[0] == 0 else "N/A " ) - v2 = ( - f"{valve_attrs[1][-4:]}=" - + ("OPEN " if v_states[1] is True or v_states[1] == 1 else - "CLOSED" if v_states[1] is False or v_states[1] == 0 else - "N/A ") + v2 = f"{valve_attrs[1][-4:]}=" + ( + "OPEN " + if v_states[1] is True or v_states[1] == 1 + else "CLOSED" if v_states[1] is False or v_states[1] == 0 else "N/A " ) # 2 valves with a single space between => width ~ 21 valve_str = f"{v1} {v2}" @@ -288,7 +1014,9 @@ def create_dynamic_eps_class(): for f_attr, _lbl in flow_items: fsig = getattr(self, f_attr, None) fval = safe_get(fsig, None) - flow_states.append(True if fval in (1, True) else False if fval in (0, False) else None) + flow_states.append( + True if fval in (1, True) else False if fval in (0, False) else None + ) ok = sum(1 for f in flow_states if f is True) fail = sum(1 for f in flow_states if f is False) @@ -318,7 +1046,9 @@ def create_dynamic_eps_class(): if both_open: if stable_until is None: stable_until = now + STABILITY - print(f"[WaterCooling] Both valves OPEN → starting {STABILITY}s stability window…") + print( + f"[WaterCooling] Both valves OPEN → starting {STABILITY}s stability window…" + ) else: if now >= stable_until: print("→ SUCCESS: Valves remained OPEN during stability window.") @@ -330,158 +1060,6 @@ def create_dynamic_eps_class(): time.sleep(POLL_PERIOD) - def show_all(self): - red = "\x1b[91m" - green = "\x1b[92m" - white = "\x1b[0m" - bold = "\x1b[1m" - cyan = "\x1b[96m" - - # ---- New: enum maps for numeric -> string rendering ---- - POSITION_ENUM = { - 0: "out of beam", - 1: "in beam", - } - STRIPE_ENUM = { - 0: "Stripe 1 W/B4C", - 1: "Stripe 2 NiV/B4C", - } - POSITION_ATTRS = {"DMM1_POSITION", "CCM1_POSITION"} - STRIPE_ATTRS = {"DMM1_STRIPE"} - - def safe_get(attr): - obj = getattr(self, attr, None) - try: - return obj.get() - except Exception: - return None - - def is_bool_like(v): - return isinstance(v, (bool, int)) and v in (0, 1, True, False) - - # ---- Changed: accept attr in formatter so we can apply enum mapping ---- - def fmt_value(value, pv, kind, attr): - if value is None: - return f"{red}MISSING{white}" - - # ---------- Explicit enum mappings by attribute ---------- - if attr in POSITION_ATTRS: - # Position comes as numeric 0/1 - try: - iv = int(value) - return POSITION_ENUM.get(iv, f"{iv}") - except Exception: - # Fallback if it’s already a string or unexpected - return f"{value}" - - if attr in STRIPE_ATTRS: - # Stripe comes as numeric 0/1 - try: - iv = int(value) - return STRIPE_ENUM.get(iv, f"{iv}") - except Exception: - return f"{value}" - - # ------------------- TEMPERATURE ------------------- - if kind == "temp" and isinstance(value, (int, float)): - return f"{value:.1f}" - - # ------------------- ENERGY ------------------------ - if kind == "energy" and isinstance(value, (int, float)): - return f"{value:.4f}" - - # ------------------- STRINGS ----------------------- - if kind in ("string", "position"): - # For other strings, just echo the value - return f"{value}" - - # ------------------- SWITCH (ACTIVE/INACTIVE) ------ - if kind == "switch" and is_bool_like(value): - return f"{green+'ACTIVE'+white if value else red+'INACTIVE'+white}" - - # ------------------- FAULT (OK/FAULT) -------------- - if kind == "fault" and is_bool_like(value): - return f"{green+'OK'+white if not value else red+'FAULT'+white}" - - # ------------------- VALVE/SHUTTER ----------------- - if kind in ("valve", "shutter") and is_bool_like(value): - return f"{green+'OPEN'+white if value else red+'CLOSED'+white}" - - # ------------------- FLOW (OK/FAIL) ---------------- - if kind == "flow" and is_bool_like(value): - return f"{green}OK{white}" if bool(value) else f"{red}FAIL{white}" - - # ------------------- FALLBACK ----------------------- - return f"{value}" - - # ------------------- PRINT START --------------------- - print(f"{bold}X12SA EPS status{white}") - - for section, items in CHANNELS.items(): - print(f"\n{bold}{section}{white}") - - rows = [] - for it in items: - val = safe_get(it["attr"]) - rows.append((it["label"], val, it["pv"], it["kind"], it["attr"])) - - label_width = max(32, *(len(label) for (label, _, _, _, _) in rows)) - - for label, value, pv, kind, _attr in rows: - fv = fmt_value(value, pv, kind, _attr) # <-- pass attr to formatter - print(f" - {label:<{label_width}} {fv}") - - if section == "Cooling Water": - v1 = safe_get("OPCSECVW0010") - v2 = safe_get("OPCSECVW0020") - - def closed(v): return is_bool_like(v) and not bool(v) - - if closed(v1) and closed(v2): - print( - f"\n{cyan}Hint:{white} Both water cooling valves are CLOSED.\n" - f"You can open them using: {bold}dev.x12saEPS.water_cooling_op(){white}" - ) - - # ---------------------------------------------------------- - # Consistency report - # ---------------------------------------------------------- - def consistency_report(self, *, verbose=True): - missing = [] - dupes = [] - seen = {} - - for section, items in CHANNELS.items(): - for it in items: - if not hasattr(self, it["attr"]): - missing.append((section, it["attr"], it["label"], it["pv"])) - - pv = it["pv"] - if pv in seen: - dupes.append((pv, seen[pv], (section, it["attr"], it["label"]))) - else: - seen[pv] = (section, it["attr"], it["label"]) - - if verbose: - print("=== Consistency Report ===") - - if missing: - print("\nMissing attributes:") - for sec, a, lbl, pv in missing: - print(f" - [{sec}] {a} {lbl} pv={pv}") - else: - print("\nNo missing attributes.") - - if dupes: - print("\nDuplicate PVs:") - for pv, f1, f2 in dupes: - print(f" {pv} → {f1} AND {f2}") - else: - print("\nNo duplicate PVs.") - - return {"missing_attrs": missing, "duplicate_pvs": dupes} - - # Attach methods to class attributes class_attrs["show_all"] = DynamicMethods.show_all class_attrs["consistency_report"] = DynamicMethods.consistency_report