From 984b33bdd063bcd948cb544945e2692ad8d1e97d Mon Sep 17 00:00:00 2001 From: DewGew Date: Wed, 1 Nov 2023 17:28:32 +0100 Subject: [PATCH] Refactor code --- modules/trait.py | 184 ++++++++++++++++++++--------------------------- 1 file changed, 77 insertions(+), 107 deletions(-) diff --git a/modules/trait.py b/modules/trait.py index 642083a..b5a3f8a 100644 --- a/modules/trait.py +++ b/modules/trait.py @@ -1,15 +1,15 @@ -import subprocess import hashlib import base64 import json -import modules.config as config -from modules.database import db, Settings + +from modules.database import Settings from modules.helpers import _tempConvert from modules.domoticz import getDomoticzState, queryDomoticz attempts = 1 blocked = {} - + + def query(custom_data, device, user_id): dbsettings = Settings.query.get_or_404(1) @@ -19,17 +19,17 @@ def query(custom_data, device, user_id): state = getDomoticzState(user_id, idx, 'scene') else: state = getDomoticzState(user_id, idx) - + response = {} if 'action.devices.traits.OnOff' in device['traits']: if domain in ['Group', 'Scene']: data = state['Status'] != 'Off' - response['on'] = data + response['on'] = data else: data = state['Data'] != 'Off' response['on'] = data - + if 'action.devices.traits.OpenClose' in device['traits']: if domain in ['BlindsPercentage', 'BlindsStop']: response['openPercent'] = state['Level'] @@ -37,15 +37,15 @@ def query(custom_data, device, user_id): response['openPercent'] = 100 else: response['openPercent'] = 0 - + if 'action.devices.traits.LockUnlock' in device['traits']: data = state['Data'] != 'Unlocked' response['isLocked'] = data - + if 'action.devices.traits.StartStop' in device['traits']: if domain in ['VenetianBlindsUS', 'VenetianBlindsEU', 'BlindsStop']: response['isRunning'] = True - + if 'action.devices.traits.Brightness' in device['traits']: response['brightness'] = int(state['Level'] * 100 / state['MaxDimLevel']) @@ -61,10 +61,10 @@ def query(custom_data, device, user_id): if color_rgb["m"] == 2: colorTemp = (color_rgb["t"] * (255 / 100)) * 10 response['color'] = {'temperatureK': round(colorTemp)} - + except ValueError: response = {} - + if 'action.devices.traits.TemperatureSetting' in device['traits']: if 'Thermostat' in domain: if 'actual_temp_idx' in custom_data: @@ -78,14 +78,14 @@ def query(custom_data, device, user_id): if 'selector_modes_idx' in custom_data: selectorModesState = getDomoticzState(user_id, custom_data['selector_modes_idx']) selectorLevel = selectorModesState['Level'] - selectorModes = base64.b64decode(selectorModesState['LevelNames']).decode('UTF-8').lower().split("|") + selectorModes = base64.b64decode(selectorModesState['LevelNames']).decode('UTF-8').lower().split("|") modeIndex = int(selectorLevel / 10) response['thermostatMode'] = selectorModes[modeIndex] else: response['thermostatMode'] = 'heat' elif domain in ['Temp', 'TempHumidity', 'TempHumidityBaro']: current_temp = float(state['Temp']) - if round(_tempConvert(current_temp, dbsettings.tempunit),1) <= 3: + if round(_tempConvert(current_temp, dbsettings.tempunit), 1) <= 3: response['thermostatMode'] = 'cool' else: response['thermostatMode'] = 'heat' @@ -95,112 +95,95 @@ def query(custom_data, device, user_id): current_humidity = state['Humidity'] if current_humidity is not None: response['thermostatHumidityAmbient'] = current_humidity - + if 'action.devices.traits.TemperatureControl' in device['traits']: if domain in ['Temp', 'TempHumidity', 'TempHumidityBaro']: current_temp = float(state['Temp']) response['temperatureAmbientCelsius'] = round(_tempConvert(current_temp, dbsettings.tempunit), 1) - + if 'action.devices.traits.Toggles' in device['traits']: levelName = base64.b64decode(state.get("LevelNames")).decode('UTF-8').split("|") level = state['Level'] index = int(level / 10) - + toggle_settings = { levelName[index]: state['Data'] != 'Off' } if toggle_settings: response["currentToggleSettings"] = toggle_settings - - # if 'action.devices.traits.Modes' in device['traits']: - # levelName = base64.b64decode(state.get("LevelNames")).decode('UTF-8').split("|") - # level = state['Level'] - # index = int(level / 10) - - # mode_settings = { - # levelName[index]: state['Data'] != 'Off' - # } - # if mode_settings: - # response["currentModeSettings"] = { - # levelName[index]: levelName[index] - # } - - + if 'action.devices.traits.SensorState' in device['traits']: if state['Data'] is not None: - response['currentSensorStateData'] = [{'name': 'SmokeLevel', 'currentSensorState': ('smoke detekted' if state['Data'] == 'On' else 'no smoke detected'),}] - + response['currentSensorStateData'] = [{'name': 'SmokeLevel', 'currentSensorState': ( + 'smoke detekted' if state['Data'] == 'On' else 'no smoke detected')}] + if 'action.devices.traits.OccupancySensing' in device['traits']: if state['Data'] is not None: response['occupancy'] = ('OCCUPIED' if state['Data'] == 'On' else 'UNOCCUPIED') if 'action.devices.traits.ArmDisarm' in device['traits']: - + response["isArmed"] = state['Data'] != "Normal" if response["isArmed"]: response["currentArmLevel"] = state['Data'] - - - response['online'] = True - + + response['online'] = True + return response + def execute(custom_data, command, params, user_id, challenge): global attempts global blocked - + idx = custom_data['idx'] domain = custom_data['domain'] - + if domain in ['Group', 'Scene']: state = getDomoticzState(user_id, idx, 'scene') else: state = getDomoticzState(user_id, idx) - + if idx in blocked: return {"status": "ERROR", "errorCode": "tooManyFailedAttempts"} - + response = {"status": "SUCCESS", "states": {}} url = '?type=command¶m=' - + if command == "action.devices.commands.OnOff": - - if domain in ['Group', 'Scene']: + + if domain in ['Group', 'Scene']: url += 'switchscene&idx=' + idx + '&switchcmd=' + ( 'On' if params['on'] else 'Off') else: url += 'switchlight&idx=' + idx + '&switchcmd=' + ( 'On' if params['on'] else 'Off') - + response['states']['on'] = params['on'] - + if command == 'action.devices.commands.LockUnlock': - + url += 'switchlight&idx=' + idx + '&switchcmd=' + ( 'On' if params['lock'] else 'Off') - + response['states']['isLocked'] = params['lock'] - - if command == "action.devices.commands.ActivateScene": - + if params['deactivate'] is False: url += 'switchscene&idx=' + idx + '&switchcmd=On' - - + if command == "action.devices.commands.BrightnessAbsolute": - + url += 'switchlight&idx=' + idx + '&switchcmd=Set%20Level&level=' + str( int(params['brightness'] * state['MaxDimLevel'] / 100)) - + response['states']['brightness'] = params['brightness'] - if command == "action.devices.commands.ColorAbsolute": - + if "temperature" in params["color"]: tempRange = 1700 - 6500 kelvinTemp = params['color']['temperature'] @@ -208,7 +191,7 @@ def execute(custom_data, command, params, user_id, challenge): url += 'setkelvinlevel&idx=' + idx + '&kelvin=' + str( round(setTemp)) - + elif "spectrumRGB" in params["color"]: # Convert decimal to hex setcolor = params['color'] @@ -220,62 +203,61 @@ def execute(custom_data, command, params, user_id, challenge): color_hex_str += str(color_hex) url += 'setcolbrightnessvalue&idx=' + idx + '&hex=' + color_hex_str - + response['states']['color'] = params['color'] - if command == 'action.devices.commands.ThermostatTemperatureSetpoint': - + url += 'setsetpoint&idx=' + idx + '&setpoint=' + str( params['thermostatTemperatureSetpoint']) - + response['states']['thermostatTemperatureSetpoint'] = params['thermostatTemperatureSetpoint'] - + if command == 'action.devices.commands.ThermostatSetMode': - + if 'selector_modes_idx' in custom_data: data = getDomoticzState(user_id, custom_data['selector_modes_idx']) selctorModeLevels = base64.b64decode(data["LevelNames"]).decode('UTF-8').split("|") selctorModeNames = [x.lower() for x in selctorModeLevels] - + if params['thermostatMode'] in selctorModeNames: thermostatMode = str(selctorModeNames.index(params['thermostatMode']) * 10) - + url += 'switchlight&idx=' + custom_data['selector_modes_idx'] + '&switchcmd=Set%20Level&level=' + thermostatMode - + if command == 'action.devices.commands.OpenClose': - + if domain in ['BlindsPercentage', 'BlindsStop']: url += 'switchlight&idx=' + idx + '&switchcmd=Set%20Level&level=' + str( params['openPercent']) - + else: - p = params.get('openPercent', 50) + p = params.get('openPercent', 50) url += 'switchlight&idx=' + idx + '&switchcmd=' if p == 100: url += 'Open' if p == 0: url += 'Close' - + response['states']['openState'] = [{'openPercent': params['openPercent']}] - if command == 'action.devices.commands.StartStop': - + if command == 'action.devices.commands.StartStop': + if params['start'] is False: url += 'switchlight&idx=' + idx + '&switchcmd=Stop' else: url += 'switchlight&idx=' + idx + '&switchcmd=' + ( 'On' if params['start'] else 'Off') - + if command == 'action.devices.commands.GetCameraStream': - + if custom_data['camurl']: response['cameraStreamAccessUrl'] = custom_data['camurl'] - response['states']['online'] = True + response['states']['online'] = True return response else: return {"status": "ERROR", "errorCode": "streamUnavailable"} - + if command == 'action.devices.commands.ArmDisarm': if params["arm"]: @@ -285,42 +267,29 @@ def execute(custom_data, command, params, user_id, challenge): url += "setsecstatus&secstatus=2" else: url += "setsecstatus&secstatus=0" - + url += '&seccode=' + hashlib.md5(str.encode(challenge.get('pin'))).hexdigest() - + response['states']['isArmed'] = params['arm'] if response['states']["isArmed"]: response['states']['currentArmLevel'] = params['armLevel'] - + if command == 'action.devices.commands.SetToggles': - + levelName = base64.b64decode(state.get("LevelNames")).decode('UTF-8').split("|") - - + for key in params['updateToggleSettings']: key = key.replace("_", " ") if key in levelName: slevel = str(levelName.index(key) * 10) - + url += 'switchlight&idx=' + idx + '&switchcmd=Set%20Level&level=' + slevel - - # if command == 'action.devices.commands.SetModes': - - # levelName = base64.b64decode(state.get("LevelNames")).decode('UTF-8').split("|") - - - # for key in params['updateModeSettings']: - # key = key.replace("_", " ") - # if key in levelName: - # slevel = str(levelName.index(key) * 10) - - # url += 'switchlight&idx=' + idx + '&switchcmd=Set%20Level&level=' + slevel - - if state['Protected'] == True: + + if state['Protected'] is True: url = url + '&passcode=' + challenge.get('pin') - + r = queryDomoticz(user_id, url) - + result = json.loads(r) status = result.get('status') @@ -329,13 +298,14 @@ def execute(custom_data, command, params, user_id, challenge): elif status == 'ERROR': if attempts > 2: blocked = {idx: True} - return {"status": "ERROR", "errorCode": "challengeNeeded", "challengeNeeded": { "type": "tooManyFailedAttempts"}} + return {"status": "ERROR", "errorCode": "challengeNeeded", "challengeNeeded": { + "type": "tooManyFailedAttempts"}} attempts += 1 - return {"status": "ERROR", "errorCode": "challengeNeeded", "challengeNeeded": { "type": "challengeFailedPinNeeded"}} + return {"status": "ERROR", "errorCode": "challengeNeeded", "challengeNeeded": { + "type": "challengeFailedPinNeeded"}} else: attempts = 1 - + response['states']['online'] = True - + return response -