ASJServer/Controller/SensorGateway/SmartSwitchController.py

# -*- 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)))