server.py

   1 #! /usr/bin/env python
   2 # UCC Door Server - hardware interface server
   3 # reads status of reed switches and provides a D-BUS interface to read them
   4 # David Adam <[email protected]>
   5 # LAT interface based on work by Mark Tearle <[email protected]>
   6 # Released under an MIT-style license; see COPYING for details.
   7
   8 # statuses: 1 means open circuit, 0 means closed circuit, -1 means error
   9
  10 import dbus, dbus.service
  11 import gobject
  12 from LATClient import LATClient
  13 from select import select
  14
  15 def check_LAT_service(servicename, retries=3):
  16     latclient = LATClient(service=servicename.upper())
  17     rfh, wfh = latclient.get_fh()
  18     results = {'error': retries, 'closed': retries, 'open': retries}
  19     test_string = "got wombles?"
  20
  21     # Only return after n(retries) consistent results, although not necessarily
  22     # in order. This means we poll at least (retries) and up to (retries ** 2) times.
  23     # This is mtearle's original code. No idea if it's the Right Thing.
  24     while True:
  25         try:
  26             wfh.write(test_string)
  27             wfh.flush()
  28         except:
  29             results['error'] -= 1
  30             if results['error'] == 0: return -1
  31             continue
  32         rr, wr, er = select([rfh], [], [], 3.0)
  33         if rfh not in rr:
  34             results['open'] -= 1
  35             if results['open'] == 0: return 1
  36             continue
  37         recv = rfh.read(len(test_string))
  38         if recv <> test_string:
  39             results['error'] -= 1
  40             if results['error'] == 0: return -1
  41             continue
  42         results['closed'] -= 1
  43         if results['closed'] == 0: return 0
  44
  45 class Door(dbus.service.Object):
  46
  47     def __init__(self, doorname, bus):
  48         self.interval = 10 # seconds
  49         self.service = doorname
  50         self.status = -1
  51         self.retries = 3
  52         if 'pir' in self.service:
  53             self.retries = 1
  54
  55         # set up D-BUS service name
  56         object_path = '/au/asn/ucc/doors/%s' % doorname
  57         dbus.service.Object.__init__(self, bus, object_path)
  58
  59         # get initial state
  60         self.poll()
  61
  62     def poll(self):
  63         try:
  64             newstatus = check_LAT_service(self.service, self.retries)
  65         except:
  66             newstatus = -1
  67
  68         if newstatus != self.status:
  69             self.status = newstatus
  70             # emit signal
  71             self.status_changed(newstatus)
  72
  73         # set up timeout again
  74         gobject.timeout_add_seconds(self.interval, self.poll)
  75
  76     @dbus.service.signal('au.asn.ucc.DoorInterface', signature='n')
  77     def status_changed(self, newstatus):
  78         pass
  79
  80     @dbus.service.method('au.asn.ucc.DoorInterface', in_signature='',
  81                          out_signature='n')
  82     def get_status(self):
  83         return self.status
  84
  85 if __name__ == '__main__':
  86     doors = ('uccdoor', 'unisfadoor', 'chdoor', 'mrdoor', 'uccpir')
  87
  88     from dbus.mainloop.glib import DBusGMainLoop
  89     DBusGMainLoop(set_as_default=True)
  90     # get on the bus
  91     system_bus = dbus.SystemBus()
  92     system_bus.request_name('au.asn.ucc.DoorServer')
  93
  94     door_objects = []
  95
  96     for door in doors:
  97         door_objects.append(Door(door, system_bus))
  98
  99     # engage!
 100     loop = gobject.MainLoop()
 101     try:
 102         loop.run()
 103     except KeyboardInterrupt:
 104         loop.quit()
 105     except:
 106         pass