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

formatting

Browse Source
This commit is contained in:
John Lancaster
2024-05-07 23:37:34 -05:00
parent a38341484d
commit 8eb9e84b9d
4 changed files with 50 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
apps/cubes/cube.py
View File

@@ -1,11 +1,14 @@
import logging
from enum import Enum
from typing import Literal, Optional
from logging import Logger
from typing import TYPE_CHECKING, Literal, Optional
from appdaemon.plugins.hass.hassapi import Hass
from appdaemon.plugins.mqtt.mqttapi import Mqtt
from pydantic import BaseModel, Field, TypeAdapter, field_validator
from room_control.console import setup_component_logging
from room_control import console
if TYPE_CHECKING:
from room_control import RoomController
class Side(int, Enum):
@@ -62,19 +65,21 @@ Callbacks = dict[str, dict[str, CallbackEntry]]
class AqaraCube(Hass, Mqtt):
app: 'RoomController'
logger: Logger
def initialize(self):
setup_component_logging(self)
self.logger.setLevel(self.args.get('rich', logging.INFO))
self.app: 'RoomController' = self.get_app(self.args['app'])
self.logger = console.load_rich_config(self.app.name, type(self).__name__, level='DEBUG')
topic = f'zigbee2mqtt/{self.args["cube"]}'
self.mqtt_subscribe(topic, namespace='mqtt')
self.listen_event(self.handle_event, 'MQTT_MESSAGE', topic=topic, namespace='mqtt')
self.log(f'Listening for cube events on: [topic]{topic}[/]')
self.app = self.get_app(self.args['app'])
self.log(f'Connected to AD app: [room]{self.app.name}[/]')
self.log(self.callbacks())
# self.log(f'Number of callbacks: {len(self.callbacks())}')
# for handle, cb in self.callbacks().items():
# self.log(repr(cb))
def terminate(self):
self.log('[bold red]Terminating[/]', level='DEBUG')

24
apps/scene_detect.py
View File

@@ -3,12 +3,18 @@ from appdaemon.plugins.hass.hassapi import Hass
class SceneDetector(Hass):
def initialize(self):
self.scene_entity = self.args['scene'] if self.args['scene'].startswith('scene.') else f'scene.{self.args["scene"]}'
self.scene_entity = (
self.args['scene']
if self.args['scene'].startswith('scene.')
else f'scene.{self.args["scene"]}'
)
self.scene_entity = self.get_entity(self.scene_entity)
self.listen_event(self.event_callback, event='call_service', domain='scene', service='turn_on')
self.listen_event(
self.event_callback, event='call_service', domain='scene', service='turn_on'
)
self.log(f'Waiting for {self.scene_entity.friendly_name} to activate')
def event_callback(self, event_name, data, cb_args):
entity_id = data['service_data']['entity_id']
if entity_id == self.scene_entity.entity_id:
@@ -23,11 +29,11 @@ class MotionCanceller(SceneDetector):
super().scene_detected()
app = self.get_app(self.args['app'])
try:
self.run_in(
callback=lambda *args, **kwargs: app.cancel_motion_callback(),
delay=0.5
)
self.run_in(callback=lambda *args, **kwargs: app.cancel_motion_callback(), delay=0.5)
except Exception:
self.log(f'Error cancelling motion callback for {self.args["app"]}', level='ERROR')
self.log(
f'Error cancelling motion callback for {self.args["app"]}',
level='ERROR',
)
else:
self.log('Cancelled motion callback')

10
apps/sleep.py
View File

@@ -21,7 +21,13 @@ class SleepSetter(Hass, Mqtt):
def setup_button(self, name: str):
topic = f'zigbee2mqtt/{name}'
self.mqtt_subscribe(topic, namespace='mqtt')
self.listen_event(self.handle_button, "MQTT_MESSAGE", topic=topic, namespace='mqtt', button=name)
self.listen_event(
self.handle_button,
'MQTT_MESSAGE',
topic=topic,
namespace='mqtt',
button=name,
)
self.log(f'Subscribed: {topic}')
@property
@@ -121,4 +127,4 @@ class SleepSetter(Hass, Mqtt):
except:
return
else:
self.log(f'Activated {app_name}')
self.log(f'Activated {app_name}')

36
apps/speakers/speakers.py
View File

@@ -7,9 +7,7 @@ from rich import print
class Speakers(Hass):
def initialize(self):
self.listen_state(
callback=self.set_volume,
entity_id='media_player.nest_minis',
new='playing'
callback=self.set_volume, entity_id='media_player.nest_minis', new='playing'
)
self.set_volume()
@@ -51,7 +49,7 @@ class Speakers(Hass):
entity_id=[
'media_player.kitchen_speaker',
'media_player.desk_speaker',
]
],
)
self.call_service(
@@ -60,15 +58,13 @@ class Speakers(Hass):
entity_id=[
'media_player.bedroom_speaker',
'media_player.bathroom_speaker',
]
],
)
self.call_service(
'media_player/volume_set',
volume_level=0.0,
entity_id=[
'media_player.patio_speaker'
]
entity_id=['media_player.patio_speaker'],
)
elif self.get_now().time() < datetime.strptime('10:00', '%H:%M').time():
@@ -79,7 +75,7 @@ class Speakers(Hass):
entity_id=[
'media_player.kitchen_speaker',
'media_player.desk_speaker',
]
],
)
self.call_service(
@@ -88,15 +84,13 @@ class Speakers(Hass):
entity_id=[
'media_player.bedroom_speaker',
'media_player.bathroom_speaker',
]
],
)
self.call_service(
'media_player/volume_set',
volume_level=0.0,
entity_id=[
'media_player.patio_speaker'
]
entity_id=['media_player.patio_speaker'],
)
elif self.solar_elevation < 0:
@@ -107,7 +101,7 @@ class Speakers(Hass):
entity_id=[
'media_player.kitchen_speaker',
'media_player.desk_speaker',
]
],
)
self.call_service(
@@ -116,15 +110,13 @@ class Speakers(Hass):
entity_id=[
'media_player.bedroom_speaker',
'media_player.bathroom_speaker',
]
],
)
self.call_service(
'media_player/volume_set',
volume_level=0.0,
entity_id=[
'media_player.patio_speaker'
]
entity_id=['media_player.patio_speaker'],
)
else:
@@ -135,7 +127,7 @@ class Speakers(Hass):
entity_id=[
'media_player.kitchen_speaker',
'media_player.desk_speaker',
]
],
)
self.call_service(
@@ -144,13 +136,11 @@ class Speakers(Hass):
entity_id=[
'media_player.bedroom_speaker',
'media_player.bathroom_speaker',
]
],
)
self.call_service(
'media_player/volume_set',
volume_level=0.0,
entity_id=[
'media_player.patio_speaker'
]
entity_id=['media_player.patio_speaker'],
)