Merge pull request #19 from myDevicesIoT/development

Development

Author: jburhenn

myDevices/__init__.py

@@ -1,4 +1,4 @@
 """
 This package contains the Cayenne agent, which is a full featured client for the Cayenne IoT project builder: https://cayenne.mydevices.com. It sends system information as well as sensor and actuator data and responds to actuator messages initiated from the Cayenne dashboard and mobile apps.
 """
-__version__ = '2.0.1'
+__version__ = '2.0.2'

myDevices/cloud/cayennemqtt.py

@@ -65,7 +65,28 @@ def add(data_list, prefix, channel=None, suffix=None, value=None, type=None, uni
         if name is not None:
             data['name'] = name
         data_list.append(data)
-        
+
+    @staticmethod
+    def add_unique(data_list, prefix, channel=None, suffix=None, value=None, type=None, unit=None, name=None):
+        """Create data channel dict and append it to a list if the channel doesn't already exist in the list"""
+        data_channel = prefix
+        if channel is not None:
+            data_channel += ':' + str(channel)
+        if suffix is not None:
+            data_channel += ';' + str(suffix)
+        item = next((item for item in data_list if item['channel'] == data_channel), None)
+        if not item:            
+            data = {}
+            data['channel'] = data_channel
+            data['value'] = value
+            if type is not None:
+                data['type'] = type
+            if unit is not None:
+                data['unit'] = unit
+            if name is not None:
+                data['name'] = name
+            data_list.append(data)
+            
 
 class CayenneMQTTClient:
     """Cayenne MQTT Client class.

myDevices/cloud/client.py

@@ -9,12 +9,11 @@
 from time import strftime, localtime, tzset, time, sleep
 from queue import Queue, Empty
 from myDevices import __version__
-from myDevices.utils.config import Config
+from myDevices.utils.config import Config, APP_SETTINGS, NETWORK_SETTINGS
 from myDevices.utils.logger import exception, info, warn, error, debug, logJson
 from myDevices.sensors import sensors
 from myDevices.system.hardware import Hardware
 from myDevices.cloud.scheduler import SchedulerEngine
-from myDevices.cloud.download_speed import DownloadSpeed
 from myDevices.cloud.updater import Updater
 from myDevices.system.systemconfig import SystemConfig
 from myDevices.utils.daemon import Daemon
@@ -25,10 +24,9 @@
 from myDevices.cloud.apiclient import CayenneApiClient
 import myDevices.cloud.cayennemqtt as cayennemqtt
 
-NETWORK_SETTINGS = '/etc/myDevices/Network.ini'
-APP_SETTINGS = '/etc/myDevices/AppSettings.ini'
 GENERAL_SLEEP_THREAD = 0.20
 
+
 def GetTime():
     """Return string with the current time"""
     tzset()
@@ -203,15 +201,14 @@ def Start(self):
             self.oSInfo = OSInfo()
             self.count = 10000
             self.buff = bytearray(self.count)
-            self.downloadSpeed = DownloadSpeed(self.config)
-            self.downloadSpeed.getDownloadSpeed()
             self.sensorsClient.SetDataChanged(self.OnDataChanged)
             self.writerThread = WriterThread('writer', self)
             self.writerThread.start()
             self.processorThread = ProcessorThread('processor', self)
             self.processorThread.start()
+            self.systemInfo = []
             TimerThread(self.SendSystemInfo, 300)
-            TimerThread(self.SendSystemState, 30, 5)
+            # TimerThread(self.SendSystemState, 30, 5)
             self.updater = Updater(self.config)
             self.updater.start()
             events = self.schedulerEngine.get_scheduled_events()
@@ -244,50 +241,63 @@ def Destroy(self):
 
     def OnDataChanged(self, data):
         """Enqueue a packet containing changed system data to send to the server"""
-        info('Send changed data: {}'.format([{item['channel']:item['value']} for item in data]))
+        try:
+            if len(data) > 15:
+                items = [{item['channel']:item['value']} for item in data if not item['channel'].startswith(cayennemqtt.SYS_GPIO)]
+                info('Send changed data: {} + {}'.format(items, cayennemqtt.SYS_GPIO))
+            else:
+                info('Send changed data: {}'.format([{item['channel']:item['value']} for item in data]))
+            # items = {}
+            # gpio_items = {}
+            # for item in data:
+            #     if not item['channel'].startswith(cayennemqtt.SYS_GPIO):
+            #         items[item['channel']] = item['value']
+            #     else:
+            #         channel = item['channel'].replace(cayennemqtt.SYS_GPIO + ':', '').split(';')
+            #         if not channel[0] in gpio_items:
+            #             gpio_items[channel[0]] = str(item['value'])
+            #         else:
+            #             gpio_items[channel[0]] += ',' + str(item['value'])
+            # info('Send changed data: {}, {}: {}'.format(items, cayennemqtt.SYS_GPIO, gpio_items))
+        except:
+            info('Send changed data')
+            pass
         self.EnqueuePacket(data)
 
     def SendSystemInfo(self):
         """Enqueue a packet containing system info to send to the server"""
         try:
-            data = []
-            cayennemqtt.DataChannel.add(data, cayennemqtt.SYS_OS_NAME, value=self.oSInfo.ID)
-            cayennemqtt.DataChannel.add(data, cayennemqtt.SYS_OS_VERSION, value=self.oSInfo.VERSION_ID)
-            cayennemqtt.DataChannel.add(data, cayennemqtt.AGENT_VERSION, value=self.config.get('Agent', 'Version', __version__))
-            cayennemqtt.DataChannel.add(data, cayennemqtt.SYS_POWER_RESET, value=0)
-            cayennemqtt.DataChannel.add(data, cayennemqtt.SYS_POWER_HALT, value=0)
+            currentSystemInfo = []
+            cayennemqtt.DataChannel.add(currentSystemInfo, cayennemqtt.SYS_OS_NAME, value=self.oSInfo.ID)
+            cayennemqtt.DataChannel.add(currentSystemInfo, cayennemqtt.SYS_OS_VERSION, value=self.oSInfo.VERSION_ID)
+            cayennemqtt.DataChannel.add(currentSystemInfo, cayennemqtt.AGENT_VERSION, value=self.config.get('Agent', 'Version', __version__))
+            cayennemqtt.DataChannel.add(currentSystemInfo, cayennemqtt.SYS_POWER_RESET, value=0)
+            cayennemqtt.DataChannel.add(currentSystemInfo, cayennemqtt.SYS_POWER_HALT, value=0)
             config = SystemConfig.getConfig()
             if config:
                 channel_map = {'I2C': cayennemqtt.SYS_I2C, 'SPI': cayennemqtt.SYS_SPI, 'Serial': cayennemqtt.SYS_UART,
                                 'OneWire': cayennemqtt.SYS_ONEWIRE, 'DeviceTree': cayennemqtt.SYS_DEVICETREE}
                 for key, channel in channel_map.items():
                     try:
-                        cayennemqtt.DataChannel.add(data, channel, value=config[key])
+                        cayennemqtt.DataChannel.add(currentSystemInfo, channel, value=config[key])
                     except:
                         pass
-            info('Send system info: {}'.format([{item['channel']:item['value']} for item in data]))
-            self.EnqueuePacket(data)
+            if currentSystemInfo != self.systemInfo:
+                data = currentSystemInfo
+                if self.systemInfo:
+                    data = [x for x in data if x not in self.systemInfo]
+                if data:
+                    self.systemInfo = currentSystemInfo
+                    info('Send system info: {}'.format([{item['channel']:item['value']} for item in data]))
+                    self.EnqueuePacket(data)
         except Exception:
             exception('SendSystemInfo unexpected error')
 
-    def SendSystemState(self):
-        """Enqueue a packet containing system information to send to the server"""
-        try:
-            data = []
-            download_speed = self.downloadSpeed.getDownloadSpeed()
-            if download_speed:
-                cayennemqtt.DataChannel.add(data, cayennemqtt.SYS_NET, suffix=cayennemqtt.SPEEDTEST, value=download_speed, type='bw', unit='mbps')
-            data += self.sensorsClient.systemData
-            info('Send system state: {} items'.format(len(data)))
-            self.EnqueuePacket(data)
-        except Exception as e:
-            exception('ThreadSystemInfo unexpected error: ' + str(e))
-
     def CheckSubscription(self):
         """Check that an invite code is valid"""
         inviteCode = self.config.get('Agent', 'InviteCode', fallback=None)
         if not inviteCode:
-            error('No invite code found in {}'.format(APP_SETTINGS))
+            error('No invite code found in {}'.format(self.config.path))
             print('Please input an invite code. This can be retrieved from the Cayenne dashboard by adding a new Raspberry Pi device.\n'
                 'The invite code will be part of the script name shown there: rpi_[invitecode].sh.')
             inviteCode = input('Invite code: ')

myDevices/cloud/doupdatecheck.py

@@ -1,6 +1,5 @@
 from myDevices.cloud.updater import Updater
-from myDevices.utils.config import Config
-from myDevices.cloud.client import APP_SETTINGS
+from myDevices.utils.config import Config, APP_SETTINGS
 from myDevices.utils.logger import setInfo
 
 if __name__ == '__main__':

myDevices/devices/digital/gpio.py

@@ -12,8 +12,11 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
+import errno
 import os
 import mmap
+import select
+from threading import Thread
 from time import sleep
 from myDevices.utils.types import M_JSON
 from myDevices.utils.logger import debug, info, error, exception
@@ -37,6 +40,8 @@
 class NativeGPIO(Singleton, GPIOPort):
     IN = 0
     OUT = 1
+    OUT_LOW = 2
+    OUT_HIGH = 3
 
     ASUS_GPIO = 44
 
@@ -63,6 +68,10 @@ def __init__(self):
         self.pinFunctionSet = set()
         self.valueFile = {pin:None for pin in self.pins}
         self.functionFile = {pin:None for pin in self.pins}
+        self.callbacks = {}
+        self.edge_poll = select.epoll()
+        thread = Thread(target=self.pollEdges, daemon=True)
+        thread.start()
         for pin in self.pins:
             # Export the pins here to prevent a delay when accessing the values for the 
             # first time while waiting for the file group to be set
@@ -151,6 +160,7 @@ def close(self):
             self.setFunction(gpio, g["func"])
             if g["value"] >= 0 and self.getFunction(gpio) == self.OUT:
                 self.__digitalWrite__(gpio, g["value"])
+        self.edge_poll.close()
 
     def checkDigitalChannelExported(self, channel):
         if not channel in self.pins:
@@ -170,6 +180,9 @@ def __getFunctionFilePath__(self, channel):
     def __getValueFilePath__(self, channel):
         return "/sys/class/gpio/gpio%s/value" % channel
 
+    def __getEdgeFilePath__(self, channel):
+        return "/sys/class/gpio/gpio%s/edge" % channel
+
     def __checkFilesystemExport__(self, channel):
         #debug("checkExport for channel %d" % channel)
         if not os.path.isdir("/sys/class/gpio/gpio%s" % channel):
@@ -295,11 +308,9 @@ def __setFunction__(self, channel, value):
         self.checkDigitalChannelExported(channel)
         self.checkPostingFunctionAllowed()
         try:
-            if value == self.IN:
-                value = 'in'
-            else:
-                value = 'out'
-            try:               
+            value_dict = {self.IN: 'in', self.OUT: 'out', self.OUT_LOW: 'low', self.OUT_HIGH: 'high'}
+            value = value_dict[value]
+            try:
                 self.functionFile[channel].write(value)
                 self.functionFile[channel].seek(0)
             except:
@@ -324,6 +335,49 @@ def __portWrite__(self, value):
         else:
             raise Exception("Please limit exported GPIO to write integers")
 
+    def setCallback(self, channel, callback, data=None):
+        debug('Set callback for GPIO pin {}'.format(channel))
+        self.__checkFilesystemValue__(channel)
+        with open(self.__getEdgeFilePath__(channel), 'w') as f:
+            f.write('both')
+        self.callbacks[channel] = {'function':callback, 'data':data}
+        try:
+            self.edge_poll.register(self.valueFile[channel], (select.EPOLLPRI | select.EPOLLET))
+        except FileExistsError as e:
+            # Ignore file exists error since it means we already registered the file.
+            pass
+             
+    def removeCallback(self, channel):
+        debug('removeCallback: {}'.format(channel))
+        self.__checkFilesystemValue__(channel)
+        with open(self.__getEdgeFilePath__(channel), 'w') as f:
+            f.write('none')
+        del self.callbacks[channel]
+        self.edge_poll.unregister(self.valueFile[channel])
+
+    def pollEdges(self):
+        while True:
+            try:
+                events = self.edge_poll.poll(1)
+            except IOError as e:
+                if e.errno != errno.EINTR:
+                    error(e)    
+            if len(events) > 0:
+                self.onEdgeEvent(events)
+
+    def onEdgeEvent(self, events):
+        debug('onEdgeEvent: {}'.format(events))
+        for fd, event in events:
+            if not (event & (select.EPOLLPRI | select.EPOLLET)):
+                continue
+            for channel, valueFile in self.valueFile.items():
+                if valueFile and valueFile.fileno() == fd:
+                    value = valueFile.read()
+                    valueFile.seek(0)
+                    callback = self.callbacks[channel]
+                    debug('onEdgeEvent: channel {}, value {}'.format(channel, value))
+                    callback['function'](callback['data'], int(value))
+
     #@request("GET", "*")
     @response(contentType=M_JSON)
     def wildcard(self, compact=False):

myDevices/devices/digital/helper.py

@@ -46,6 +46,14 @@ def setGPIOInstance(self):
             if self.gpio:
                 self.gpio.setFunction(self.channel, GPIO.IN)
 
+    def setCallback(self, callback, data=None):
+        if self.gpioname == "GPIO" and self.__family__() == "DigitalSensor":
+            self.gpio.setCallback(self.channel, callback, data)
+
+    def removeCallback(self):
+        if self.gpioname == "GPIO" and self.__family__() == "DigitalSensor":
+            self.gpio.removeCallback(self.channel)
+
     #@request("GET", "value")
     @response("%d")
     def read(self):
@@ -63,9 +71,17 @@ def __str__(self):
         return "MotionSensor"
 
 class DigitalActuator(DigitalSensor):
-    def __init__(self, gpio, channel, invert=False):
+    def __init__(self, gpio, channel, invert=False, last_state=None):
         DigitalSensor.__init__(self, gpio, channel, invert)
-        self.gpio.setFunction(self.channel, GPIO.OUT)
+        function = GPIO.OUT
+        if gpio == 'GPIO' and last_state is not None:
+            if self.invert:
+                last_state = int(not last_state)
+            if last_state == 1:
+                function = GPIO.OUT_HIGH
+            elif last_state == 0:
+                function = GPIO.OUT_LOW
+        self.gpio.setFunction(self.channel, function)
 
     def __str__(self):
         return "DigitalActuator"
@@ -82,30 +98,29 @@ def write(self, value):
         return self.read()
 
 class LightSwitch(DigitalActuator):        
-    def __init__(self, gpio, channel, invert=False):
-        DigitalActuator.__init__(self, gpio, channel, invert)
+    def __init__(self, gpio, channel, invert=False, last_state=None):
+        DigitalActuator.__init__(self, gpio, channel, invert, last_state)
 
     def __str__(self):
         return "LightSwitch"
 
 class MotorSwitch(DigitalActuator):        
-    def __init__(self, gpio, channel, invert=False):
-        DigitalActuator.__init__(self, gpio, channel, invert)
+    def __init__(self, gpio, channel, invert=False, last_state=None):
+        DigitalActuator.__init__(self, gpio, channel, invert, last_state)
 
     def __str__(self):
         return "MotorSwitch"
 
 class RelaySwitch(DigitalActuator):        
-    def __init__(self, gpio, channel, invert=False):
-        DigitalActuator.__init__(self, gpio, channel, invert)
+    def __init__(self, gpio, channel, invert=False, last_state=None):
+        DigitalActuator.__init__(self, gpio, channel, invert, last_state)
 
     def __str__(self):
         return "RelaySwitch"
 
 class ValveSwitch(DigitalActuator):        
-    def __init__(self, gpio, channel, invert=False):
-        DigitalActuator.__init__(self, gpio, channel, invert)
+    def __init__(self, gpio, channel, invert=False, last_state=None):
+        DigitalActuator.__init__(self, gpio, channel, invert, last_state)
 
     def __str__(self):
         return "ValveSwitch"
-