Add real-time support for plugin devices.

Author: jburhenn

myDevices/devices/manager.py

@@ -68,7 +68,7 @@ def removeDevice(name):
         if name in DEVICES:
             if name in DYNAMIC_DEVICES:
                 if hasattr(DEVICES[name]["device"], 'close'):
-                        DEVICES[name]["device"].close()
+                    DEVICES[name]["device"].close()
                 del DEVICES[name]
                 del DYNAMIC_DEVICES[name]
                 json_devices = getJSON(DYNAMIC_DEVICES)

myDevices/plugins/manager.py

@@ -16,16 +16,17 @@
 
 
 class PluginManager():
-    """Loads plugins and reads/writes plugin data"""
+    """Loads plugins and reads/writes plugin data."""
 
-    def __init__(self):
-        """Initializes the plugin manager and loads the plugin list"""
+    def __init__(self, callback=None):
+        """Initializes the plugin manager and loads the plugin list."""
         self.plugin_folder = PLUGIN_FOLDER
+        self.callback = callback
         self.plugins = {}
         self.load_plugins()
 
     def load_plugin_from_file(self, filename):
-        """Loads a plugin from a specified plugin config file and adds it to the plugin list"""
+        """Loads a plugin from a specified plugin config file and adds it to the plugin list."""
         try:
             info('Loading plugin: {}'.format(filename))
             loaded = []
@@ -41,8 +42,9 @@ def load_plugin_from_file(self, filename):
                             'filename': filename,
                             'section': section,
                             'channel': config.get(section, 'channel'),
-                            'name': config.get(section, 'name', section)
+                            'name': config.get(section, 'name', section),
                         }
+                        plugin['id'] = plugin_name + ':' + plugin['channel']
                         inherit_items = {}
                         if inherit in config.sections():
                             if inherit == section:
@@ -70,7 +72,16 @@ def load_plugin_from_file(self, filename):
                             self.override_plugin_value(config, section, 'write', plugin)
                         except:
                             pass
-                        self.plugins[plugin_name + ':' + plugin['channel']] = plugin
+                        try:
+                            self.override_plugin_value(config, section, 'register_callback', plugin)
+                            getattr(plugin['instance'], plugin['register_callback'])(lambda x: self.data_changed(x, plugin))
+                        except:
+                            pass
+                        try:
+                            self.override_plugin_value(config, section, 'unregister_callback', plugin)
+                        except:
+                            pass                                                    
+                        self.plugins[plugin['id']] = plugin
                         loaded.append(section)
                 except Exception as e:
                     error(e)
@@ -79,13 +90,13 @@ def load_plugin_from_file(self, filename):
         info('Loaded sections: {}'.format(loaded))
 
     def load_plugins(self):
-        """Loads plugins from any plugin config files found in the plugin folder"""
+        """Loads plugins from any plugin config files found in the plugin folder."""
         for root, dirnames, filenames in os.walk(self.plugin_folder):
             for filename in fnmatch.filter(filenames, '*.plugin'):
                 self.load_plugin_from_file(os.path.join(root, filename))
 
     def get_plugin(self, filename, section):
-        """Return the plugin for the corresponding filename and section"""
+        """Return the plugin for the corresponding filename and section."""
         return next(plugin for plugin in self.plugins.values() if plugin['filename'] == filename and plugin['section'] == section)
 
     def override_plugin_value(self, config, section, key, plugin):
@@ -96,7 +107,7 @@ def override_plugin_value(self, config, section, key, plugin):
             plugin[key] = config.get(section, key, plugin[key])
 
     def get_plugin_readings(self):
-        """Return a list with current readings for all plugins"""
+        """Return a list with current readings for all plugins."""
         readings = []
         for key, plugin in self.plugins.items():
             try:
@@ -111,7 +122,7 @@ def get_plugin_readings(self):
         return readings
 
     def convert_to_dict(self, value):
-        """Convert a tuple value to a dict containing value, type and unit"""
+        """Convert a tuple value to a dict containing value, type and unit."""
         value_dict = {}
         try:
             if value is None or value[0] is None:
@@ -123,18 +134,19 @@ def convert_to_dict(self, value):
             if not value_dict:
                 value_dict['value'] = value
             if 'type' in value_dict and 'unit' not in value_dict:
-                if value_dict['type'] == 'digital_actuator':
+                if value_dict['type'] in ('digital_sensor', 'digital_actuator'):
                     value_dict['unit'] = 'd'
-                elif value_dict['type'] == 'analog_actuator':
+                elif value_dict['type'] in ('analog_sensor', 'analog_actuator'):
                     value_dict['unit'] = 'null'
         return value_dict
 
     def is_plugin(self, plugin, channel=None):
-        """Returns True if the specified plugin or plugin:channel are valid plugins"""
+        """Returns True if the specified plugin or plugin:channel are valid plugins."""
         try:
             key = plugin
             if channel is not None:
                 key = plugin + ':' + channel
+            info('Checking for {} in {}'.format(key, self.plugins.keys()))
             return key in self.plugins.keys()
         except:
             return False
@@ -155,18 +167,37 @@ def write_value(self, plugin, channel, value):
             return False
         return True
 
-    def disable(self, plugin):
-        """Disable the specified plugin"""
+    def disable(self, plugin_id):
+        """Disable the specified plugin."""
         disabled = False
         try:
-            output, result = executeCommand('sudo python3 -m myDevices.plugins.disable "{}" "{}"'.format(self.plugins[plugin]['filename'], self.plugins[plugin]['section']))
+            plugin = self.plugins[plugin_id]
+            output, result = executeCommand('sudo python3 -m myDevices.plugins.disable "{}" "{}"'.format(plugin['filename'], plugin['section']))
             if result == 0:
                 disabled = True
-                info('Plugin \'{}\' disabled'.format(plugin))
+                info('Plugin \'{}\' disabled'.format(plugin_id))
             else:
-                info('Plugin \'{}\' not disabled'.format(plugin))
-            del self.plugins[plugin]
+                info('Plugin \'{}\' not disabled'.format(plugin_id))
+            if 'unregister_callback' in plugin:
+                getattr(plugin['instance'], plugin['unregister_callback'])()
+            del self.plugins[plugin_id]
         except Exception as e:
             info(e)
             pass
-        return disabled
+        return disabled
+
+    def register_callback(self, callback):
+        """Register the callback to use when plugin data has changed."""
+        self.callback = callback
+
+    def unregister_callback(self):
+        """Unregister the callback to use when plugin data has changed."""
+        self.callback = None
+
+    def data_changed(self, value, plugin):
+        """Callback that is called when data has changed."""
+        if self.callback:
+            data = self.convert_to_dict(value)
+            data['name'] = plugin['name']
+            data['id'] = plugin['id']
+            self.callback(data)

myDevices/sensors/sensors.py

@@ -23,7 +23,7 @@
 from myDevices.utils.types import M_JSON
 
 REFRESH_FREQUENCY = 15 #seconds
-DIGITAL_FREQUENCY = 60/55 #Seconds/messages, this is done to keep messages under the rate limit
+REAL_TIME_FREQUENCY = 60/55 #Seconds/messages, this is done to keep messages under the rate limit
 
 class SensorsClient():
     """Class for interfacing with sensors and actuators"""
@@ -36,8 +36,8 @@ def __init__(self):
         self.onDataChanged = None
         self.systemData = []
         self.currentSystemState = []
-        self.currentDigitalData = {}                                
-        self.queuedDigitalData = {}
+        self.currentRealTimeData = {}                                
+        self.queuedRealTimeData = {}
         self.disabledSensors = {}
         self.disabledSensorTable = "disabled_sensors"
         checkAllBus()
@@ -50,8 +50,8 @@ def __init__(self):
         if results:
             for row in results:
                 self.disabledSensors[row[0]] = 1
-        self.pluginManager = PluginManager()
-        self.digitalMonitorRunning = False
+        self.realTimeMonitorRunning = False
+        self.pluginManager = PluginManager(self.OnPluginChange)
         self.InitCallbacks()
         self.StartMonitoring()
 
@@ -72,10 +72,28 @@ def OnSensorChange(self, device, value):
         """
         debug('OnSensorChange: {}, {}'.format(device, value))
         with self.digitalMutex:
-            if device['name'] not in self.currentDigitalData:
-                self.currentDigitalData[device['name']] = {'device': device, 'value': value}
+            data = {'name': device['description'], 'value': value, 'type': 'digital_sensor', 'unit': 'd'}
+            if 'args' in device:
+                data['args'] = device['args']
+            if device['name'] not in self.currentRealTimeData:
+                self.currentRealTimeData[device['name']] = data
             else:
-                self.queuedDigitalData[device['name']] = {'device': device, 'value': value}
+                self.queuedRealTimeData[device['name']] = data
+
+    def OnPluginChange(self, data):
+        """Callback that is called when digital sensor data has changed
+
+        Args:
+            data: The new data value
+        """
+        debug('OnPluginChange: {}'.format(data))
+        with self.digitalMutex:
+            if data['id'] not in self.currentRealTimeData:
+                self.currentRealTimeData[data['id']] = data
+            else:
+                self.queuedRealTimeData[data['id']] = data
+            if not self.realTimeMonitorRunning:
+                ThreadPool.Submit(self.RealTimeMonitor)
 
     def InitCallbacks(self):
         """Set callback function for any digital devices that support them"""
@@ -85,8 +103,8 @@ def InitCallbacks(self):
             if 'DigitalSensor' in device['type'] and hasattr(sensor, 'setCallback'):
                 debug('Set callback for {}'.format(sensor))
                 sensor.setCallback(self.OnSensorChange, device)
-                if not self.digitalMonitorRunning:
-                    ThreadPool.Submit(self.DigitalMonitor)
+                if not self.realTimeMonitorRunning:
+                    ThreadPool.Submit(self.RealTimeMonitor)
 
     def RemoveCallbacks(self):
         """Remove callback function for all digital devices"""
@@ -136,35 +154,35 @@ def Monitor(self):
                 exception('Monitoring sensors and os resources failed')
         debug('Monitoring sensors and os resources finished')
 
-    def DigitalMonitor(self):
-        """Monitor digital state changes and report changed data via callbacks"""
-        self.digitalMonitorRunning = True
-        info('Monitoring digital sensor changes')
+    def RealTimeMonitor(self):
+        """Monitor real-time state changes and report changed data via callbacks"""
+        self.realTimeMonitorRunning = True
+        info('Monitoring real-time state changes')
         nextTime = datetime.now()
         while not self.exiting.is_set():
             try:
                 if not self.exiting.wait(0.5):
                     if datetime.now() > nextTime:
-                        nextTime = datetime.now() + timedelta(seconds=DIGITAL_FREQUENCY)
+                        nextTime = datetime.now() + timedelta(seconds=REAL_TIME_FREQUENCY)
                         data = []
                         with self.digitalMutex:
-                            if self.currentDigitalData:
-                                for name, item in self.currentDigitalData.items():
-                                    cayennemqtt.DataChannel.add_unique(data, cayennemqtt.DEV_SENSOR, name, value=item['value'], name=item['device']['description'], type='digital_sensor', unit='d')
+                            if self.currentRealTimeData:
+                                for name, item in self.currentRealTimeData.items():
+                                    cayennemqtt.DataChannel.add_unique(data, cayennemqtt.DEV_SENSOR, name, value=item['value'], name=item['name'], type=item['type'], unit=item['unit'])
                                     try:
-                                        cayennemqtt.DataChannel.add_unique(data, cayennemqtt.SYS_GPIO, item['device']['args']['channel'], cayennemqtt.VALUE, item['value'])
+                                        cayennemqtt.DataChannel.add_unique(data, cayennemqtt.SYS_GPIO, item['args']['channel'], cayennemqtt.VALUE, item['value'])
                                     except:
                                         pass
-                                    if name in self.queuedDigitalData and self.queuedDigitalData[name]['value'] == item['value']:
-                                        del self.queuedDigitalData[name]
-                                self.currentDigitalData = self.queuedDigitalData
-                                self.queuedDigitalData = {}
+                                    if name in self.queuedRealTimeData and self.queuedRealTimeData[name]['value'] == item['value']:
+                                        del self.queuedRealTimeData[name]
+                                self.currentRealTimeData = self.queuedRealTimeData
+                                self.queuedRealTimeData = {}
                         if data:
                             self.onDataChanged(data)
             except:
-                exception('Monitoring digital sensor changes failed')
-        debug('Monitoring digital sensor changes finished')
-        self.digitalMonitorRunning = False
+                exception('Monitoring real-time changes failed')
+        debug('Monitoring real-time changes finished')
+        self.realTimeMonitorRunning = False
 
     def MonitorSensors(self):
         """Check sensor states for changes"""