Source code for csaxs_bec.devices.omny.omny_dewar

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()")