网关场景/编辑场景联动插座

Ansjer/Config/gatewaySensorConfig.py

@@ -10,6 +10,7 @@ SMART_SCENE_TOPIC = 'loocam/gateway_sensor/smart_scene/{}'
 GET_SCENE_TOPIC = 'loocam/gateway_sensor/get_scene/{}'
 SUB_DEVICE_TOPIC = 'loocam/gateway_sensor/sub_device/{}'
 VOICE_AUDITION_TOPIC = 'loocam/gateway_sensor/voice_audition/{}'
+SMART_SOCKET_TOPIC = 'loocam/smart-socket/{}'
 
 # 智能场景事件
 SCENE_EVENT_CREATE = 1
@@ -22,6 +23,13 @@ SCENE_EVENT_SOS = 5
 SCENE_STATUS_ON = 1
 SCENE_STATUS_OFF = 0
 
+# 设备类型
+DEVICE_TYPE = {
+    'gateway': 200,
+    'socket': 201,
+    'switch': 202
+}
+
 # 传感器类型
 SENSOR_TYPE = {
     'door_magnet': 215,
@@ -67,4 +75,7 @@ EVENT_TYPE = {
     'temperature': 2200,
     'humidity': 2201,
     'tem_hum_sensor_low_power': 2202,
+    # 插座电源
+    'socket_power_on': 2010,
+    'socket_power_off': 2011,
 }

Controller/SensorGateway/SmartSceneController.py

@@ -14,9 +14,10 @@ from django.views import View
 
 from Ansjer.Config.gatewaySensorConfig import SMART_SCENE_TOPIC, SENSOR_TYPE, EVENT_TYPE, SCENE_EVENT_CREATE, \
     SCENE_EVENT_EDIT, SCENE_EVENT_DELETE, SCENE_STATUS_ON, SCENE_STATUS_OFF, SCENE_EVENT_EDIT_STATUS, \
-    VOICE_AUDITION_TOPIC
+    VOICE_AUDITION_TOPIC, SMART_SOCKET_TOPIC, DEVICE_TYPE
 from Model.models import FamilyRoomDevice, GatewaySubDevice, FamilyRoom, SmartScene, EffectiveTime, Device_Info, \
-    SceneLog
+    SceneLog, GatewayPush
+from Object.ApschedulerObject import ApschedulerObject
 from Object.ResponseObject import ResponseObject
 from Service.CommonService import CommonService
 
@@ -192,7 +193,7 @@ class SmartSceneView(View):
         @param user_id: 用户id
         @return: res
         """
-        device_info_qs = Device_Info.objects.filter(userID_id=user_id, Type__in=[201]).\
+        device_info_qs = Device_Info.objects.filter(userID_id=user_id, Type__in=[DEVICE_TYPE['socket']]).\
             values('id', 'NickName', 'Type', 'serial_number')
         if device_info_qs.exists():
             for device_info in device_info_qs:
@@ -251,11 +252,20 @@ class SmartSceneView(View):
             if smart_scene_qs.exists():
                 return response.json(179)
 
+            # 从gateway_push表查询时区
+            gateway_push_qs = GatewayPush.objects.filter(user_id=user_id).values('tz')
+            if gateway_push_qs.exists():
+                # 截掉.00然后转为浮点型
+                tz = float(gateway_push_qs[0]['tz'][:-3])
+            else:
+                tz = 0
+
             smart_scene_dict = {
                 'user_id': user_id,
                 'scene_name': scene_name,
                 'conditions': conditions,
                 'tasks': tasks,
+                'tz': tz,
                 'created_time': now_time,
                 'updated_time': now_time,
             }
@@ -279,7 +289,7 @@ class SmartSceneView(View):
                 serial_number = device_info_qs[0]['serial_number']
 
                 # 网关数据
-                msg['sensor_type'] = 200
+                msg['sensor_type'] = DEVICE_TYPE['gateway']
                 msg['sensor_status'] = 2002
                 msg['sensor_ieee_addr'] = 'FFFFFFFFFFFFFFFF'
 
@@ -370,9 +380,11 @@ class SmartSceneView(View):
                 msg['scene_id'] = smart_scene_qs.id
 
                 # 获取设备任务数据
-                msg['task'] = cls.get_msg_task_list(tasks_list)
+                msg['task'], mqtt_tasks = cls.get_msg_task_list(tasks_list, conditions_dict, tz, now_time)
 
                 smart_scene_qs.device_data = json.dumps(msg)
+                if mqtt_tasks:
+                    smart_scene_qs.mqtt_tasks = json.dumps(mqtt_tasks)
                 smart_scene_qs.save()
                 # 发布MQTT消息通知网关设备
                 thing_name = serial_number
@@ -624,6 +636,8 @@ class SmartSceneView(View):
             if not smart_scene_qs.exists():
                 return response.json(173)
 
+            tz = smart_scene_qs[0].tz
+            mqtt_tasks_list = eval(smart_scene_qs[0].mqtt_tasks)
             scene_status = 1 if smart_scene_qs[0].is_enable else 0
             msg = {
                 'scene_id': smart_scene_id,
@@ -677,15 +691,17 @@ class SmartSceneView(View):
                     return response.json(173)
                 serial_number = device_qs[0]['serial_number']
                 # 网关数据
-                msg['sensor_type'] = 200
+                msg['sensor_type'] = DEVICE_TYPE['gateway']
                 msg['sensor_status'] = 2002
                 msg['sensor_ieee_addr'] = 'FFFFFFFFFFFFFFFF'
 
             # 获取设备任务数据
-            msg['task'] = cls.get_msg_task_list(tasks_list)
+            msg['task'], mqtt_tasks = cls.get_msg_task_list(tasks_list, conditions_dict, tz, now_time, mqtt_tasks_list)
+            if mqtt_tasks:
+                mqtt_tasks = json.dumps(mqtt_tasks)
 
             with transaction.atomic():
-                smart_scene_qs.update(scene_name=scene_name, conditions=conditions, tasks=tasks,
+                smart_scene_qs.update(scene_name=scene_name, conditions=conditions, tasks=tasks, mqtt_tasks=mqtt_tasks,
                                       device_data=json.dumps(msg), updated_time=now_time, device_id=device_id,
                                       sub_device_id=sub_device_id)
 
@@ -1017,43 +1033,76 @@ class SmartSceneView(View):
         except Exception as e:
             return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
 
-    @staticmethod
-    def get_msg_task_list(tasks_list):
+    @classmethod
+    def get_msg_task_list(cls, tasks_list, conditions_dict, tz, now_time, mqtt_tasks_list=None):
         """
         获取设备任务数据
         @param tasks_list: app任务列表
-        @return: task_list
+        @param conditions_dict: 条件
+        @param tz: 时区
+        @param now_time: 当前时间
+        @param mqtt_tasks_list: mqtt定时任务时列表
+        @return: task_list, mqtt_task_list
         """
+        # 传入任务列表,删除旧任务
+        if mqtt_tasks_list is not None:
+            cls.del_aps_job(mqtt_tasks_list)
         task_list = []
+        mqtt_task_list = []
         for task in tasks_list:
             sensor_type = int(task['device_type'])
-            task_temp = {
-                'sensor_type': sensor_type,
-                'sensor_delay': 0
-            }
+            # 处理插座数据
+            # 不用添加到设备的任务列表,添加到mqtt任务列表
+            if sensor_type == DEVICE_TYPE['socket']:
+                serial_number = task['serial_number']
+                event_type = int(task['event_type'])
+                delay_time = task['delay_time']
+                task_temp = {
+                    'device_type': sensor_type,
+                    'event_type': event_type,
+                    'delay_time': delay_time,
+                    'serial_number': serial_number
+                }
 
-            # 延时
-            if 'delay_time' in task and task['delay_time'] != 0:
-                task_temp['sensor_delay'] = task['delay_time']
-
-            # 不为-1时需要其他数据
-            if sensor_type != -1:
-                task_temp['sensor_action'] = int(task['event_type'])
-                # 子设备返回长地址
-                sub_device_id = task.get('subDeviceId', None)
-                if sub_device_id:
-                    sub_device_qs = GatewaySubDevice.objects.filter(id=sub_device_id).values('ieee_addr').first()
-                    task_temp['sensor_ieee_addr'] = sub_device_qs['ieee_addr']
-                # 网关添加报警类型数据
-                else:
-                    task_temp['voice_type'] = task.get('voice_type')
-                    task_temp['voice_id'] = task.get('voice_id')
-                    task_temp['count'] = task.get('count')
-                    task_temp['delay_time'] = task.get('delay_time')
-                    task_temp['duration'] = task.get('duration')
-                    task_temp['value_type'] = task.get('value_type')
-            task_list.append(task_temp)
-        return task_list
+                # 如果条件为设置时间,创建或修改定时任务
+                if conditions_dict['type'] == 1:
+                    minutes = conditions_dict['time']['minutes']
+                    repeat = conditions_dict['time']['repeat']
+
+                    task_temp['task_id'], task_temp['time_dict'] = cls.create_aps_job(
+                        minutes, delay_time, tz, now_time, repeat, sensor_type, event_type, serial_number)
+                mqtt_task_list.append(task_temp)
+            else:
+                task_temp = {
+                    'sensor_type': sensor_type,
+                    'sensor_delay': 0
+                }
+
+                # 延时
+                if 'delay_time' in task and task['delay_time'] != 0:
+                    task_temp['sensor_delay'] = task['delay_time']
+
+                # 不为-1时需要其他数据
+                if sensor_type != -1:
+                    task_temp['sensor_action'] = int(task['event_type'])
+                    # 子设备返回长地址
+                    sub_device_id = task.get('subDeviceId', None)
+                    if sub_device_id:
+                        sub_device_qs = GatewaySubDevice.objects.filter(id=sub_device_id).values('ieee_addr').first()
+                        task_temp['sensor_ieee_addr'] = sub_device_qs['ieee_addr']
+                    # 网关添加报警类型数据
+                    else:
+                        task_temp['voice_type'] = task.get('voice_type')
+                        task_temp['voice_id'] = task.get('voice_id')
+                        task_temp['count'] = task.get('count')
+                        task_temp['delay_time'] = task.get('delay_time')
+                        task_temp['duration'] = task.get('duration')
+                        task_temp['value_type'] = task.get('value_type')
+                task_list.append(task_temp)
+        # 列表为空保存为空字符串
+        if not mqtt_task_list:
+            mqtt_task_list = ''
+        return task_list, mqtt_task_list
 
     @staticmethod
     def time_conflict(sub_device_id, conditions, is_all_day, request_dict, smart_scene_id=None):
@@ -1117,6 +1166,123 @@ class SmartSceneView(View):
                                 return True
             return False
 
+    @classmethod
+    def create_aps_job(cls, minutes, delay_time, tz, now_time, repeat, device_type, event_type, serial_number):
+        """
+        创建定时任务
+        返回任务id和时间
+        @param minutes: 分钟时间
+        @param delay_time: 延迟时间
+        @param tz: 时区
+        @param now_time: 当前时间
+        @param repeat: 星期周期的十进制数
+        @param device_type: 设备类型
+        @param event_type: 事件类型
+        @param serial_number: 序列号
+        @return: task_id, time_dict
+        """
+        task_id = serial_number + '_'
+        apscheduler_obj = ApschedulerObject()
+        # 一次性任务
+        if repeat == 0:
+            # 根据时间戳和时区获取年月日,拼接由分钟转换出来的时间
+            time_string = CommonService.get_date_from_timestamp(now_time, tz)
+            hour, minute = divmod(minutes, 60)
+            time_string += ' {:02d}:{:02d}:00'.format(hour, minute)
+            time_stamp = CommonService.convert_to_timestamp(tz, time_string)
+            # 加上延时,如果执行时间小于当前时间,延迟24小时执行
+            time_stamp += delay_time
+            if time_stamp < now_time:
+                time_stamp += 24 * 60 * 60
+            task_id += str(time_stamp)
+            apscheduler_obj.create_date_job(func=cls.pub_mqtt, task_id=task_id, time_stamp=time_stamp,
+                                            args=(device_type, event_type, serial_number))
+            time_dict = {'time_stamp': time_stamp}
+        # 周期任务
+        else:
+            hour, minute, second, is_next_day = cls.handle_delay_time(minutes, delay_time)
+            # 加上延时,如果执行时间超过23:59,隔天执行
+            weeks = cls.int_to_weeks(repeat, is_next_day)
+            time_str = weeks + '_{:02d}{:02d}{:02d}'.format(hour, minute, second)
+            task_id += time_str
+            apscheduler_obj.create_cron_job(func=cls.pub_mqtt, task_id=task_id, day_of_week=weeks,
+                                            hour=hour, minute=minute,
+                                            args=(device_type, event_type, serial_number))
+            time_dict = {'weeks': weeks, 'hour': hour, 'minute': minute, 'second': second}
+        return task_id, time_dict
+
+    @staticmethod
+    def handle_delay_time(minutes, delay_time):
+        """
+        处理延迟时间
+        @param minutes: 时间分钟数,如1439代表23:59
+        @param delay_time: 延迟时间,单位:秒
+        @return: hour, minute, second, is_next_day
+        """
+        is_next_day = False
+        hour, minute = divmod(minutes, 60)
+        # 延迟时间转为分钟,如果加上时间分钟数大于1439,隔天执行
+        minute, second = divmod(delay_time, 60)
+        total_minutes = minutes + minute
+        ex_min = total_minutes - 1439
+        if ex_min <= 0:
+            hour, minute = divmod(total_minutes, 60)
+        else:
+            hour, minute = divmod(ex_min, 60)
+            is_next_day = True
+        return hour, minute, second, is_next_day
+
+    @staticmethod
+    def int_to_weeks(repeat, is_next_day):
+        """
+        十进制转星期周期
+        @param repeat: 星期周期的十进制数,如127 -> 0,1,2,3,4,5,6
+        @param is_next_day: 是否隔天
+        @return: weeks
+        """
+        # 十进制转为7位的二进制,低六位倒序
+        bin_str = bin(repeat)[2:].zfill(7)
+        bin_str = bin_str[:1] + bin_str[-6:][::-1]
+        # 生成星期周期字符串
+        weeks = ''
+        next_day = 1 if is_next_day else 0
+        for i, bit in enumerate(bin_str):
+            if bit == '1':
+                weeks += str(i+next_day) + ','
+        # 删除最后一个逗号并返回结果
+        return weeks[:-1]
+
+    @staticmethod
+    def pub_mqtt(device_type, event_type, serial_number):
+        """
+        发布mqtt消息
+        @param device_type: 设备类型
+        @param event_type: 事件类型
+        @param serial_number: 序列号
+        @return:
+        """
+        if device_type == DEVICE_TYPE['socket']:
+            topic_name = SMART_SOCKET_TOPIC.format(serial_number)
+            status = 1 if event_type == EVENT_TYPE['socket_power_on'] else 0
+            msg = {
+                'type': 1,
+                'data': {'deviceSwitch': status}
+            }
+            CommonService.req_publish_mqtt_msg(serial_number, topic_name, msg)
+
+    @staticmethod
+    def del_aps_job(mqtt_tasks_list):
+        """
+        删除定时任务
+        @param mqtt_tasks_list: mqtt任务列表
+        @return:
+        """
+        apscheduler_obj = ApschedulerObject()
+        for mqtt_task in mqtt_tasks_list:
+            task_id = mqtt_task['task_id']
+            apscheduler_obj.del_job(task_id)
+
+
 #
 #                   ___====-_  _-====___
 #             _--^^^#####//      \\#####^^^--_

Object/ApschedulerObject.py

@@ -16,8 +16,9 @@ class ApschedulerObject:
         now_time = time.time()
         print('hello world:[{}]'.format(now_time))
 
-    def create_cron_job(self, func, task_id, day_of_week, hour, minute, args):  # 周期任务
+    def create_cron_job(self, func, task_id, day_of_week, hour, minute, args, second=0):  # 周期任务
         job = self.scheduler.add_job(func=func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
+                                     second=second,
                                      replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args,
                                      misfire_grace_time=300)
         print(job)
@@ -43,3 +44,9 @@ class ApschedulerObject:
     @staticmethod
     def del_job(task_id):  # 删除任务
         DjangoJob.objects.filter(id=task_id).delete()
+
+    def pause_job(self, task_id):  # 暂停任务
+        self.scheduler.pause_job(task_id)
+
+    def resume_job(self, task_id):  # 恢复任务
+        self.scheduler.resume_job(task_id)

Service/CommonService.py

@@ -320,6 +320,16 @@ class CommonService:
         else:
             return time.strptime(format)
 
+    @staticmethod
+    def get_date_from_timestamp(timestamp, timezone_offset):
+        # 创建时区对象
+        tz = datetime.timezone(datetime.timedelta(hours=timezone_offset))
+        # 使用时间戳创建 datetime 对象
+        dt = datetime.datetime.fromtimestamp(timestamp, tz)
+        # 格式化成 '%Y-%m-%d'
+        formatted_date = dt.strftime('%Y-%m-%d')
+        return formatted_date
+
     # 计算N个月后的时间戳
     @staticmethod
     def calcMonthLater(addMonth, unix_timestamp=None):