john/room_control
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added entity callbacks

Browse Source
This commit is contained in:
John Lancaster
2024-09-12 21:15:30 -05:00
parent 8b7294b445
commit b1638be692
1 changed files with 37 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
src/room_control/motion.py
View File

@@ -37,12 +37,23 @@ class MotionSensor:
assert self.sensor_entity.exists()
assert self.ref_entity.exists()
base_kwargs = dict(
entity_id=self.ref_entity_id,
immediate=True, # avoids needing to sync the state
)
self.ref_entity.listen_state(self.callback_light_on, attribute='all', **base_kwargs)
self.ref_entity.listen_state(self.callback_light_off, new='off', **base_kwargs)
self.ref_entity.listen_state(self.light_state_callback, immediate=True)
self.match_new_state(new=self.sensor_entity.get_state())
self.logger.info('Initialized motion sensor')
def entity_callbacks(self, entity_id: str | None = None) -> dict[str, dict]:
callbacks = self.adapi.get_callback_entries()
if self.adapi.name in callbacks:
return {
handle: cb
for handle, cb in callbacks[self.adapi.name].items()
if cb.get('entity') == entity_id
}
else:
return {}
def sensor_callbacks(self) -> dict[str, dict]:
return self.entity_callbacks(entity_id=self.sensor_entity_id)
@property
def sensor_entity(self) -> Entity:
@@ -64,47 +75,50 @@ class MotionSensor:
def state_mismatch(self) -> bool:
return self.sensor_state != self.ref_state
def callback_light_on(self, entity: str, attribute: str, old: str, new: str, **kwargs: dict):
"""Called when the light turns on"""
if new['state'] == 'on':
self.logger.debug(f'Detected {entity} turning on')
duration = self.adapi.off_duration()
self.listen_motion_off(duration)
def light_state_callback(self, entity: str, attribute: str, old: str, new: str, **kwargs: dict):
for handle in self.sensor_callbacks():
self.adapi.cancel_listen_state(handle)
self.match_new_state(new)
def callback_light_off(self, entity: str, attribute: str, old: str, new: str, **kwargs: dict):
"""Called when the light turns off"""
self.logger.debug(f'Detected {entity} turning off')
self.listen_motion_on()
def match_new_state(self, new: Literal['on', 'off']):
match new:
case 'on':
duration = self.adapi.off_duration()
self.listen_motion_off(duration)
case 'off':
self.listen_motion_on()
def listen_motion_on(self):
"""Sets up the motion on callback to activate the room"""
# self.cancel_motion_callback()
self.adapi.listen_state(
self.sensor_entity.listen_state(
lambda *args, **kwargs: self.adapi.activate_all_off(cause='motion on'),
entity_id=self.sensor_entity_id,
new='on',
oneshot=True,
)
self.logger.info(
f'Waiting for sensor motion on [friendly_name]{self.sensor_entity.friendly_name}[/]'
'Waiting for sensor motion on '
f'[friendly_name]{self.sensor_entity.friendly_name}[/]'
)
if self.sensor_state:
self.logger.warning(
f'Sensor [friendly_name]{self.sensor_entity.friendly_name}[/] is already on',
'Sensor '
f'[friendly_name]{self.sensor_entity.friendly_name}[/] is already on',
)
def listen_motion_off(self, duration: timedelta):
"""Sets up the motion off callback to deactivate the room"""
# self.cancel_motion_callback()
self.adapi.listen_state(
self.sensor_entity.listen_state(
lambda *args, **kwargs: self.adapi.deactivate(cause='motion off'),
entity_id=self.sensor_entity_id,
new='off',
duration=duration.total_seconds(),
oneshot=True,
)
self.logger.debug(
f'Waiting for sensor [friendly_name]{self.sensor_entity.friendly_name}[/] to be clear for {duration}'
'Waiting for sensor '
f'[friendly_name]{self.sensor_entity.friendly_name}[/] '
f'to be clear for {duration}'
)
if not self.sensor_state: