Skip to content

Commit

Permalink
Refactor code
Browse files Browse the repository at this point in the history
  • Loading branch information
@DewGew
DewGew authored Nov 1, 2023
1 parent 95bb295 commit 984b33b
Showing 1 changed file with 77 additions and 107 deletions.
184 changes: 77 additions & 107 deletions modules/trait.py
View file
Original file line number Diff line number Diff line change
@@ -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)
Expand All @@ -19,33 +19,33 @@ 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']
elif state['Data'] == 'Open':
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'])

Expand All @@ -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:
Expand All @@ -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'
Expand All @@ -95,120 +95,103 @@ 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&param='

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']
setTemp = 100 - (((kelvinTemp - 1700) / tempRange) * 100)

url += 'setkelvinlevel&idx=' + idx + '&kelvin=' + str(
round(setTemp))

elif "spectrumRGB" in params["color"]:
# Convert decimal to hex
setcolor = params['color']
Expand All @@ -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"]:
Expand All @@ -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')

Expand All @@ -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

0 comments on commit 984b33b

Please sign in to comment.