Skip to content

Door Watcher

door_watcher

Classes

Watchman 

Watchman(door_state, voltage_read, voltage_write)
Source code in scripts/door_watcher/door_watcher.py 
10
11
12
13
14
def __init__(self, door_state, voltage_read, voltage_write):
    self.door_state = EpicsSignalRO(door_state)
    self.mesh_voltage = EpicsSignal(voltage_read,
                                    write_pv=voltage_write)
    print('Using door pv {}'.format(door_state))
Attributes
cbid class-attribute instance-attribute 
cbid = None
door_state instance-attribute 
door_state = EpicsSignalRO(door_state)
mesh_voltage instance-attribute 
mesh_voltage = EpicsSignal(voltage_read, write_pv=voltage_write)
retries class-attribute instance-attribute 
retries = 3
Functions
callback 
callback(old_value=None, value=None, **kwargs)

To be run every time the door state changes

Source code in scripts/door_watcher/door_watcher.py 
23
24
25
26
27
28
29
30
31
32
33
def callback(self, old_value=None, value=None, **kwargs):
    """To be run every time the door state changes"""
    print('Door PV changed from {} to {}'.format(old_value, value))
    if old_value == 1 and value == 0:
        print('Zeroing the mesh voltage because door was opened')
        # Let's retry a few times in case of a network error
        for i in range(self.retries):
            try:
                self.mesh_voltage.put(0)
            except Exception:
                pass
start 
start()

Sets up self.callback to be run when door state changes

Source code in scripts/door_watcher/door_watcher.py 
16
17
18
19
20
21
def start(self):
    """Sets up self.callback to be run when door state changes"""
    # Make sure we don't start 2 processes
    self.stop()
    print('Starting door watcher')
    self.door_state.subscribe(self.callback)
stop 
stop()

Shutdown the watcher

Source code in scripts/door_watcher/door_watcher.py 
35
36
37
38
39
40
def stop(self):
    """Shutdown the watcher"""
    if self.cbid is not None:
        print('Stopping door watcher')
        self.door_state.unsubscribe(self.cbid)
        self.cbid = None

Functions

main 

main()
Source code in scripts/door_watcher/door_watcher.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def main():
    watcher = Watchman(
        'PPS:FEH1:45:DOORA',
        'MFX:USR:ai1:1',
        'MFX:USR:ao1:0')

    # If door already open, maybe we crashed? Let's be safe and turn it off.
    if watcher.door_state.get() == 0:
        print('Door is already open! Setting to zero at startup')
        watcher.mesh_voltage.put(0)

    watcher.start()
    try:
        # Start a loop for user text feedback.
        # We want reassurance that the program is still running.
        while True:
            print('Door watcher still running. Door PV is {}, '
                  'mesh voltage is {}'.format(watcher.door_state.get(),
                                              watcher.mesh_voltage.get()*1000))
            time.sleep(5)
    except KeyboardInterrupt:
        print('Watcher shutting down')
    finally:
        watcher.stop()