Add support for scheduler commands.

Author: jburhenn

myDevices/cloud/cayennemqtt.py

@@ -30,6 +30,7 @@
 AGENT_VERSION = 'agent:version'
 AGENT_DEVICES = 'agent:devices'
 AGENT_MANAGE = 'agent:manage'
+AGENT_SCHEDULE = 'agent:schedule'
 DEV_SENSOR = 'dev'
 
 # Channel Suffixes
@@ -68,7 +69,7 @@ def add(data_list, prefix, channel=None, suffix=None, value=None, type=None, uni
 class CayenneMQTTClient:
     """Cayenne MQTT Client class.
     
-    This is the main client class for connecting to Cayenne and sending and recFUeiving data.
+    This is the main client class for connecting to Cayenne and sending and receiving data.
     
     Standard usage:
     * Set on_message callback, if you are receiving data.
@@ -150,6 +151,26 @@ def disconnect_callback(self, client, userdata, rc):
                 print("Reconnect failed, retrying")
                 time.sleep(5)
 
+    def transform_command(self, command, payload=[], channel=[]):
+        """Transform a command message into an object.
+
+        command is the command object that will be transformed in place.
+        payload is an optional list of payload data items.
+        channel is an optional list containing channel and suffix data.
+        """
+        if not payload:
+            command['payload'] = command.pop('value')
+            channel = command['channel'].split('/')[-1].split(';')
+        else:
+            if len(payload) > 1:
+                command['cmdId'] = payload[0]
+                command['payload'] = payload[1]
+            else:
+                command['payload'] = payload[0]
+        command['channel'] = channel[0]
+        if len(channel) > 1:
+            command['suffix'] = channel[1]
+
     def message_callback(self, client, userdata, msg):
         """The callback for when a message is received from the server.
 
@@ -160,21 +181,10 @@ def message_callback(self, client, userdata, msg):
         try:
             message = {}
             if msg.topic[-len(COMMAND_JSON_TOPIC):] == COMMAND_JSON_TOPIC:
-                payload = loads(msg.payload.decode())
-                message['payload'] = payload['value']
-                message['cmdId'] = payload['cmdId']
-                channel = payload['channel'].split('/')[-1].split(';')
+                message = loads(msg.payload.decode())
+                self.transform_command(message)
             else:
-                payload = msg.payload.decode().split(',')
-                if len(payload) > 1:
-                    message['cmdId'] = payload[0]
-                    message['payload'] = payload[1]
-                else:
-                    message['payload'] = payload[0]
-                channel = msg.topic.split('/')[-1].split(';')
-            message['channel'] = channel[0]
-            if len(channel) > 1:
-                message['suffix'] = channel[1]
+                self.transform_command(message, msg.payload.decode().split(','), msg.topic.split('/')[-1].split(';'))
             debug('message_callback: {}'.format(message))
             if self.on_message:
                 self.on_message(message)

myDevices/cloud/client.py

@@ -13,7 +13,7 @@
 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.scheduler import SchedulerEngine
 from myDevices.cloud.download_speed import DownloadSpeed
 from myDevices.cloud.updater import Updater
 from myDevices.system.systemconfig import SystemConfig
@@ -195,7 +195,7 @@ def Start(self):
             if not self.Connect():
                 error('Error starting agent')
                 return
-            # self.schedulerEngine = SchedulerEngine(self, 'client_scheduler')
+            self.schedulerEngine = SchedulerEngine(self, 'client_scheduler')
             self.sensorsClient = sensors.SensorsClient()
             self.readQueue = Queue()
             self.writeQueue = Queue()
@@ -398,6 +398,8 @@ def ExecuteMessage(self, message):
             self.ProcessConfigCommand(message)
         elif channel == cayennemqtt.AGENT_MANAGE:
             self.ProcessAgentCommand(message)
+        elif channel == cayennemqtt.AGENT_SCHEDULE:
+            self.ProcessScheduleCommand(message)
         else:
             info('Unknown message')
 
@@ -514,8 +516,33 @@ def ProcessDeviceCommand(self, message):
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
 
+    def ProcessScheduleCommand(self, message):
+        """Process command to add/edit/remove a scheduled action"""
+        error = None
+        try:
+            if 'actions' in message['payload']:
+                for action in message['payload']['actions']:
+                    self.mqttClient.transform_command(action)
+            if message['suffix'] == 'add':
+                result = self.schedulerEngine.add_scheduled_item(message['payload'], True)
+            elif message['suffix'] == 'edit':
+                result = self.schedulerEngine.update_scheduled_item(message['payload'])
+            elif message['suffix'] == 'delete':
+                result = self.schedulerEngine.remove_scheduled_item(message['payload'])
+            else:
+                error = 'Unknown schedule command: {}'.format(message['suffix'])
+            debug('ProcessScheduleCommand result: {}'.format(result))
+            if result is False:
+                error = 'Schedule command failed'
+        except Exception as ex:
+            error = '{}: {}'.format(type(ex).__name__, ex)
+        self.EnqueueCommandResponse(message, error)
+
     def EnqueueCommandResponse(self, message, error):
         """Send response after processing a command message"""
+        if not hasattr(message, 'cmdId'):
+            # If there is no command idea we assume this is a scheduled command and don't send a response message.
+            return
         debug('EnqueueCommandResponse error: {}'.format(error))
         if error:
             response = 'error,{}={}'.format(message['cmdId'], error)