packaged with rye for export_mode

src/room_control/button.py

@@ -0,0 +1,75 @@
+import json
+from dataclasses import dataclass
+from logging import Logger
+from typing import TYPE_CHECKING, List
+
+from appdaemon.plugins.mqtt.mqttapi import Mqtt
+from console import setup_component_logging
+from model import ButtonConfig
+
+if TYPE_CHECKING:
+    from room_control import RoomController
+
+
+@dataclass(init=False)
+class Button(Mqtt):
+    button: str | List[str]
+    rich: bool = False
+    config: ButtonConfig
+    logger: Logger
+
+    async def initialize(self):
+        self.config = ButtonConfig(**self.args)
+        self.logger = setup_component_logging(self)
+        self.app: 'RoomController' = await self.get_app(self.args['app'])
+        self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
+
+        self.button = self.config.button
+        self.setup_buttons(self.button)
+
+    def setup_buttons(self, buttons):
+        if isinstance(buttons, list):
+            for button in buttons:
+                self.setup_button(button)
+        else:
+            self.setup_button(buttons)
+
+    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.log(f'MQTT topic [topic]{topic}[/] controls app [room]{self.app.name}[/]')
+
+    def handle_button(self, event_name, data, kwargs):
+        try:
+            payload = json.loads(data['payload'])
+            action = payload['action']
+        except json.JSONDecodeError:
+            self.log(f'Error decoding JSON from {data["payload"]}', level='ERROR')
+        except KeyError:
+            return
+        else:
+            if isinstance(action, str) and action != '':
+                self.log(f'Action: [yellow]{action}[/]')
+                self.handle_action(action)
+
+    def handle_action(self, action: str):
+        if action == 'single':
+            state = self.get_state(self.args['ref_entity'])
+            kwargs = {'kwargs': {'cause': f'button single click: toggle while {state}'}}
+
+            if manual_entity := self.args.get('manual_mode'):
+                self.set_state(entity_id=manual_entity, state='off')
+
+            if state == 'on':
+                self.app.deactivate(**kwargs)
+            else:
+                self.app.activate(**kwargs)
+        else:
+            pass

src/room_control/console.py

@@ -0,0 +1,181 @@
+import logging
+import logging.config
+import re
+from abc import ABC
+from dataclasses import asdict
+
+from rich.console import Console
+from rich.highlighter import RegexHighlighter
+from rich.theme import Theme
+
+console = Console(
+    width=100,
+    theme=Theme(
+        {
+            'log.time': 'none',
+            # 'logging.level.info': 'none',
+            'room': 'italic bright_cyan',
+            'component': 'dark_violet',
+            'friendly_name': 'yellow',
+            'light': 'light_slate_blue',
+            'sensor': 'green',
+            'time': 'yellow',
+            'z2m': 'bright_black',
+            'topic': 'chartreuse2',
+            'true': 'green',
+            'false': 'red',
+        }
+    ),
+    log_time_format='%Y-%m-%d %I:%M:%S %p',
+    # highlighter=RCHighlighter(),
+)
+
+
+class RCHighlighter(RegexHighlighter):
+    highlights = [
+        r'(?P<light>(light|switch)\.\w+)',
+        r'(?P<time>\d+:\d+:\d+)',
+        r'(?P<z2m>zigbee2mqtt/)',
+        r'(?P<sensor>binary_sensor\.\w+)',
+        # r"'state': '(?P<on>on)|(?P<off>off)'"
+        r'(?P<true>True)|(?P<false>False)',
+    ]
+
+
+RICH_HANDLER_CFG = {
+    '()': 'rich.logging.RichHandler',
+    'markup': True,
+    'show_path': False,
+    # 'show_time': False,
+    'omit_repeated_times': False,
+    'console': console,
+    'highlighter': RCHighlighter(),
+}
+
+
+class ContextSettingFilter(logging.Filter, ABC):
+    def filter(self, record: logging.LogRecord) -> logging.LogRecord:
+        for name, val in asdict(self).items():
+            if val is not None:
+                setattr(record, name, val)
+        return record
+
+
+class RoomFilter(logging.Filter):
+    """Used to filter out messages that have a component field because they will have already been printed by their respective logger."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return getattr(record, 'component', None) is None
+
+
+# @dataclass
+# class RoomControllerFilter(ContextSettingFilter):
+#     room: str
+#     component: Optional[str] = None
+
+
+# class RoomControllerFormatter(logging.Formatter):
+#     def format(self, record: logging.LogRecord):
+#         return super().format(record)
+
+
+class UnMarkupFilter(logging.Filter):
+    md_regex = re.compile(r'(?P<open>\[.*?\])(?P<text>.*?)(?P<close>\[\/\])')
+
+    def filter(self, record: logging.LogRecord) -> logging.LogRecord:
+        record.msg = self.md_regex.sub(r'\g<text>', record.msg)
+        return record
+
+
+def room_logging_config(name: str):
+    return {
+        'version': 1,
+        'disable_existing_loggers': False,
+        'filters': {
+            'room': {'()': 'console.RoomFilter'},
+            'unmarkup': {'()': 'console.UnMarkupFilter'},
+        },
+        'formatters': {
+            'rich_room': {
+                'style': '{',
+                'format': '[room]{room}[/] {message}',
+                'datefmt': '%H:%M:%S.%f',
+            },
+            'file': {
+                'style': '{',
+                'format': '{asctime}.{msecs:03.0f} {levelname:8} {name}: {message}',
+                'datefmt': '%Y-%m-%d %H:%M:%S',
+            },
+        },
+        'handlers': {
+            'rich_room': {
+                'formatter': 'rich_room',
+                'filters': ['room'],
+                **RICH_HANDLER_CFG,
+            },
+            'file': {
+                'filters': ['unmarkup'],
+                'formatter': 'file',
+                'class': 'logging.handlers.RotatingFileHandler',
+                # 'class': 'logging.FileHandler',
+                'filename': f'/logs/{name}.log',
+                'mode': 'w',
+                'maxBytes': 1000000,
+                'backupCount': 3,
+            },
+        },
+        'loggers': {
+            f'AppDaemon.{name}': {
+                'level': 'INFO',
+                'propagate': False,
+                'handlers': ['rich_room', 'file'],
+            },
+        },
+    }
+
+
+def component_logging_config(parent_room: str, component: str):
+    logger_name = f'AppDaemon.{parent_room}.{component}'
+
+    LOG_CFG = {
+        'version': 1,
+        'disable_existing_loggers': False,
+        'formatters': {
+            'rich_component': {
+                'style': '{',
+                'format': '[room]{room}[/] [component]{component}[/] {message}',
+                'datefmt': '%H:%M:%S.%f',
+            },
+        },
+        'handlers': {
+            'rich_component': {
+                'formatter': 'rich_component',
+                **RICH_HANDLER_CFG,
+            },
+        },
+        'loggers': {
+            logger_name: {
+                # 'level': 'INFO',
+                'propagate': True,
+                'handlers': ['rich_component'],
+            }
+        },
+    }
+    return LOG_CFG
+
+
+def setup_component_logging(self) -> logging.Logger:
+    """Creates a logger for a subcomponent with a RichHandler"""
+    component = type(self).__name__
+    parent = self.args['app']
+    cfg_dict = component_logging_config(parent_room=parent, component=component)
+    logger_name = next(iter(cfg_dict['loggers']))
+
+    try:
+        logging.config.dictConfig(cfg_dict)
+    except Exception:
+        console.print_exception()
+    else:
+        logger = logging.getLogger(logger_name)
+        logger = logging.LoggerAdapter(logger, {'room': parent, 'component': component})
+        return logger

src/room_control/door.py

@@ -0,0 +1,21 @@
+from logging import Logger
+from appdaemon.plugins.hass.hassapi import Hass
+from console import setup_component_logging
+
+from room_control import RoomController
+
+
+class Door(Hass):
+    logger: Logger
+
+    async def initialize(self):
+        self.logger = setup_component_logging(self)
+        self.app: RoomController = await self.get_app(self.args['app'])
+        self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
+
+        await self.listen_state(
+            self.app.activate_all_off,
+            entity_id=self.args['door'],
+            new='on',
+            cause='door open',
+        )

src/room_control/model.py

@@ -0,0 +1,143 @@
+from datetime import datetime, time, timedelta
+from pathlib import Path
+from typing import Annotated, Dict, List, Optional, Self
+
+import yaml
+from astral import SunDirection
+from pydantic import BaseModel, BeforeValidator, Field, root_validator
+from pydantic_core import PydanticCustomError
+from rich.console import Console, ConsoleOptions, RenderResult
+from rich.table import Column, Table
+
+
+def str_to_timedelta(input_str: str) -> timedelta:
+    try:
+        hours, minutes, seconds = map(int, input_str.split(':'))
+        return timedelta(hours=hours, minutes=minutes, seconds=seconds)
+    except Exception:
+        return timedelta()
+
+
+def str_to_direction(input_str: str) -> SunDirection:
+    try:
+        return getattr(SunDirection, input_str.upper())
+    except AttributeError:
+        raise PydanticCustomError(
+            'invalid_dir', 'Invalid sun direction: {dir}', dict(dir=input_str)
+        )
+
+
+OffDuration = Annotated[timedelta, BeforeValidator(str_to_timedelta)]
+
+
+class State(BaseModel):
+    state: bool = True
+    brightness: Optional[int] = Field(default=None, ge=1, le=255)
+    color_temp: Optional[int] = Field(default=None, ge=200, le=650)
+    rgb_color: Optional[list[int]] = Field(default=None, min_length=3, max_length=3)
+
+
+class ApplyKwargs(BaseModel):
+    """Arguments to call with the 'scene/apply' service"""
+
+    entities: Dict[str, State]
+    transition: Optional[int] = None
+
+
+class ControllerStateConfig(BaseModel):
+    time: Optional[str | datetime] = None
+    elevation: Optional[float] = None
+    direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = (
+        None
+    )
+    off_duration: Optional[OffDuration] = None
+    scene: dict[str, State] | str
+
+    @root_validator(pre=True)
+    def check_args(cls, values):
+        time, elevation = values.get('time'), values.get('elevation')
+        if time is not None and elevation is not None:
+            raise PydanticCustomError(
+                'bad_time_spec', 'Only one of time or elevation can be set.'
+            )
+        elif elevation is not None and 'direction' not in values:
+            raise PydanticCustomError(
+                'no_sun_dir', 'Needs sun direction with elevation'
+            )
+        return values
+
+    def to_apply_kwargs(self, **kwargs):
+        return ApplyKwargs(entities=self.scene, **kwargs).model_dump(exclude_none=True)
+
+
+class RoomControllerConfig(BaseModel):
+    states: List[ControllerStateConfig] = Field(default_factory=list)
+    off_duration: Optional[OffDuration] = None
+    sleep_state: Optional[ControllerStateConfig] = None
+    rich: Optional[str] = None
+    manual_mode: Optional[str] = None
+
+    @classmethod
+    def from_yaml(cls: Self, yaml_path: Path) -> Self:
+        yaml_path = Path(yaml_path)
+        with yaml_path.open('r') as f:
+            for appname, app_cfg in yaml.load(f, Loader=yaml.SafeLoader).items():
+                if app_cfg['class'] == 'RoomController':
+                    return cls.model_validate(app_cfg)
+
+    def __rich_console__(
+        self, console: Console, options: ConsoleOptions
+    ) -> RenderResult:
+        table = Table(
+            Column('Time', width=15),
+            Column('Scene'),
+            highlight=True,
+            padding=1,
+            collapse_padding=True,
+        )
+        for state in self.states:
+            if isinstance(state, str):
+                pass
+            elif isinstance(state, dict):
+                scene_json = state.to_apply_kwargs()
+                lines = [
+                    f'{name:20}{state["state"]}   Brightness: {state.get("brightness", ""):<4}  Temp: {state.get("color_temp", "")}'
+                    for name, state in scene_json['entities'].items()
+                ]
+                table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines))
+        yield table
+
+    def sort_states(self):
+        """Should only be called after all the times have been resolved"""
+        assert all(
+            isinstance(state.time, time) for state in self.states
+        ), 'Times have not all been resolved yet'
+        self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
+
+    def current_state(self, now: time) -> ControllerStateConfig:
+        self.sort_states()
+        for state in self.states:
+            if state.time <= now:
+                return state
+        else:
+            return self.states[0]
+
+    def current_scene(self, now: time) -> Dict:
+        state = self.current_state(now)
+        return state.scene
+
+    def current_off_duration(self, now: time) -> timedelta:
+        state = self.current_state(now)
+        if state.off_duration is None:
+            if self.off_duration is None:
+                raise ValueError('Need an off duration')
+            else:
+                return self.off_duration
+        else:
+            return state.off_duration
+
+
+class ButtonConfig(BaseModel):
+    app: str
+    button: str | List[str]
+    ref_entity: str

src/room_control/motion.py

@@ -0,0 +1,182 @@
+import re
+from datetime import timedelta
+from logging import Logger
+from typing import TYPE_CHECKING, Literal, Optional
+
+from appdaemon.entity import Entity
+from appdaemon.plugins.hass.hassapi import Hass
+from console import setup_component_logging
+from pydantic import BaseModel, TypeAdapter
+
+if TYPE_CHECKING:
+    from room_control import RoomController
+
+
+class CallbackEntry(BaseModel):
+    entity: str
+    event: Optional[str] = None
+    type: Literal['state', 'event']
+    kwargs: str
+    function: str
+    name: str
+    pin_app: bool
+    pin_thread: int
+
+
+Callbacks = dict[str, dict[str, CallbackEntry]]
+
+
+class Motion(Hass):
+    logger: Logger
+    app: 'RoomController'
+
+    @property
+    def sensor(self) -> Entity:
+        return self.get_entity(self.args['sensor'])
+
+    @property
+    def sensor_state(self) -> bool:
+        return self.sensor.state == 'on'
+
+    @property
+    def ref_entity(self) -> Entity:
+        return self.get_entity(self.args['ref_entity'])
+
+    @property
+    def ref_entity_state(self) -> bool:
+        return self.ref_entity.get_state() == 'on'
+
+    @property
+    def state_mismatch(self) -> bool:
+        return self.sensor_state != self.ref_entity_state
+
+    def initialize(self):
+        self.logger = setup_component_logging(self)
+        self.app: 'RoomController' = self.get_app(self.args['app'])
+        self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG')
+
+        assert self.entity_exists(self.args['sensor'])
+        assert self.entity_exists(self.args['ref_entity'])
+
+        base_kwargs = dict(
+            entity_id=self.ref_entity.entity_id,
+            immediate=True,  # avoids needing to sync the state
+        )
+
+        if self.state_mismatch:
+            self.log(
+                f'Sensor is {self.sensor_state} '
+                f'and light is {self.ref_entity_state}',
+                level='WARNING',
+            )
+            if self.sensor_state:
+                self.app.activate(
+                    kwargs={'cause': f'Syncing state with {self.sensor.entity_id}'}
+                )
+
+        # don't need to await these because they'll already get turned into a task by the utils.sync_wrapper decorator
+        self.listen_state(
+            **base_kwargs,
+            attribute='brightness',
+            callback=self.callback_light_on,
+        )
+        self.listen_state(**base_kwargs, new='off', callback=self.callback_light_off)
+
+        if callbacks := self.callbacks():
+            for handle, entry in callbacks.items():
+                self.log(
+                    f'Handle [yellow]{handle[:4]}[/]: {entry.function}', level='DEBUG'
+                )
+
+    def callbacks(self):
+        """Returns a dictionary of validated CallbackEntry objects that are associated with this app"""
+        data = TypeAdapter(Callbacks).validate_python(self.get_callback_entries())
+        return data.get(self.name, {})
+
+    def listen_motion_on(self):
+        """Sets up the motion on callback to activate the room"""
+        self.cancel_motion_callback()
+        self.listen_state(
+            callback=self.app.activate_all_off,
+            entity_id=self.sensor.entity_id,
+            new='on',
+            oneshot=True,
+            cause='motion on',
+        )
+        self.log(
+            f'Waiting for sensor motion on [friendly_name]{self.sensor.friendly_name}[/]'
+        )
+        if self.sensor_state:
+            self.log(
+                f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is already on',
+                level='WARNING',
+            )
+
+    def listen_motion_off(self, duration: timedelta):
+        """Sets up the motion off callback to deactivate the room"""
+        self.cancel_motion_callback()
+        self.listen_state(
+            callback=self.app.deactivate,
+            entity_id=self.sensor.entity_id,
+            new='off',
+            duration=duration.total_seconds(),
+            oneshot=True,
+            cause='motion off',
+        )
+        self.log(
+            f'Waiting for sensor [friendly_name]{self.sensor.friendly_name}[/] to be clear for {duration}'
+        )
+
+        if not self.sensor_state:
+            self.log(
+                f'Sensor [friendly_name]{self.sensor.friendly_name}[/] is currently off',
+                level='WARNING',
+            )
+
+    def callback_light_on(
+        self, entity=None, attribute=None, old=None, new=None, kwargs=None
+    ):
+        """Called when the light turns on"""
+        if new is not None:
+            self.log(f'Detected {entity} turning on', level='DEBUG')
+            duration = self.app.off_duration()
+            self.listen_motion_off(duration)
+
+    def callback_light_off(
+        self, entity=None, attribute=None, old=None, new=None, kwargs=None
+    ):
+        """Called when the light turns off"""
+        self.log(f'Detected {entity} turning off', level='DEBUG')
+        self.listen_motion_on()
+
+    def get_app_callbacks(self, name: str = None):
+        """Gets all the callbacks associated with the app"""
+        name = name or self.name
+        callbacks = {
+            handle: info
+            for app_name, callbacks in self.get_callback_entries().items()
+            for handle, info in callbacks.items()
+            if app_name == name
+        }
+        return callbacks
+
+    def get_sensor_callbacks(self):
+        return {
+            handle: info
+            for handle, info in self.get_app_callbacks().items()
+            if info['entity'] == self.sensor.entity_id
+        }
+
+    def cancel_motion_callback(self):
+        callbacks = self.get_sensor_callbacks()
+        # self.log(f'Found {len(callbacks)} callbacks for {self.sensor.entity_id}')
+        for handle, info in callbacks.items():
+            entity = info['entity']
+            kwargs = info['kwargs']
+            if (m := re.match('new=(?P<new>.*?)\s', kwargs)) is not None:
+                new = m.group('new')
+                self.cancel_listen_state(handle)
+                self.log(
+                    f'cancelled callback for sensor {entity} turning {new}',
+                    level='DEBUG',
+                )

src/room_control/room_control.py

@@ -0,0 +1,247 @@
+import datetime
+import logging
+import logging.config
+from copy import deepcopy
+from typing import Dict, List
+
+from appdaemon.entity import Entity
+from appdaemon.plugins.hass.hassapi import Hass
+from appdaemon.plugins.mqtt.mqttapi import Mqtt
+from console import console, room_logging_config
+from model import ControllerStateConfig, RoomControllerConfig
+
+logger = logging.getLogger(__name__)
+
+
+class RoomController(Hass, Mqtt):
+    """Class for linking room's lights with a motion sensor.
+
+    - Separate the turning on and turning off functions.
+    - Use the state change of the light to set up the event for changing to the other state
+        - `handle_on`
+        - `handle_off`
+    - When the light comes on, check if it's attributes match what they should, given the time.
+    """
+
+    @property
+    def states(self) -> List[ControllerStateConfig]:
+        return self._room_config.states
+
+    @states.setter
+    def states(self, new: List[ControllerStateConfig]):
+        assert all(isinstance(s, ControllerStateConfig) for s in new), f'Invalid: {new}'
+        self._room_config.states = new
+
+    def configure_logging(self) -> dict:
+        logging.config.dictConfig(room_logging_config(self.name))
+        self.logger = logging.LoggerAdapter(
+            logging.getLogger(f'AppDaemon.{self.name}'), {'room': self.name}
+        )
+
+    def initialize(self):
+        self.configure_logging()
+        self.app_entities = self.gather_app_entities()
+        # self.log(f'entities: {self.app_entities}')
+        self.refresh_state_times()
+        self.run_daily(callback=self.refresh_state_times, start='00:00:00')
+        self.log(f'Initialized [bold green]{type(self).__name__}[/]')
+
+    def terminate(self):
+        self.log('[bold red]Terminating[/]', level='DEBUG')
+
+    def gather_app_entities(self) -> List[str]:
+        """Returns a list of all the entities involved in any of the states"""
+
+        def generator():
+            for settings in deepcopy(self.args['states']):
+                if scene := settings.get('scene'):
+                    if isinstance(scene, str):
+                        assert scene.startswith(
+                            'scene.'
+                        ), f"Scene definition must start with 'scene.' for app {self.name}"
+                        entity: Entity = self.get_entity(scene)
+                        entity_state = entity.get_state('all')
+                        attributes = entity_state['attributes']
+                        for entity in attributes['entity_id']:
+                            yield entity
+                    else:
+                        for key in scene.keys():
+                            yield key
+                else:
+                    yield self.args['entity']
+
+        return set(list(generator()))
+
+    def refresh_state_times(self, *args, **kwargs):
+        """Resets the `self.states` attribute to a newly parsed version of the states.
+
+        Parsed states have an absolute time for the current day.
+        """
+        # re-parse the state strings into times for the current day
+        self._room_config = RoomControllerConfig.model_validate(self.args)
+        self.log(
+            f'{len(self._room_config.states)} states in the app configuration',
+            level='DEBUG',
+        )
+
+        for state in self._room_config.states:
+            if state.time is None and state.elevation is not None:
+                state.time = self.AD.sched.location.time_at_elevation(
+                    elevation=state.elevation, direction=state.direction
+                ).time()
+            elif isinstance(state.time, str):
+                state.time = self.parse_time(state.time)
+
+            assert isinstance(state.time, datetime.time), f'Invalid time: {state.time}'
+
+        # if self.logger.isEnabledFor(logging.DEBUG):
+        #     # table = self._room_config.rich_table(self.name)
+        #     console.print(self._room_config)
+
+        self.states = sorted(self.states, key=lambda s: s.time, reverse=True)
+
+        # schedule the transitions
+        for state in self.states[::-1]:
+            # t: datetime.time = state['time']
+            t: datetime.time = state.time
+            try:
+                self.run_at(
+                    callback=self.activate_any_on,
+                    start=t.strftime('%H:%M:%S'),
+                    cause='scheduled transition',
+                )
+            except ValueError:
+                # happens when the callback time is in the past
+                pass
+            except Exception as e:
+                self.log(f'Failed with {type(e)}: {e}')
+
+    def current_state(self, now: datetime.time = None) -> ControllerStateConfig:
+        if self.sleep_bool():
+            self.log('sleep: active')
+            if state := self.args.get('sleep_state'):
+                return ControllerStateConfig(**state)
+            else:
+                return ControllerStateConfig(scene={})
+        else:
+            now = now or self.get_now().time().replace(microsecond=0)
+            self.log(f'Getting state for {now.strftime("%I:%M:%S %p")}', level='DEBUG')
+
+            state = self._room_config.current_state(now)
+            self.log(f'Current state: {state.time}', level='DEBUG')
+            return state
+
+    def app_entity_states(self) -> Dict[str, str]:
+        states = {entity: self.get_state(entity) for entity in self.app_entities}
+        return states
+
+    def all_off(self) -> bool:
+        """ "All off" is the logic opposite of "any on"
+
+        Returns:
+            bool: Whether all the lights associated with the app are off
+        """
+        states = self.app_entity_states()
+        return all(state != 'on' for entity, state in states.items())
+
+    def any_on(self) -> bool:
+        """ "Any on" is the logic opposite of "all off"
+
+        Returns:
+            bool: Whether any of the lights associated with the app are on
+        """
+        states = self.app_entity_states()
+        return any(state == 'on' for entity, state in states.items())
+
+    def sleep_bool(self) -> bool:
+        if sleep_var := self.args.get('sleep'):
+            return self.get_state(sleep_var) == 'on'
+        else:
+            return False
+
+    def manual_mode(self) -> bool:
+        if manual_entity := self.args.get('manual_mode'):
+            return self.get_state(manual_entity) == 'on'
+        else:
+            return False
+
+    # @sleep_bool.setter
+    # def sleep_bool(self, val) -> bool:
+    #     if (sleep_var := self.args.get('sleep')):
+    #         if isinstance(val, str):
+    #             self.set_state(sleep_var, state=val)
+    #         elif isinstance(val, bool):
+    #             self.set_state(sleep_var, state='on' if val else 'off')
+    #     else:
+    #         raise ValueError('Sleep variable is undefined')
+
+    def off_duration(self, now: datetime.time = None) -> datetime.timedelta:
+        """Determines the time that the motion sensor has to be clear before deactivating
+
+        Priority:
+        - Value in scene definition
+        - Default value
+            - Normal - value in app definition
+            - Sleep - 0
+
+        """
+        sleep_mode_active = self.sleep_bool()
+        if sleep_mode_active:
+            self.log(f'Sleeping mode active: {sleep_mode_active}')
+            return datetime.timedelta()
+        else:
+            now = now or self.get_now().time()
+            return self._room_config.current_off_duration(now)
+
+    def activate(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
+        if kwargs is not None:
+            cause = kwargs.get('cause', 'unknown')
+        else:
+            cause = 'unknown'
+
+        self.log(f'Activating: {cause}')
+        scene_kwargs = self.current_state().to_apply_kwargs(transition=0)
+
+        if isinstance(scene_kwargs, str):
+            self.turn_on(scene_kwargs)
+            self.log(f'Turned on scene: {scene_kwargs}')
+
+        elif isinstance(scene_kwargs, dict):
+            self.call_service('scene/apply', **scene_kwargs)
+            if self.logger.isEnabledFor(logging.INFO):
+                self.log('Applied scene:')
+                console.print(scene_kwargs['entities'])
+
+        elif scene_kwargs is None:
+            self.log('No scene, ignoring...')
+            # Need to act as if the light had just turned off to reset the motion (and maybe other things?)
+            # self.callback_light_off()
+        else:
+            self.log(f'ERROR: unknown scene: {scene_kwargs}')
+
+    def activate_all_off(self, *args, **kwargs):
+        """Activate if all of the entities are off. Args and kwargs are passed directly to self.activate()"""
+        if self.all_off():
+            self.activate(*args, **kwargs)
+        else:
+            self.log('Skipped activating - everything is not off')
+
+    def activate_any_on(self, *args, **kwargs):
+        """Activate if any of the entities are on. Args and kwargs are passed directly to self.activate()"""
+        if self.any_on() and not self.manual_mode():
+            self.activate(*args, **kwargs)
+        else:
+            self.log('Skipped activating - everything is off')
+
+    def toggle_activate(self, *args, **kwargs):
+        if self.any_on():
+            self.deactivate(*args, **kwargs)
+        else:
+            self.activate(*args, **kwargs)
+
+    def deactivate(self, entity=None, attribute=None, old=None, new=None, kwargs=None):
+        cause = kwargs.get('cause', 'unknown')
+        self.log(f'Deactivating: {cause}')
+        for e in self.app_entities:
+            self.turn_off(e)
+            self.log(f'Turned off {e}')