优化定时器

Controller/SensorGateway/SmartSwitchController.py

@@ -12,10 +12,10 @@ import time
 
 from django.views import View
 
-from Model.models import SwitchInfo, SwitchDimmingSettings, SwitchChronopher, Device_Info, SceneLog, FamilyRoomDevice
+from Model.models import SwitchInfo, SwitchDimmingSettings, SwitchChronopher, Device_Info, SceneLog, FamilyRoomDevice, \
+    iotdeviceInfoModel
 from Service.CommonService import CommonService
 from Object.ApschedulerObject import ApschedulerObject
-from Object.RedisObject import RedisObject
 from django.db import transaction
 from Ansjer.config import LOGGER
 
@@ -297,6 +297,9 @@ class SmartSwitchView(View):
                 # 设置定时任务
                 serial_number = device_qs[0]['serial_number']
                 topic_name = APSCHEDULER_TOPIC_NAME.format(serial_number)
+                endpoint, token, signature = SmartSwitchView.get_mqtt(serial_number)
+                if not all([endpoint, token, signature]):
+                    return response.json(173)
                 if time_type_radio == 1:
                     task_id = 'switchchronopher_{}'.format(chronopher_id)
                     if time_point_device_will_doing in ['0', '1']:  # 开启或关闭
@@ -313,8 +316,10 @@ class SmartSwitchView(View):
                             'slowTime': slow_open_or_close_speed
                         }
                     time_str = datetime.datetime.fromtimestamp(int(time_point))
+                    # apscheduler_obj.create_cron_job('server:send_mqtt', task_id, repeat, time_str.hour,
+                    #                                 time_str.minute, [topic_name, msg, endpoint, token, signature])
                     apscheduler_obj.create_cron_job(SmartSwitchView.send_mqtt, task_id, repeat, time_str.hour,
-                                                    time_str.minute, (serial_number, topic_name, msg, task_id))
+                                                    time_str.minute, [serial_number, topic_name, msg, task_id])
                 else:
                     start_hour = int(time_quantum_start_time / 60 // 60)
                     start_minute = int(time_quantum_start_time / 60 % 60)
@@ -326,11 +331,11 @@ class SmartSwitchView(View):
                         msg = {"taskId": chronopher_id,
                                "deviceSwitch": int(time_quantum_device_will_doing)}
                         apscheduler_obj.create_cron_job(SmartSwitchView.send_mqtt, begin_task_id, repeat, start_hour,
-                                                        start_minute, (serial_number, topic_name, msg, begin_task_id))
+                                                        start_minute, [serial_number, topic_name, msg, begin_task_id])
                         msg = {"taskId": chronopher_id,
                                "deviceSwitch": 0 if int(time_quantum_device_will_doing) == 1 else 1}
                         apscheduler_obj.create_cron_job(SmartSwitchView.send_mqtt, end_task_id, repeat, end_hour,
-                                                        end_minute, (serial_number, topic_name, msg, end_task_id))
+                                                        end_minute, [serial_number, topic_name, msg, end_task_id])
 
                     else:  # 间隔任务
                         minute = int(time_quantum_device_will_doing)
@@ -344,7 +349,7 @@ class SmartSwitchView(View):
                             hour = '{}-{}'.format(start_hour, end_hour)
                             minute = '{}/{}'.format(start_minute, minute)
                         apscheduler_obj.create_cron_job(SmartSwitchView.send_mqtt, task_id, repeat, hour, minute,
-                                                        (serial_number, topic_name, msg, task_id))
+                                                        [serial_number, topic_name, msg, task_id])
                 return response.json(0)
         except Exception as e:
             print(e)
@@ -393,6 +398,20 @@ class SmartSwitchView(View):
                                                                          int(time.time()), task_id,
                                                                          threading.get_ident(), os.getpid()))
 
+    @staticmethod
+    def get_mqtt(identification_code):
+        if identification_code.endswith('11L'):
+            thing_name = 'LC_' + identification_code
+        else:
+            thing_name = 'Ansjer_Device_' + identification_code
+        iot = iotdeviceInfoModel.objects.filter(thing_name=thing_name).values('endpoint', 'token_iot_number')
+        if not iot.exists():
+            return None, None, None
+        endpoint = iot[0]['endpoint']
+        token = iot[0]['token_iot_number']
+        signature = CommonService.rsa_sign(token)  # Token签名
+        return endpoint, token, signature
+
     @staticmethod
     def create_log(request_dict, response):
         """

Model/models.py

@@ -4182,7 +4182,7 @@ class SwitchChronopher(models.Model):
     time_quantum_device_will_doing = models.SmallIntegerField(default=0,
                                                               verbose_name='设备将会')  # 0: 开启, 1: 关闭, x: 开启/关闭切换间隔(分钟)
     slow_open_or_close_speed = models.SmallIntegerField(default=0, verbose_name='缓慢开/关速度')  # 秒
-    repeat = models.CharField(default=0, max_length=11, verbose_name=u'重复周期')  # 0-6:星期天到星期六
+    repeat = models.CharField(default=0, max_length=11, verbose_name=u'重复周期')  # 0-6:星期一到星期日
 
     class Meta:
         db_table = 'switch_chronopher'

Object/ApschedulerObject.py

@@ -4,13 +4,19 @@ from django_apscheduler.jobstores import DjangoJobStore
 from Ansjer.config import LOGGER
 from django_apscheduler.models import DjangoJob
 import datetime
+import rpyc
+from apscheduler.executors.pool import ProcessPoolExecutor
 
 
 class ApschedulerObject:
     def __init__(self):
-        self.scheduler = BackgroundScheduler()
+        self.scheduler = BackgroundScheduler(
+            executors={
+                'default': ProcessPoolExecutor(1)  # 最多3个进程同时运行
+            })
         self.scheduler.add_jobstore(DjangoJobStore(), 'default')
         self.scheduler.start()
+        self.conn = rpyc.connect('localhost', 12345)
 
     @staticmethod
     def auto_hello():  # 任务
@@ -18,9 +24,13 @@ class ApschedulerObject:
         print('hello world:[{}]'.format(now_time))
 
     def create_cron_job(self, func, task_id, day_of_week, hour, minute, args):  # 周期任务
-        self.scheduler.add_job(func=func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
-                               replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args,
-                               misfire_grace_time=300)
+        job = self.scheduler.add_job(func=func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
+                                     replace_existing=True, id=task_id, max_instances=1, coalesce=False, args=args,
+                                     misfire_grace_time=300)
+
+        # job = self.conn.root.add_job(func, trigger='cron', day_of_week=day_of_week, hour=hour, minute=minute,
+        #                              id=task_id, args=args, max_instances=1, coalesce=False, misfire_grace_time=300)
+        print(job)
 
     def create_interval_job(self, func, task_id, minutes, start_time, end_time, args):  # 间隔任务
         self.scheduler.add_job(func=func, trigger='interval', minutes=minutes,