dxs
/
hass-4button
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Restructure and add HTTP API 

- Move check_action and run_action into actions.py
- Add HTTP API
- Minor formatting changes
master
Dessa Simpson 2022-05-28 06:08:22 +00:00
parent e58b1b21ed
commit 96a25cf79b
3 changed files with 75 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
actions.py
View File

@ -1,4 +1,5 @@
#!/usr/bin/python3
import inspect
###############
# Action Tree #
@ -506,21 +507,54 @@ tree = {
def set_brightness(hass, device, brightness):
# Convert brightness from percent to 0-255
brightness = brightness * 255 / 100;
hass.turn_on(device, brightness=brightness);
brightness = brightness * 255 / 100
hass.turn_on(device, brightness=brightness)
def set_temperature(hass, device, temperature):
hass.call_service("climate/set_temperature",entity_id="climate."+device,temperature=temperature);
hass.call_service("climate/set_temperature",entity_id="climate."+device,temperature=temperature)
def gree_set_mode(hass, device, mode):
hass.call_service("climate/set_hvac_mode",entity_id="climate."+device,hvac_mode=mode);
hass.call_service("climate/set_hvac_mode",entity_id="climate."+device,hvac_mode=mode)
def gree_set_fan_level(hass, device, level):
if (level == "quiet"):
hass.turn_on(f"switch.{device}_quiet");
hass.turn_on(f"switch.{device}_quiet")
else:
hass.turn_off(f"switch.{device}_quiet");
hass.call_service("climate/set_fan_mode",entity_id="climate."+device,fan_mode=level);
hass.turn_off(f"switch.{device}_quiet")
hass.call_service("climate/set_fan_mode",entity_id="climate."+device,fan_mode=level)
def select_source(hass, device, source):
hass.call_service("media_player/select_source",entity_id=device,source=source)
###############################
# Action Processing Functions #
###############################
def check_action(sequence, tree=tree):
if (len(sequence) > 0):
# Deeper into the rabbit hole
category = sequence.pop(0)
if (category in tree): return check_action(sequence,tree[category])
else: return False
else:
# We've the end of the sequence thus far, check if this is the end of a branch
# If any of {1..4} exist as keys in tree there's more options
return not any(i in tree for i in range(1,4))
def run_action(hass, sequence, tree=tree):
if (len(sequence) > 0):
# Deeper into the rabbit hole
category = sequence.pop(0)
if (category in tree): return run_action(hass,sequence,tree[category])
else:
hass.log("No match")
return False
else:
# Perform an action from here and log the description
if ("function" in tree): tree["function"](hass)
else:
hass.log("No match")
return False
if ("description" in tree): hass.log(tree["description"])
else: hass.log(inspect.getsource(tree["function"]).strip())
return True

15
http_api.py Normal file
View File

@ -0,0 +1,15 @@
#!/usr/bin/python3
import hassapi as hass
from actions import run_action
class HttpApi(hass.Hass):
def initialize(self):
self.register_endpoint(self.api_request, '4button')
def api_request(self, data, kwargs):
if 'sequence' not in data: return 'Missing sequence', 400
# Let AppDaemon handle API response on error
if (run_action(self,data['sequence'])):
return 'Success', 200
else:
return 'No match', 400

70
wallmote.py
View File

@ -1,17 +1,12 @@
#!/usr/bin/python3
import hassapi as hass
import inspect
from threading import Timer
from actions import tree
##################
# Sequence Logic #
##################
from actions import check_action, run_action
# Short presses will be recorded into a sequence array. Long presses are reserved for
# actions directly in hass for now; use KeyReleased event to capture them. A sequence
# can be ended either immediately by recognizing a full sequence or by TIMEOUT seconds
# passing. (The former is NYI but the code is capable as is.)
# passing.
# - self.currentSequence contains an array of keys representing a sequence that is
# actively being entered.
@ -21,58 +16,31 @@ TIMEOUT=1
class WallmotePlumbing(hass.Hass):
def initialize(self):
self.currentSequence = [];
self.listen_event(self.onevent, "zwave_js_value_notification");
self.currentSequence = []
self.listen_event(self.onevent, "zwave_js_value_notification")
def onevent(self, _, event, kwargs):
if (event.get('value') == 'KeyHeldDown'): return;
if (event.get('device_id') != '3f7add36f94839b1446886b5dab35e6d'): return;
key = int(event.get('property_key'));
if (event.get('value') == 'KeyHeldDown'): return
if (event.get('device_id') != '3f7add36f94839b1446886b5dab35e6d'): return
key = int(event.get('property_key'))
# We don't get a KeyPressed event at all for long presses
evt = event.get('value');
if (evt == "KeyPressed"): self.shortpress(key);
evt = event.get('value')
if (evt == "KeyPressed"): self.shortpress(key)
def shortpress(self, key):
self.currentSequence.append(key);
self.reset_tot();
self.currentSequence.append(key)
self.reset_tot()
# Don't wait for tot to expire if we've already hit the end of a branch
if (self.check_action(self.currentSequence.copy())): self.process_sequence();
if (check_action(self.currentSequence.copy())): self.process_sequence()
def reset_tot(self):
if (hasattr(self,"tot")): self.tot.cancel();
self.tot = Timer(TIMEOUT,self.process_sequence);
self.tot.start();
if (hasattr(self,"tot")): self.tot.cancel()
self.tot = Timer(TIMEOUT,self.process_sequence)
self.tot.start()
def process_sequence(self):
self.tot.cancel(); # So process_sequence can be called directly
sequence = self.currentSequence;
self.currentSequence = [];
self.log(sequence);
self.run_action(sequence);
def check_action(self, sequence, tree=tree):
if (len(sequence) > 0):
# Deeper into the rabbit hole
category = sequence.pop(0);
if (category in tree): return self.check_action(sequence,tree[category]);
else: return false;
else:
# We've the end of the sequence thus far, check if this is the end of a branch
# If any of {1..4} exist as keys in tree there's more options
return not any(i in tree for i in range(1,4));
def run_action(self, sequence, tree=tree):
if (len(sequence) > 0):
# Deeper into the rabbit hole
category = sequence.pop(0);
if (category in tree): self.run_action(sequence,tree[category]);
else: self.log("No match");
else:
# Perform an action from here and log the description
if ("function" in tree): tree["function"](self);
elif ("service" in tree): self.call_service(tree["service"]);
else:
self.log("No match");
return;
if ("description" in tree): self.log(tree["description"]);
else: self.log(inspect.getsource(tree["function"]).strip());
sequence = self.currentSequence
self.currentSequence = []
self.log(sequence)
run_action(self,sequence)