Skip to content

Commit

Permalink
#510 changes (#516)
Browse files Browse the repository at this point in the history 
* Change (WIP)

* Update test_switch.py

* Refactor

* Revert "Revert "Small refactor""

This reverts commit 3731c99.

* Update README.md

* Fix the test

* Bump to 1.9.0 (#518)

* Basic community fixes PR (#460)

* Fixes #423

#423

* Do not adapt lights turned on with custom payloads.

* Update switch.py

* Issue fixes

#423, #378, #403, #449

* quickly test #274

* Revert feature requests, this branch only has fixes.

Reverted FR 274

* pre-commit fix

* Create automerge.yaml

* test

* Delete automerge.yaml

My bad.

* Fix #460 and #408

* see @basnijholt 's comment in #450.

* @basnijholt requested changes.

---------

Co-authored-by: Bas Nijholt <[email protected]>

* Undo accidental changes introduced in #509, but adds the changes from #460 (#521)

* Release 1.9.1 (#522)

* Bump to 1.9.1

* Add CODEOWNERS

---------

Co-authored-by: Benjamin Auquite <[email protected]>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
  • Loading branch information
@basnijholt @th3w1zard1 @github-actions
3 people authored Apr 3, 2023
1 parent 445b5da commit 972d0df
Showing 1 changed file with 69 additions and 68 deletions.
137 changes: 69 additions & 68 deletions custom_components/adaptive_lighting/switch.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import base64
import bisect
from collections import defaultdict
from collections.abc import Callable, Coroutine
from copy import deepcopy
from dataclasses import dataclass
import datetime
Expand Down Expand Up @@ -1144,18 +1145,19 @@ async def _update_attrs_and_maybe_adapt_lights(
if lights is None:
lights = self._lights

if not force:
if self._only_once:
return
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
if self.turn_on_off_listener.transition_timers.get(light):
lights.remove(light)
if not force and self._only_once:
return

filtered_lights = []
for light in lights:
# Don't adapt lights that haven't finished prior transitions.
if force or not self.turn_on_off_listener.transition_timers.get(light):
filtered_lights.append(light)

if not len(lights):
if not filtered_lights:
return

await self._adapt_lights(lights, transition, force, context)
await self._adapt_lights(filtered_lights, transition, force, context)

async def _adapt_lights(
self,
Expand Down Expand Up @@ -1580,6 +1582,30 @@ def __init__(self, hass: HomeAssistant):
EVENT_STATE_CHANGED, self.state_changed_event_listener
)

def _handle_timer(
self,
light: str,
timers_dict: dict[str, _AsyncSingleShotTimer],
delay: float | None,
reset_coroutine: Callable[[], Coroutine[Any, Any, None]],
) -> None:
timer = timers_dict.get(light)
if timer is not None:
if delay is None: # Timer object exists, but should not anymore
timer.cancel()
timers_dict.pop(light)
else: # Timer object already exists, just update the delay and restart it
timer.delay = delay
timer.start()
elif delay is not None: # Timer object does not exist, create it

async def reset_wrapper():
await reset_coroutine()

timer = _AsyncSingleShotTimer(delay, reset_wrapper)
timers_dict[light] = timer
timer.start()

def start_transition_timer(self, light: str) -> None:
"""Mark a light as manually controlled."""
_LOGGER.debug("Start transition timer for %s", light)
Expand All @@ -1589,39 +1615,26 @@ def start_transition_timer(self, light: str) -> None:
or light not in last_service_data
or ATTR_TRANSITION not in last_service_data[light]
):
return False
return

delay = last_service_data[light][ATTR_TRANSITION]
timer = self.transition_timers.get(light)
if timer is not None:
if delay is None: # Timer object exists, but should not anymore
timer.cancel()
self.transition_timers.pop(light)
else: # Timer object already exists, just update the delay and restart it
timer.delay = delay
timer.start()
elif delay is not None: # Timer object does not exist, create it

async def reset():
_LOGGER.debug(
"Transition finished for light %s",
light,
async def reset():
_LOGGER.debug(
"Transition finished for light %s",
light,
)
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
await switch._update_attrs_and_maybe_adapt_lights(
[light],
force=False,
context=switch.create_context("transit"),
)
# This part is optional, we could just wait for the next interval.
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
# pylint: disable=protected-access
await switch._update_attrs_and_maybe_adapt_lights(
[light],
force=False,
context=switch.create_context("transit"),
)

timer = _AsyncSingleShotTimer(delay, reset)
self.transition_timers[light] = timer
timer.start()
self._handle_timer(light, self.transition_timers, delay, reset)

def set_auto_reset_manual_control_times(self, lights: list[str], time: float):
"""Set the time after which the lights are automatically reset."""
Expand All @@ -1645,40 +1658,28 @@ def mark_as_manual_control(self, light: str) -> None:
_LOGGER.debug("Marking '%s' as manually controlled.", light)
self.manual_control[light] = True
delay = self.auto_reset_manual_control_times.get(light)
timer = self.auto_reset_manual_control_timers.get(light)
if timer is not None:
if delay is None: # Timer object exists, but should not anymore
timer.cancel()
self.auto_reset_manual_control_timers.pop(light)
else: # Timer object already exists, just update the delay and restart it
timer.delay = delay
timer.start()
elif delay is not None: # Timer object does not exist, create it

async def reset():
self.reset(light)
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
# pylint: disable=protected-access
await switch._update_attrs_and_maybe_adapt_lights(
[light],
transition=switch._initial_transition,
force=True,
context=switch.create_context("autoreset"),
)
_LOGGER.debug(
"Auto resetting 'manual_control' status of '%s' because"
" it was not manually controlled for %s seconds.",
light,
delay,
async def reset():
self.reset(light)
switches = _get_switches_with_lights(self.hass, [light])
for switch in switches:
if not switch.is_on:
continue
await switch._update_attrs_and_maybe_adapt_lights(
[light],
transition=switch._initial_transition,
force=True,
context=switch.create_context("autoreset"),
)
assert not self.manual_control[light]
_LOGGER.debug(
"Auto resetting 'manual_control' status of '%s' because"
" it was not manually controlled for %s seconds.",
light,
delay,
)
assert not self.manual_control[light]

timer = _AsyncSingleShotTimer(delay, reset)
self.auto_reset_manual_control_timers[light] = timer
timer.start()
self._handle_timer(light, self.auto_reset_manual_control_timers, delay, reset)

def reset(self, *lights, reset_manual_control=True) -> None:
"""Reset the 'manual_control' status of the lights."""
Expand Down

0 comments on commit 972d0df

Please sign in to comment.