remote/adaptive-lighting
1
0
Fork 0
mirror of https://github.com/basnijholt/adaptive-lighting.git synced 2026-09-27 20:34:20 +02:00
Code Issues Projects Releases Packages Wiki Activity

simplify turn on and off listening

Browse source
This commit is contained in:
Bas Nijholt 2020-09-28 14:35:32 +02:00
commit 555d405399
1 changed files with 26 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

74
custom_components/adaptive_lighting/switch.py
View file

@ -9,12 +9,7 @@ from typing import Dict, Tuple
import voluptuous as vol
from homeassistant.components.homeassistant import (
DOMAIN as HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
)
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_BRIGHTNESS_PCT,
ATTR_COLOR_TEMP,
ATTR_RGB_COLOR,
@ -285,7 +280,6 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
all_lights.extend(group)
_LOGGER.debug("%s: Unpacked %s to %s", self._name, light, group)
else:
_LOGGER.debug("%s: Did not unpack %s to %s", self._name, light)
all_lights.append(light)
self._lights = all_lights
@ -531,13 +525,12 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
_LOGGER.debug(
"%s: Detected an 'off' → 'on' event for '%s'", self._name, entity_id
)
on_to_off_event = self._on_to_off_event.get(entity_id)
lock = self._locks.setdefault(entity_id, asyncio.Lock())
async with lock:
if await self.turn_on_off_listener.maybe_cancel_adjusting(
entity_id,
off_to_on_event=event,
on_to_off_event=on_to_off_event,
on_to_off_event=self._on_to_off_event.get(entity_id),
):
# Stop if a rapid 'off' → 'on' → 'off' happens.
_LOGGER.debug(
@ -560,14 +553,14 @@ class AdaptiveSwitch(SwitchEntity, RestoreEntity):
class TurnOnOffListener:
"""Track 'light.turn_off(..., transition=...)' and 'light.turn_on' service calls."""
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
def __init__(self, hass):
"""Initialize the TurnOnOffListener that is shared among all switches."""
self.hass = hass
self.lights = set()
# Tracks 'light.turn_off(..., transition=...)' service calls
# Tracks 'light.turn_off' service calls
self.turn_off_event: Dict[str, Tuple[str, float]] = {}
# Tracks 'light.turn_on' service calls
self.turn_on_event: Dict[str, Tuple[str]] = {}
@ -602,10 +595,14 @@ class TurnOnOffListener:
# State change 'off' → 'on' triggered by 'light.turn_on'.
return False
if id_on_to_off == id_turn_off and id_on_to_off is not None:
# State change 'off' → 'on' and 'light.turn_off(..., transition=...)' come
if (
id_on_to_off == id_turn_off
and id_on_to_off is not None
and transition is not None # 'turn_off' is called with transition=...
):
# State change 'on' → 'off' and 'light.turn_off(..., transition=...)' come
# from the same event, so wait at least the 'turn_off' transition time.
delay = transition
delay = max(transition, TURNING_OFF_DELAY)
else:
# State change 'off' → 'on' happened because the light state was set.
# Possibly because of polling.
@ -621,44 +618,26 @@ class TurnOnOffListener:
# is 'off' or the time has passed.
delay -= delta_time # delta_time has passed since the 'off' → 'on' event
_LOGGER.debug("Waiting with adjusting '%s' for %s.", entity_id, delay)
current_state = self.hass.states.get(entity_id)
_LOGGER.debug("'%s' state before sleep is '%s'", entity_id, current_state)
_LOGGER.debug("Waiting with adjusting '%s' for %s", entity_id, delay)
for _ in range(3):
# It can happen that the actual transition time is longer than the
# specified time in the 'turn_off' service, so we check whether the
# brightness is still going down, if so, we wait a little longer.
# specified time in the 'turn_off' service.
await asyncio.sleep(delay)
await self.hass.services.async_call(
HA_DOMAIN,
SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
old_state = current_state
current_state = self.hass.states.get(entity_id)
if current_state.state == "off":
if not is_on(self.hass, entity_id):
return True
old_brightness = old_state.attributes.get(ATTR_BRIGHTNESS, 0)
current_brightness = current_state.attributes.get(ATTR_BRIGHTNESS, 0)
brightness_going_down = old_brightness > current_brightness
_LOGGER.debug("'%s' state after sleep is '%s'", entity_id, current_state)
if not brightness_going_down:
break
delay = TURNING_OFF_DELAY # next time only wait this long
if transition is not None:
# Always ignore when there's a transition and light is still on.
# TODO: I am doing this because it seems like HA cannot detect
# whether a light is transitioning into 'off'. Because in my
# tests `brightness_going_down == False` even when it is actually
# still going down... Maybe needs some discussion/input?
# Always ignore when there's a 'turn_off' transition.
# Because it seems like HA cannot detect whether a light is
# transitioning into 'off'. Maybe needs some discussion/input?
return True
return current_state.state == "off"
return False
async def turn_on_off_event_listener(self, event):
"""Track 'light.turn_off(..., transition=...)' and 'light.turn_on' service calls."""
"""Track 'light.turn_off' and 'light.turn_on' service calls."""
domain = event.data.get(ATTR_DOMAIN)
if domain != LIGHT_DOMAIN:
return
@ -675,14 +654,13 @@ class TurnOnOffListener:
if service == SERVICE_TURN_OFF:
transition = service_data.get(ATTR_TRANSITION)
if transition is not None and transition > 0:
_LOGGER.debug(
"Detected an 'light.turn_off('%s', transition=%s)' event",
entity_id,
transition,
)
for eid in entity_id:
self.turn_off_event[eid] = (event.context.id, transition)
_LOGGER.debug(
"Detected an 'light.turn_off('%s', transition=%s)' event",
entity_id,
transition,
)
for eid in entity_id:
self.turn_off_event[eid] = (event.context.id, transition)
elif service == SERVICE_TURN_ON:
_LOGGER.debug("Detected an 'light.turn_on('%s')' event", entity_id)