Files
ShellyEffectsDriver/app/animations/base.py
T

77 lines
2.0 KiB
Python

"""Base class for all animations."""
from __future__ import annotations
import abc
import logging
import threading
import time
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from shelly import ShellyRGBW2
log = logging.getLogger("animation")
class Animation(abc.ABC):
"""Subclass this and implement `tick()` to create a new animation.
`tick()` is called roughly every `interval` seconds while the animation
is active. Keep it fast and non-blocking.
"""
name: str = "base"
interval: float = 0.12 # seconds between ticks
def __init__(self, shelly: "ShellyRGBW2"):
self.shelly = shelly
self._stop = threading.Event()
self._thread: threading.Thread | None = None
def start(self) -> None:
if self._thread and self._thread.is_alive():
return
self._stop.clear()
self.on_start()
self._thread = threading.Thread(
target=self._run, name=f"anim-{self.name}", daemon=True
)
self._thread.start()
log.info("Animation '%s' started", self.name)
def stop(self) -> None:
self._stop.set()
if self._thread:
self._thread.join(timeout=2.0)
self._thread = None
self.on_stop()
log.info("Animation '%s' stopped", self.name)
@property
def running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
def _run(self) -> None:
while not self._stop.is_set():
try:
self.tick()
except Exception:
log.exception("Error in animation '%s'", self.name)
self._stop.wait(self.interval)
# ---- hooks for subclasses ----
def on_start(self) -> None:
"""Called once when the animation starts (before first tick)."""
self.shelly.turn_on()
def on_stop(self) -> None:
"""Called once when the animation stops."""
pass
@abc.abstractmethod
def tick(self) -> None:
"""Called every `interval` seconds. Update the light here."""
...