added device for dewar monitoring

csaxs_bec/device_configs/omny_config.yaml

@@ -9,6 +9,17 @@ omny_samples:
   onFailure: buffer
   readOnly: false
   readoutPriority: baseline
+############################################################
+##################### OMNY dewar ###########################
+############################################################
+omny_dewar:
+  description: OMNY Dewar Information
+  deviceClass: csaxs_bec.devices.omny.omny_dewar.OMNYDewar
+  deviceConfig: {}
+  enabled: true
+  onFailure: buffer
+  readOnly: false
+  readoutPriority: baseline
 
 ############################################################
 ##################### OMNY temperatures ####################

csaxs_bec/devices/omny/omny_dewar.py

@@ -0,0 +1,111 @@
+import time
+import datetime
+
+from ophyd import Component as Cpt
+from ophyd import Device
+from ophyd import DynamicDeviceComponent as Dcpt
+from ophyd import EpicsSignal
+from prettytable import FRAME, PrettyTable
+import numpy as np
+
+class OMNYDewarError(Exception):
+    pass
+
+
+class OMNYDewar(Device):
+    USER_ACCESS = [
+        "show_all",
+        "is_flow_low",
+        "help",
+    ]
+    SUB_VALUE = "value"
+    _default_sub = SUB_VALUE
+
+
+
+    dewar_press_0 = Cpt(
+        EpicsSignal, name="dewar_press_0", read_pv="XOMNY-TEMP-DEWAR-PRESS0:GET"
+    )
+
+    dewar_refilling = Cpt(
+        EpicsSignal, name="dewar_refilling", read_pv="XOMNY-TEMP-DEWAR-Refilling:GET"
+    )
+
+    dewar_press_1 = Cpt(
+        EpicsSignal, name="dewar_press_1", read_pv="XOMNY-TEMP-DEWAR-PRESS1:GET"
+    )
+
+    dewar_press_2 = Cpt(
+        EpicsSignal, name="dewar_press_2", read_pv="XOMNY-TEMP-DEWAR-PRESS2:GET"
+    )
+
+    dewar_flow_0 = Cpt(
+        EpicsSignal, name="dewar_flow_0", read_pv="XOMNY-TEMP-DEWAR-FLOW0:GET"
+    )
+
+    dewar_valvepos = Cpt(
+        EpicsSignal, name="dewar_valvepos", read_pv="XOMNY-TEMP-DEWAR-ValvePos:GET"
+    )
+
+    dewar_supply_voltage = Cpt(
+        EpicsSignal, name="dewar_supply_voltage", read_pv="XOMNY-TEMP-DEWAR-SupplyVoltage:GET"
+    )
+
+    dewar_uptime_h = Cpt(
+        EpicsSignal, name="dewar_uptime_h", read_pv="XOMNY-TEMP-DEWAR-UpH:GET"
+    )
+    dewar_uptime_m = Cpt(
+        EpicsSignal, name="dewar_uptime_m", read_pv="XOMNY-TEMP-DEWAR-UpM:GET"
+    )
+    dewar_uptime_s = Cpt(
+        EpicsSignal, name="dewar_uptime_s", read_pv="XOMNY-TEMP-DEWAR-UpS:GET"
+    )
+
+    dewar_valve_movements_h = Cpt(
+        EpicsSignal, name="dewar_valve_movements_h", read_pv="XOMNY-TEMP-DEWAR-MovH:GET"
+    )
+    dewar_valve_movements_m = Cpt(
+        EpicsSignal, name="dewar_valve_movements_m", read_pv="XOMNY-TEMP-DEWAR-MovM:GET"
+    )
+    dewar_valve_movements_s = Cpt(
+        EpicsSignal, name="dewar_valve_movements_s", read_pv="XOMNY-TEMP-DEWAR-MovS:GET"
+    )
+
+    def __init__(self, prefix="", *, name, **kwargs):
+        super().__init__(prefix, name=name, **kwargs)
+        self.dewar_flow_0.subscribe(self._emit_value)
+
+    def _emit_value(self, **kwargs):
+        timestamp = kwargs.pop("timestamp", time.time())
+        self.wait_for_connection()
+        self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
+
+
+    def is_flow_low(self):
+        if(float(self.dewar_flow_0.get())<3.8):
+            return True
+        else:
+            return False
+        
+    def show_all(self):
+        red = "\x1b[91m"
+        white = "\x1b[0m"
+
+
+        print("OMNY Dewar Status")
+        print(f"  DewarPressure:           {float(self.dewar_press_0.get()):.0f} mbar")
+        print(f"  Dewar Refilling:         {float(self.dewar_refilling.get()):.0f}")
+        print(f"  LN2flow In Pressure:     {float(self.dewar_press_1.get()):.0f} mbar")
+        print(f"  LN2flow Out Pressure:    {float(self.dewar_press_2.get()):.0f} mbar")
+        print(f"  LN2flow In Flow Rate:    {float(self.dewar_flow_0.get()):.1f} l/s")
+        if self.is_flow_low():
+            print(red + "This flow rate is LOW. Increase the LN2flow In Pressure" + white)
+        print(f"  Valve Opening (0-1):     {float(self.dewar_valvepos.get()):.2f}")
+        print(f"  Valve Supply Voltage:    {float(self.dewar_supply_voltage.get()):.2f} V")
+        print(f"  Uptime of System:        {float(self.dewar_uptime_h.get()):2.0f}:{float(self.dewar_uptime_m.get()):2.0f}:{float(self.dewar_uptime_s.get()):2.0f}")
+        print(f"  Active Valve Movements:  {float(self.dewar_valve_movements_h.get()):2.0f}:{float(self.dewar_valve_movements_m.get()):2.0f}:{float(self.dewar_valve_movements_s.get()):2.0f}")
+  
+
+    def help(self):
+        print("Help for OMNY Dewar:")
+        print("This device shows an overview of the dewar status. Use show_all()")