Do HTTP requests after scheduled jobs run.

Author: jburhenn

myDevices/cloud/cayennemqtt.py

@@ -9,6 +9,7 @@
 COMMAND_TOPIC = 'cmd'
 COMMAND_JSON_TOPIC = 'cmd.json'
 COMMAND_RESPONSE_TOPIC = 'response'
+JOBS_TOPIC = 'jobs.json'
 
 # Data Channels
 SYS_HARDWARE_MAKE = 'sys:hw:make'
@@ -30,7 +31,7 @@
 AGENT_VERSION = 'agent:version'
 AGENT_DEVICES = 'agent:devices'
 AGENT_MANAGE = 'agent:manage'
-AGENT_SCHEDULE = 'agent:schedule'
+AGENT_SCHEDULER = 'agent:scheduler'
 DEV_SENSOR = 'dev'
 
 # Channel Suffixes

myDevices/cloud/client.py

@@ -115,7 +115,7 @@ def run(self):
                 if topic or message:
                     got_packet = True
                 try:
-                    if message:
+                    if message or topic == cayennemqtt.JOBS_TOPIC:
                         # debug('WriterThread, topic: {} {}'.format(topic, message))
                         if not isinstance(message, str):
                             message = dumps(message)
@@ -214,6 +214,8 @@ def Start(self):
             TimerThread(self.SendSystemState, 30, 5)
             self.updater = Updater(self.config)
             self.updater.start()
+            events = self.schedulerEngine.get_scheduled_events()
+            self.EnqueuePacket(events, cayennemqtt.JOBS_TOPIC)
             # self.sentHistoryData = {}
             # self.historySendFails = 0
             # self.historyThread = Thread(target=self.SendHistoryData)
@@ -367,8 +369,12 @@ def OnMessage(self, message):
 
     def RunAction(self, action):
         """Run a specified action"""
-        debug('RunAction')
-        self.ExecuteMessage(action)
+        debug('RunAction: {}'.format(action))
+        result = True
+        command = action.copy()
+        self.mqttClient.transform_command(command)
+        result = self.ExecuteMessage(command)
+        return result
 
     def ProcessMessage(self):
         """Process a message from the server"""
@@ -381,30 +387,36 @@ def ProcessMessage(self):
         self.ExecuteMessage(messageObject)
 
     def ExecuteMessage(self, message):
-        """Execute an action described in a message object"""
+        """Execute an action described in a message object
+        
+        Returns: True if action was executed, False otherwise."""
+        result = False
         if not message:
-            return
+            return result
         channel = message['channel']
         info('ExecuteMessage: {}'.format(message))
         if channel in (cayennemqtt.SYS_POWER_RESET, cayennemqtt.SYS_POWER_HALT):
-            self.ProcessPowerCommand(message)
+            result = self.ProcessPowerCommand(message)
         elif channel.startswith(cayennemqtt.DEV_SENSOR):
-            self.ProcessSensorCommand(message)
+            result = self.ProcessSensorCommand(message)
         elif channel.startswith(cayennemqtt.SYS_GPIO):
-            self.ProcessGpioCommand(message)
+            result = self.ProcessGpioCommand(message)
         elif channel == cayennemqtt.AGENT_DEVICES:
-            self.ProcessDeviceCommand(message)
+            result = self.ProcessDeviceCommand(message)
         elif channel in (cayennemqtt.SYS_I2C, cayennemqtt.SYS_SPI, cayennemqtt.SYS_UART, cayennemqtt.SYS_ONEWIRE):
-            self.ProcessConfigCommand(message)
+            result = self.ProcessConfigCommand(message)
         elif channel == cayennemqtt.AGENT_MANAGE:
-            self.ProcessAgentCommand(message)
-        elif channel == cayennemqtt.AGENT_SCHEDULE:
-            self.ProcessScheduleCommand(message)
+            result = self.ProcessAgentCommand(message)
+        elif channel == cayennemqtt.AGENT_SCHEDULER:
+            result = self.ProcessSchedulerCommand(message)
         else:
             info('Unknown message')
+        return result
 
     def ProcessPowerCommand(self, message):
-        """Process command to reboot/shutdown the system"""
+        """Process command to reboot/shutdown the system
+        
+        Returns: True if command was processed, False otherwise."""
         error_message = None
         try:
             self.EnqueueCommandResponse(message, error_message)
@@ -427,9 +439,13 @@ def ProcessPowerCommand(self, message):
             data = []
             cayennemqtt.DataChannel.add(data, message['channel'], value=0)
             self.EnqueuePacket(data)
+            raise ExecuteMessageError(error_message)
+        return error_message == None
 
     def ProcessAgentCommand(self, message):
-        """Process command to manage the agent state"""
+        """Process command to manage the agent state
+        
+        Returns: True if command was processed, False otherwise."""
         error = None
         try:
             if message['suffix'] == 'uninstall':
@@ -450,9 +466,14 @@ def ProcessAgentCommand(self, message):
         except Exception as ex:
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
+        if error:
+            raise ExecuteMessageError(error)
+        return error == None
 
     def ProcessConfigCommand(self, message):
-        """Process system configuration command"""
+        """Process system configuration command
+        
+        Returns: True if command was processed, False otherwise."""
         error = None
         try:
             value = 1 - int(message['payload']) #Invert the value since the config script uses 0 for enable and 1 for disable
@@ -464,9 +485,12 @@ def ProcessConfigCommand(self, message):
         except Exception as ex:
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
-    
+        return error == None
+
     def ProcessGpioCommand(self, message):
-        """Process GPIO command"""
+        """Process GPIO command
+        
+        Returns: True if command was processed, False otherwise."""
         error = None
         try:
             channel = int(message['channel'].replace(cayennemqtt.SYS_GPIO + ':', ''))
@@ -477,9 +501,12 @@ def ProcessGpioCommand(self, message):
         except Exception as ex:
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
+        return error == None
 
     def ProcessSensorCommand(self, message):
-        """Process sensor command"""
+        """Process sensor command
+        
+        Returns: True if command was processed, False otherwise."""
         error = None
         try:
             sensor_info = message['channel'].replace(cayennemqtt.DEV_SENSOR + ':', '').split(':')
@@ -494,9 +521,12 @@ def ProcessSensorCommand(self, message):
         except Exception as ex:
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
+        return error == None
 
     def ProcessDeviceCommand(self, message):
-        """Process a device command to add/edit/remove a sensor"""
+        """Process a device command to add/edit/remove a sensor
+        
+        Returns: True if command was processed, False otherwise."""
         error = None
         try:
             payload = message['payload']
@@ -515,28 +545,32 @@ def ProcessDeviceCommand(self, message):
         except Exception as ex:
             error = '{}: {}'.format(type(ex).__name__, ex)
         self.EnqueueCommandResponse(message, error)
+        return error == None
 
-    def ProcessScheduleCommand(self, message):
-        """Process command to add/edit/remove a scheduled action"""
+    def ProcessSchedulerCommand(self, message):
+        """Process command to add/edit/remove a scheduled action
+        
+        Returns: True if command was processed, False otherwise."""
         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_event(message['payload'], True)
             elif message['suffix'] == 'edit':
                 result = self.schedulerEngine.update_scheduled_event(message['payload'])
             elif message['suffix'] == 'delete':
                 result = self.schedulerEngine.remove_scheduled_event(message['payload'])
+            elif message['suffix'] == 'get':
+                events = self.schedulerEngine.get_scheduled_events()
+                self.EnqueuePacket(events, cayennemqtt.JOBS_TOPIC)
             else:
                 error = 'Unknown schedule command: {}'.format(message['suffix'])
-            debug('ProcessScheduleCommand result: {}'.format(result))
+            debug('ProcessSchedulerCommand 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)
+        return error == None
 
     def EnqueueCommandResponse(self, message, error):
         """Send response after processing a command message"""

myDevices/cloud/scheduler.py

@@ -5,6 +5,7 @@
 from time import sleep
 
 import myDevices.schedule as schedule
+from myDevices.requests_futures.sessions import FuturesSession
 from myDevices.utils.logger import debug, error, exception, info, logJson, setDebug, warn
 
 
@@ -82,13 +83,13 @@ def update_scheduled_event(self, event):
                 try:
                     old_item = self.schedule_items[event['id']]
                     schedule.cancel_job(old_item['job'])
+                    result = self.create_job(schedule_item)
+                    debug('Update scheduled event result: {}'.format(result))
+                    if result == True:
+                        self.update_database_record(event['id'], event)
+                        self.schedule_items[event['id']] = schedule_item
                 except KeyError:
                     debug('Old schedule with id = {} not found'.format(event['id']))
-                result = self.create_job(schedule_item)
-                debug('Update scheduled event result: {}'.format(result))
-                if result == True:
-                    self.update_database_record(event['id'], event)
-                    self.schedule_items[event['id']] = schedule_item
         except:
             exception('Failed to update scheduled event')
         return result
@@ -122,6 +123,12 @@ def get_scheduled_events(self):
         try:
             with self.mutex:
                 events = [schedule_item['event'] for schedule_item in self.schedule_items.values()]
+                for event in events:
+                    try:
+                        # Don't include last run value that is only used locally.
+                        del event['last_run']
+                    except:
+                        pass
         except:
             exception('Failed to get scheduled events')
         return events
@@ -178,8 +185,8 @@ def create_job(self, schedule_item):
                         schedule_item['job'] = schedule.every(config['interval'], config['start_date']).months.at(config['start_date'])
                     if config['unit'] == 'year':
                         schedule_item['job'] = schedule.every(config['interval'], config['start_date']).years.at(config['start_date'])
-                if 'last_run' in config:
-                    schedule_item['job'].set_last_run(config['last_run'])
+                if 'last_run' in schedule_item['event']:
+                    schedule_item['job'].set_last_run(schedule_item['event']['last_run'])
                 schedule_item['job'].do(self.run_scheduled_item, schedule_item)
         except:
             exception('Failed setting up scheduler')
@@ -197,17 +204,43 @@ def run_scheduled_item(self, schedule_item):
         result = True
         event = schedule_item['event']
         config = event['config']
-        config['last_run'] = datetime.strftime(datetime.utcnow(), '%Y-%m-%d %H:%M')
+        event['last_run'] = datetime.strftime(datetime.utcnow(), '%Y-%m-%d %H:%M')
         with self.mutex:
             self.update_database_record(event['id'], event)
+        action_executed = False
         for action in event['actions']:
             info('Executing scheduled action: {}'.format(action))
-            result = self.client.RunAction(action)
+            result = self.client.RunAction(action)           
             if result == False:
                 error('Failed to execute action: {}'.format(action))
+            else:
+                action_executed = True
         if config['type'] == 'date' and result == True:
             with self.mutex:
                 schedule.cancel_job(schedule_item['job'])
+        if action_executed and 'http_push' in event:
+            info('Scheduler making HTTP request')
+            http_push = event['http_push']
+            try:
+                future = None
+                session = FuturesSession(max_workers=1)
+                session.headers = http_push['headers']
+                if http_push['method'] == 'GET':
+                    future = session.get(http_push['url'])
+                if http_push['method'] == 'POST':
+                    future = session.post(http_push['url'], dumps(http_push['payload']))
+                if http_push['method'] == 'PUT':
+                    future = session.put(http_push['url'], dumps(http_push['payload']))
+                if http_push['method'] == 'DELETE':
+                    future = session.delete(http_push['url'])
+            except Exception as ex:
+                error('Scheduler HTTP request exception: {}'.format(ex))
+                return None
+            try:
+                response = future.result(30)
+                info('Scheduler HTTP response: {}'.format(response))
+            except:
+                pass
 
     def add_database_record(self, id, event):
         """Add a scheduled event to the database