# -*- coding: utf-8 -*- """ # @Author : cheng # @Time : 2023/7/10 11:20 # @File: SmartSwitchController.py """ from django.views import View from Model.models import SwitchInfo, SwitchDimmingSettings, SwitchChronopher from Service.CommonService import CommonService class SmartSwitchView(View): def get(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.GET, request, operation) def post(self, request, *args, **kwargs): request.encoding = 'utf-8' operation = kwargs.get('operation') return self.validation(request.POST, request, operation) def validation(self, request_dict, request, operation): token_code, user_id, response = CommonService.verify_token_get_user_id(request_dict, request) if token_code != 0: return response.json(token_code) if operation == 'get-switch-info': # 设备获取智能开关数据 return self.get_switch_info(request_dict, response) elif operation == 'get-switch-setting': # 获取智能开关调光设置 return self.get_switch_setting(request_dict, response) elif operation == 'get-chronopher-setting': # 获取定时计划 return self.get_chronopher_setting(request_dict, response) elif operation == 'add-or-edit-chronopher': # 添加/编辑定时计划 return self.add_or_edit_chronopher(request_dict, response) elif operation == 'delete-chronopher': # 删除定时计划 return self.delete_chronopher(request_dict, response) else: return response.json(414) @staticmethod def get_switch_info(request_dict, response): """ 获取智能开关设备信息 @param request_dict: 请求参数 @request_dict deviceId: 设备id @param response: 响应对象 @return: response """ device_id = request_dict.get('deviceId', None) if not device_id: return response.json(444) try: switch_info_qs = SwitchInfo.objects.filter(device_id=device_id).values() if not switch_info_qs.exists(): return response.json(173) res = { 'model': switch_info_qs[0]['model'], 'hardwareVersion': switch_info_qs[0]['hardware_version'], 'firmwareVersion': switch_info_qs[0]['firmware_version'], 'mac': switch_info_qs[0]['mac'], } return response.json(0, res) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def get_switch_setting(request_dict, response): """ 获取智能开关设备设置信息 @param request_dict: 请求参数 @request_dict deviceId: 设备id @param response: 响应对象 @return: response """ device_id = request_dict.get('deviceId', None) if not device_id: return response.json(444) try: switch_setting_info_qs = SwitchDimmingSettings.objects.filter(device_id=device_id).values() if not switch_setting_info_qs.exists(): return response.json(173) res = { 'clickTurnOnSpeed': switch_setting_info_qs[0]['click_turn_on_speed'], 'clickTurnOffSpeed': switch_setting_info_qs[0]['click_turn_off_speed'], 'doubleClick': switch_setting_info_qs[0]['double_click'], 'press': switch_setting_info_qs[0]['press'], 'doublePressClickTurnOnSpeed': switch_setting_info_qs[0]['double_press_click_turn_on_speed'], 'doublePressClickTurnOffSpeed': switch_setting_info_qs[0]['double_press_click_turn_off_speed'], 'led': switch_setting_info_qs[0]['led'], 'dimmingCorrection': switch_setting_info_qs[0]['dimming_correction'], } return response.json(0, res) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def edit_switch_setting(request_dict, response): device_id = request_dict.get('deviceId', None) device_nick_name = request_dict.get('deviceNickname', None) location = request_dict.get('location', None) led = request_dict.get('led', None) dimming_correction = request_dict.get('dimmingCorrection', None) if not device_id: return response.json(444) try: switch_setting_data = { 'device_id': device_id, 'device_nick_name': device_nick_name, 'location': location, 'led': led, 'dimming_correction': dimming_correction, } SwitchChronopher.objects.filter(device_id=device_id).update(**switch_setting_data) return response.json(0) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def get_chronopher_setting(request_dict, response): """ 获取定时计划设置 @param request_dict: 请求参数 @request_dict deviceId: 设备id @param response: 响应对象 @return: response """ device_id = request_dict.get('deviceId', None) if not device_id: return response.json(444) try: switch_chronopher_qs = SwitchChronopher.objects.filter(device_id=device_id).values() if not switch_chronopher_qs.exists(): return response.json(173) switch_chronopher_list = [] for item in switch_chronopher_qs: switch_chronopher_list.append({ 'chronopherId': item['id'], 'timeTypeRadio': item['time_type_radio'], 'timePoint': item['time_point'], 'timeQuantumStartTime': item['time_quantum_start_time'], 'timeQuantumEndTime': item['time_quantum_end_time'], 'timePointDeviceWillDoing': item['time_point_device_will_doing'], 'timeQuantumDeviceWillDoing': item['time_quantum_device_will_doing'], 'slowOpenOrCloseSpeed': item['slow_open_or_close_speed'], 'repeat': item['repeat'], }) return response.json(0, {'list': switch_chronopher_list}) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def add_or_edit_chronopher(request_dict, response): """ 添加/编辑定时计划 @param request_dict: 请求参数 @request_dict deviceId: 设备id @request_dict chronopherId: 定时计划id @request_dict timeTypeRadio: 切换时间点/时间段 @request_dict timePoint: 时间点 @request_dict timeQuantumStartTime: 时间段开始时间 @request_dict timeQuantumEndTime: 时间段结束时间 @request_dict timePointDeviceWillDoing: 设备将会 @request_dict timeQuantumDeviceWillDoing: 设备将会 @request_dict slowOpenOrCloseSpeed: 缓慢开/关速度 @request_dict repeat: 重复周期 @param response: 响应对象 @return: response """ is_edit = request_dict.get('isEdit', None) device_id = request_dict.get('deviceId', None) chronopher_id = request_dict.get('chronopherId', None) time_type_radio = int(request_dict.get('timeTypeRadio', 0)) time_point = request_dict.get('timePoint', None) time_quantum_start_time = request_dict.get('timeQuantumStartTime', None) time_quantum_end_time = request_dict.get('timeQuantumEndTime', None) time_point_device_will_doing = request_dict.get('timePointDeviceWillDoing', None) time_quantum_device_will_doing = request_dict.get('timeQuantumDeviceWillDoing', None) slow_open_or_close_speed = request_dict.get('slowOpenOrCloseSpeed', None) repeat = request_dict.get('repeat', None) if not all([device_id, time_type_radio, repeat]): return response.json(444) if time_type_radio == 1: if not all([time_point, slow_open_or_close_speed]): return response.json(444) chronopher_data = { 'device_id': device_id, 'time_type_radio': time_type_radio, 'time_point': time_point, 'time_point_device_will_doing': time_point_device_will_doing, 'slow_open_or_close_speed': slow_open_or_close_speed } else: if not all([time_quantum_start_time, time_quantum_end_time]): return response.json(444) chronopher_data = { 'device_id': device_id, 'time_type_radio': time_type_radio, 'time_quantum_start_time': time_quantum_start_time, 'time_quantum_end_time': time_quantum_end_time, 'time_quantum_device_will_doing': time_quantum_device_will_doing, } try: if is_edit: if not chronopher_id: return response.json(444) SwitchChronopher.objects.filter(device_id=device_id, id=chronopher_id).update(**chronopher_data) else: SwitchChronopher.objects.create(**chronopher_data) return response.json(0) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e))) @staticmethod def delete_chronopher(request_dict, response): """ 删除定时计划 @param request_dict: 请求参数 @request_dict deviceId: 设备id @request_dict chronopherId: 定时计划id @param response: 响应对象 @return: response """ device_id = request_dict.get('deviceId', None) chronopher_id = request_dict.get('chronopherId', None) if not chronopher_id: return response.json(444, {'error param': 'deviceId or chronopherId'}) try: SwitchChronopher.objects.filter(device_id=device_id, id=chronopher_id).delete() return response.json(0) except Exception as e: print(e) return response.json(500, 'error_line:{}, error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))