john/ad-nix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more changes for dictionary unpacking

Browse Source
This commit is contained in:
John Lancaster
2024-08-27 19:14:26 -05:00
parent a8e0b53c53
commit 4114fb4c54
8 changed files with 26 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

309
apps/controller.py
View File

@@ -1,309 +0,0 @@
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import logging
from typing import List
from appdaemon.entity import Entity
from appdaemon.plugins.hass.hassapi import Hass
from daylight_adjuster import DaylightAdjuster
@dataclass(init=False)
class ControllerEntities(Hass):
entities: List[Entity]
def initialize(self):
# assign fields
for arg, val in self.args.items():
if arg not in ['class', 'module']:
setattr(self, arg, val)
# self.log(f'Set {arg} to {val}')
for entity in self.entities:
assert self.entity_exists(entity), f'{entity} does not exist'
self.entities = [self.get_entity(e) for e in self.entities]
@dataclass(init=False)
class ControllerRoomLights(ControllerEntities):
sleep: str
def initialize(self):
super().initialize()
self.log(f'Initialized light controller for {[e.friendly_name for e in self.entities]}')
self.register_service(f'{self.name}/activate', self.activate)
self.register_service(f'{self.name}/deactivate', self.deactivate)
def activate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
# if self.is_sleeping:
# self.log(f'Sleep mode is on, returning early')
# return
for entity in self.entities:
self.log(f'Turning on {entity.name}')
entity.turn_on()
def deactivate(self, namespace: str = None, domain: str = None, service=None, kwargs=None):
self.log(self.entities)
for entity in self.entities:
self.log(f'Turning off {entity.name}')
entity.turn_off()
@property
def state(self) -> bool:
return any([e.get_state() == 'on' for e in self.entities])
@property
def is_sleeping(self) -> bool:
if 'sleep' in self.args:
return self.get_entity(self.args['sleep']).is_state('on')
else:
return False
@is_sleeping.setter
def is_sleeping(self, val: bool):
if 'sleep' in self.args:
self.get_entity(self.args['sleep']).set_state(state='on' if val else 'off')
@dataclass(init=False)
class ControllerMotion(ControllerEntities):
room: ControllerRoomLights
off_duration: timedelta
def initialize(self):
super().initialize()
# self.log('Motion Controller init')
# convert room to App
self.room: ControllerRoomLights = self.get_app(self.room)
# convert off_duration
try:
hours, minutes, seconds = map(int, self.args['off_duration'].split(':'))
self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds)
except Exception:
self.off_duration = timedelta()
if self.current_state:
self.room.activate()
self.listen_motion_off()
self.listen_motion_on()
@property
def current_state(self) -> bool:
return any(e.get_state() == 'on' for e in self.entities)
def listen_motion_on(self):
self.listen_state(
callback=self.callback_motion_on,
entity_id=[e.entity_id for e in self.entities],
new='on',
)
self.log(f'Waiting for motion on {[e.friendly_name for e in self.entities]} to turn on room {self.room.name}')
def listen_motion_off(self):
self.listen_state(
callback=self.callback_motion_off,
entity_id=[e.entity_id for e in self.entities],
new='off',
duration=self.off_duration.total_seconds(),
)
self.log(f'Waiting for motion off {[e.friendly_name for e in self.entities]} for {self.off_duration}')
def callback_motion_on(self, entity, attribute, old, new, kwargs):
self.log(f'Motion detected on {self.friendly_name(entity)}')
if not self.room.is_sleeping:
self.room.activate()
def callback_motion_off(self, entity, attribute, old, new, kwargs):
self.log(f'Motion stopped on {self.friendly_name(entity)} for {self.off_duration}')
self.room.deactivate()
@dataclass(init=False)
class ControllerButton(Hass):
room: ControllerRoomLights
buttons: List[str]
def initialize(self):
self.buttons = self.args['buttons']
# convert room to App
self.room: ControllerRoomLights = self.get_app(self.args['room'])
for button in self.buttons:
self.listen_event(
self.callback_button,
event='deconz_event',
id=button,
)
self.log(f'Listening to presses on button ID={button}')
def callback_button(self, event_name, data, kwargs):
# single press
if data['event'] == 1002:
self.log(f"Single press: {data['id']}")
if self.room.state:
self.room.deactivate()
else:
self.room.activate()
# double click
elif data['event'] == 1004:
self.log(f'{data["id"]} double click')
self.room.is_sleeping = not self.room.is_sleeping
if not self.room.is_sleeping:
self.room.activate()
@dataclass(init=False)
class ControllerDaylight(Hass):
room: ControllerRoomLights
entities: List[str]
latitude: float
longitude: float
def initialize(self):
# convert room to App
self.room: ControllerRoomLights = self.get_app(self.args['room'])
# convert entities
for entity in self.args['entities']:
assert self.entity_exists(entity), f'{entity} does not exist'
self.entities = [self.get_entity(e) for e in self.args['entities']]
# self.log(self.entities)
# create Adjuster
self.adjuster = DaylightAdjuster(
latitude=self.args['latitude'],
longitude=self.args['longitude'],
periods=self.args['periods'],
resolution=500
)
# self.log(self.adjuster)
self.listen_state(callback=self.handle_off,
entity_id=[e.entity_id for e in self.entities],
new='off')
self.listen_state(callback=self.handle_state_change,
entity_id=[e.entity_id for e in self.entities],
attribute='brightness')
self.listen_state(callback=self.handle_state_change,
entity_id=[e.entity_id for e in self.entities],
attribute='color_temp')
ents = [e.friendly_name for e in self.entities]
if len(ents) > 1:
ents[-1] = f'and {ents[-1]}'
delim = ', ' if len(ents) >= 3 else ' '
ents = delim.join(ents)
self.log(f'Listening for state changes on {ents}')
interval = self.args.get('interval', 5)
self.run_every(
callback=self.update_sensors,
start='now',
interval=interval
)
self.log(f'Updating sensors every {timedelta(seconds=interval)}')
for entity in self.entities:
self.run_every(
callback=self.ongoing_adjustment,
start='now',
interval=interval,
entity=entity.entity_id
)
self.log(f'Updating sensors every {timedelta(seconds=interval)}')
if (entity_name := self.args.get('enable')) is not None:
self.enable_entity: Entity = self.get_entity(entity_name)
self.log(f'enabled by {self.enable_entity.friendly_name}[{entity_name}]')
self.listen_state(
callback=lambda entity, attribute, old, new, kwargs: self.ongoing_adjustment({'entity': entity}),
entity_id=entity_name,
new='on'
)
def handle_state_change(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
if not self.matching_state(entity):
self.log(f'{entity}.{attribute}: {old} -> {new}')
self.log(f'State does not match adjuster settings, disabling adjustments')
self.enabled = False
def matching_state(self, entity_id: str) -> bool:
"""Checks whether the current state of the light matches the settings from the DaylightAdjuster
Args:
entity_id (str): full entity ID
Returns:
bool
"""
state = self.get_state(entity_id=entity_id, attribute='all')['attributes']
settings = self.adjuster.current_settings
try:
state = {s: state[s] for s in settings.keys()}
except KeyError:
for s in settings.keys():
if s not in state:
self.log(f'{s} not in {state}')
return False
else:
valid = all((state[s] == val) for s, val in settings.items())
# if not valid:
# for s, val in settings.items():
# if state[s] != val:
# self.log(f'{entity_id}.{s}: {state[s]} != {val}')
return valid
@property
def enabled(self) -> bool:
if hasattr(self, 'enable_entity'):
return self.enable_entity.is_state('on')
else:
return True
@enabled.setter
def enabled(self, new: bool) -> bool:
self.enable_entity.set_state(state='on' if new else 'off')
def ongoing_adjustment(self, kwargs=None):
# self.log('Ongoing adjustment')
entity: Entity = self.get_entity(kwargs['entity'])
if self.enabled:
if entity.get_state() == 'on':
self.log(f'Ongoing adjustment for {entity.friendly_name}')
settings = self.adjuster.current_settings
if not self.matching_state(entity_id=entity.entity_id):
if not self.room.is_sleeping:
self.turn_on(entity_id=entity.entity_id, **settings)
self.log(f'Adjusted {entity.friendly_name} with {settings}')
else:
self.log(f'Sleeping mode active')
else:
self.log(f'{entity.friendly_name} settings already match')
else:
self.log(f'{entity.friendly_name} is off - no adjustment')
else:
self.log(f'App disabled by {self.enable_entity.friendly_name}')
def update_sensors(self, kwargs):
for key, val in self.adjuster.current_settings.items():
id = f'sensor.{self.name}_{key}'
self.set_state(
entity_id=id, state=val,
attributes={'friendly_name': f'Daylight, {key}, {self.name}',
'state_class': 'measurement'})
def handle_off(self, entity, attribute, old, new, kwargs):
self.log('Off handle')
self.run_in(
callback=lambda kwargs: self.enable_entity.set_state(state='on'),
delay=1.0
)
self.log('Re-enabled')

2
apps/cubes/cube.py
View File

@@ -92,7 +92,7 @@ class AqaraCube(Hass, Mqtt):
except KeyError:
return
def handle_event(self, event_name, data, cb_args):
def handle_event(self, event_name, data, **kwargs):
data = MQTTResponse.model_validate(data)
action = data.payload.action

32
apps/hello_world/hello.py
View File

@@ -2,26 +2,28 @@ import asyncio
import json
from appdaemon.adapi import ADAPI
from appdaemon.adbase import ADBase
from appdaemon.plugins.hass.hassapi import Hass
class HelloWorld(Hass):
class MyClass(Hass):
def initialize(self):
self.log('Hello World')
# self.listen_state(
# callback=self.temp_callback,
# entity_id='sensor.temperature_nest',
# attribute='state',
# threshold=65.0,
# )
# self.listen_state(
# callback=self.temp_callback,
# entity_id='light.living_room',
# attribute='state',
# threshold=65.0,
# )
async def temp_callback(self, entity, attribute, old, new, kwargs):
return
def my_callback(self, title: str, message: str, **kwargs) -> None:
self.log(f'{title}: {message}')
self.log(kwargs)
class HelloWorld(ADBase):
def initialize(self):
self.adapi = self.get_ad_api()
def my_callback(self, cb_args: dict) -> None:
self.adapi.log(f'{cb_args["title"]}: {cb_args["message"]}')
async def temp_callback(self, entity, attribute, old, new, **kwargs):
self.log('Temp callback')
temp = await self.get_state('sensor.temperature_nest')
# self.log(json.dumps(temp, indent=4))

32
apps/hello_world/weather.py
View File

@@ -97,35 +97,3 @@ class Weather(Hass):
async def publish_temp(self, json_data):
vals = convert_vals_dict(json_data['timelines']['hourly'][0]['values'])
await self.set_state('weather.tomorrowio', state=vals['condition'], **vals)
# def create_sensor(self, name: str, **kwargs) -> None:
# mqtt = self.app.get_plugin_api("MQTT")
# if "friendly_name" in kwargs:
# friendly_name = kwargs["friendly_name"]
# del kwargs["friendly_name"]
# else:
# friendly_name = name
# self.mqtt_registry[name] = {"entity_id": f"sensor.{name}"}
# if "initial_value" in kwargs:
# # create the state topic first and set the value
# self.set_state(name, state=kwargs["initial_value"])
# del kwargs["initial_value"]
# config = {
# "name": friendly_name,
# "object_id": name,
# "state_topic": f"appdaemon/{name}/state",
# "value_template": "{{ value_json.state }}",
# }
# for key, value in kwargs.items():
# config[key] = value
# mqtt.mqtt_publish(
# f"homeassistant/sensor/appdaemon/{name}/config",
# json.dumps(config),
# retain=True,
# )

2
apps/leaving.py
View File

@@ -5,7 +5,7 @@ class Leaving(Hass):
def initialize(self):
self.listen_state(self.turn_everything_off, entity_id=self.args['person'], old='home')
def turn_everything_off(self, entity, attribute, old, new, kwargs):
def turn_everything_off(self, entity, attribute, old, new, **kwargs):
self.log(f'turning everything off')
self.log(kwargs)
for app_name in self.args['apps']:

4
apps/patio.py
View File

@@ -17,11 +17,11 @@ class Patio(Hass):
def light(self) -> Entity:
return self.get_entity(self.args['light'])
def handle_door_open(self, entity: str, attribute: str, old: str, new: str, kwargs: dict):
def handle_door_open(self, entity: str, attribute: str, old: str, new: str, **kwargs):
self.log('Door open')
if self.AD.sched.location.solar_elevation() <= 0:
self.light.turn_on(**self.args['state'])
def handle_door_close(self, entity: str, attribute: str, old: str, new: str, kwargs: dict):
def handle_door_close(self, entity: str, attribute: str, old: str, new: str, **kwargs):
self.log('Door close')
self.run_in(callback=lambda *args: self.light.turn_off(), delay=30.0)

27
apps/speakers/speakers.py
View File

@@ -1,7 +1,6 @@
from datetime import datetime, time
from datetime import datetime
from appdaemon.plugins.hass.hassapi import Hass
from rich import print
class Speakers(Hass):
@@ -11,34 +10,14 @@ class Speakers(Hass):
)
self.set_volume()
# self.listen_state(
# callback=self.state_test,
# entity_id='media_player.nest_minis',
# # new=lambda n: n == 'playing',
# attribute='all'
# )
@property
def solar_elevation(self) -> float:
return self.AD.sched.location.solar_elevation(self.get_now())
def state_test(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
try:
print(new.keys())
self.log(new.keys())
except:
self.log('Error')
# for k, v in new.items():
# self.log(f'{k}: {v}')
# self.log(new['state'])
# self.log(entity)
# self.log(f'New state:\n{type(new)}')
# self.log(new['entity_id'])
def set_volume(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
def set_volume(self, entity=None, attribute=None, old=None, new=None, **kwargs):
self.log('Callback - state changed to playing')
if old == 'paused':
self.log(f'Unpaused - skipping volume adjust')
self.log('Unpaused - skipping volume adjust')
return
if self.get_now().time() < datetime.strptime('03:00', '%H:%M').time():

4
apps/tv.py
View File

@@ -54,10 +54,10 @@ class SoundBar(Hass):
command=command
)
def turn_on_soundbar(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
def turn_on_soundbar(self, entity=None, attribute=None, old=None, new=None, **kwargs):
self.log(f'{self.playing_entity} is playing')
self.send_remote_command(command='TV')
def hardware_off_callback(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
def hardware_off_callback(self, entity=None, attribute=None, old=None, new=None, **kwargs):
self.log(f'{self.off_entity} is off')
self.send_remote_command(command='power')