380 lines
14 KiB
Python
380 lines
14 KiB
Python
"""Service de controle de l'anneau LED WS2812b (GPIO18, 35 LEDs).
|
|
|
|
Mode reel : rpi_ws281x (Raspberry Pi, root ou /dev/mem)
|
|
Mode mock : log uniquement (developpement)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import math
|
|
import threading
|
|
import time
|
|
from typing import Callable
|
|
|
|
from backend.services.config_service import Config, LEDConfig
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
from rpi_ws281x import PixelStrip, Color as WS_Color
|
|
HAS_WS281X = True
|
|
except ImportError:
|
|
HAS_WS281X = False
|
|
logger.warning("rpi_ws281x non disponible -- mode mock LED")
|
|
|
|
class WS_Color: # type: ignore
|
|
def __init__(self, r: int, g: int, b: int):
|
|
self.r, self.g, self.b = r, g, b
|
|
def __repr__(self):
|
|
return f"Color({self.r},{self.g},{self.b})"
|
|
|
|
class PixelStrip: # type: ignore
|
|
def __init__(self, *a, **kw): pass
|
|
def begin(self): pass
|
|
def show(self): pass
|
|
def setPixelColor(self, i, c): pass
|
|
def numPixels(self): return 35
|
|
def setBrightness(self, b): pass
|
|
|
|
|
|
def _color(rgb: list[int]) -> WS_Color:
|
|
return WS_Color(int(rgb[0]), int(rgb[1]), int(rgb[2]))
|
|
|
|
|
|
class LEDService:
|
|
EFFECTS = ("idle", "countdown", "capture", "captured", "finished",
|
|
"printing", "error", "disabled", "off")
|
|
|
|
def __init__(self, config: Config):
|
|
self._cfg: LEDConfig = config.leds
|
|
self._strip: PixelStrip | None = None
|
|
self._thread: threading.Thread | None = None
|
|
self._stop_event = threading.Event()
|
|
self._cmd_event = threading.Event()
|
|
self._current_effect: str = "idle"
|
|
self._countdown_duration: float = 5.0
|
|
self._lock = threading.Lock()
|
|
self._on_change_cb: Callable | None = None
|
|
self._loop: asyncio.AbstractEventLoop | None = None
|
|
self._ambient_lux: float = -1.0 # -1 = inconnu
|
|
|
|
# ── Lifecycle ─────────────────────────────────────────────────────────────
|
|
|
|
def start(self):
|
|
if HAS_WS281X:
|
|
self._strip = PixelStrip(
|
|
self._cfg.count,
|
|
self._cfg.pin,
|
|
self._cfg.freq_hz,
|
|
self._cfg.dma,
|
|
False,
|
|
self._cfg.brightness,
|
|
0,
|
|
)
|
|
try:
|
|
self._strip.begin()
|
|
logger.info("Strip WS2812b initialise (%d LEDs, GPIO%d)", self._cfg.count, self._cfg.pin)
|
|
except Exception as e:
|
|
logger.error("Erreur init strip LED: %s", e)
|
|
self._strip = PixelStrip()
|
|
else:
|
|
self._strip = PixelStrip()
|
|
|
|
self._stop_event.clear()
|
|
self._thread = threading.Thread(target=self._run, daemon=True, name="led-service")
|
|
self._thread.start()
|
|
|
|
def stop(self):
|
|
self._stop_event.set()
|
|
self._cmd_event.set()
|
|
if self._thread:
|
|
self._thread.join(timeout=2)
|
|
self._all_off()
|
|
|
|
def set_loop(self, loop: asyncio.AbstractEventLoop):
|
|
self._loop = loop
|
|
|
|
def set_on_change(self, cb: Callable):
|
|
self._on_change_cb = cb
|
|
|
|
# ── Commandes publiques ───────────────────────────────────────────────────
|
|
|
|
def play(self, effect: str, countdown_duration: float = 5.0):
|
|
with self._lock:
|
|
self._current_effect = effect
|
|
self._countdown_duration = countdown_duration
|
|
self._cmd_event.set()
|
|
logger.debug("LED effet: %s", effect)
|
|
self._notify_change(effect)
|
|
|
|
def set_color(self, r: int, g: int, b: int):
|
|
self._fill(WS_Color(r, g, b))
|
|
|
|
def set_brightness(self, value: int):
|
|
"""Regle la luminosite globale (0-255) et l'applique immediatement."""
|
|
value = max(0, min(255, value))
|
|
self._cfg.brightness = value
|
|
if HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(value)
|
|
self._strip.show()
|
|
logger.info("Luminosite LED: %d", value)
|
|
|
|
def set_ambient_lux(self, lux: float):
|
|
"""Definit la luminosite ambiante mesuree (0-255 echelle pixel)."""
|
|
self._ambient_lux = lux
|
|
logger.debug("Lux ambiant: %.1f", lux)
|
|
|
|
# ── Thread principal ──────────────────────────────────────────────────────
|
|
|
|
def _run(self):
|
|
effect_func = {
|
|
"idle": self._effect_idle,
|
|
"countdown": self._effect_countdown,
|
|
"capture": self._effect_capture,
|
|
"captured": self._effect_solid,
|
|
"finished": self._effect_finished,
|
|
"printing": self._effect_spin,
|
|
"error": self._effect_error,
|
|
"disabled": self._effect_solid,
|
|
"off": self._effect_off,
|
|
}
|
|
|
|
while not self._stop_event.is_set():
|
|
self._cmd_event.clear()
|
|
with self._lock:
|
|
current = self._current_effect
|
|
cd_dur = self._countdown_duration
|
|
|
|
fn = effect_func.get(current, self._effect_idle)
|
|
|
|
try:
|
|
if current == "countdown":
|
|
fn(cd_dur)
|
|
else:
|
|
fn()
|
|
except Exception as e:
|
|
logger.error("Erreur effet LED '%s': %s", current, e)
|
|
|
|
if not self._cmd_event.is_set() and not self._stop_event.is_set():
|
|
with self._lock:
|
|
if self._current_effect == current:
|
|
if current in ("capture", "captured", "finished", "error"):
|
|
self._current_effect = "idle"
|
|
self._cmd_event.wait(timeout=0.1)
|
|
|
|
# ── Effets ───────────────────────────────────────────────────────────────
|
|
|
|
def _effect_idle(self):
|
|
cfg = self._cfg.get_effect("idle")
|
|
c = cfg.color
|
|
|
|
if cfg.mode == "solid":
|
|
self._fill(WS_Color(c[0], c[1], c[2]))
|
|
self._cmd_event.wait()
|
|
return
|
|
|
|
# mode "breathe" (defaut)
|
|
speed = cfg.speed
|
|
step = 0
|
|
while not self._cmd_event.is_set() and not self._stop_event.is_set():
|
|
t = (1 + math.sin(step * 0.1)) / 2
|
|
brightness = max(0.05, t)
|
|
self._fill(WS_Color(
|
|
int(c[0] * brightness),
|
|
int(c[1] * brightness),
|
|
int(c[2] * brightness),
|
|
))
|
|
time.sleep(speed)
|
|
step += 1
|
|
|
|
def _effect_countdown(self, duration: float = 5.0):
|
|
n = self._cfg.count
|
|
blank = WS_Color(0, 0, 0)
|
|
|
|
# Mode sombre : rampe progressive avec couleur flash pour stabiliser AWB caméra
|
|
lux = self._ambient_lux
|
|
dark_mode = self._cfg.auto_brightness and lux >= 0 and lux < 50
|
|
|
|
if dark_mode:
|
|
flash_cfg = self._cfg.get_effect("capture")
|
|
fc = flash_cfg.color
|
|
color = WS_Color(fc[0], fc[1], fc[2])
|
|
# Brightness cible du flash
|
|
target_br = 255 if lux < 50 else (int(255 - (lux - 50) / 130 * 135) if lux < 180 else 80)
|
|
# Rampe : atteindre target_br quand il reste pre_flash_advance secondes
|
|
ramp_duration = max(0.1, duration - self._cfg.pre_flash_advance)
|
|
orig_br = self._cfg.brightness
|
|
logger.info("Countdown mode sombre: rampe 0→%d sur %.1fs puis stable", target_br, ramp_duration)
|
|
if HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(0)
|
|
self._fill(color)
|
|
else:
|
|
cfg = self._cfg.get_effect("countdown")
|
|
c = cfg.color
|
|
color = WS_Color(c[0], c[1], c[2])
|
|
self._fill(blank)
|
|
|
|
t_start = time.time()
|
|
while not self._cmd_event.is_set() and not self._stop_event.is_set():
|
|
elapsed = time.time() - t_start
|
|
ratio = min(elapsed / duration, 1.0)
|
|
|
|
if dark_mode:
|
|
# Rampe de 0 → target_br jusqu'à ramp_duration, puis stable à target_br
|
|
if elapsed < ramp_duration:
|
|
br = int(target_br * elapsed / ramp_duration)
|
|
else:
|
|
br = target_br
|
|
if HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(br)
|
|
self._strip.show()
|
|
else:
|
|
leds_on = int(ratio * n)
|
|
for i in range(n):
|
|
self._strip.setPixelColor(i, color if i < leds_on else blank)
|
|
self._strip.show()
|
|
|
|
if ratio >= 1.0:
|
|
break
|
|
time.sleep(0.04)
|
|
|
|
# En mode sombre, restaurer brightness normale (capture le fera aussi, mais par sécurité)
|
|
if dark_mode and HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(orig_br)
|
|
|
|
def _effect_capture(self):
|
|
"""Flash photo.
|
|
|
|
Utilise la couleur configuree dans settings.yaml > leds > effects > capture.
|
|
Booste la luminosite au maximum pendant le flash pour maximiser l'eclairage,
|
|
puis la restaure a la valeur normale.
|
|
|
|
Conseil couleur WS2812b : les LEDs bleues sont plus efficaces que les rouges.
|
|
Pour un blanc neutre/chaud utilisez par ex. [255, 180, 60] au lieu de [255,255,255].
|
|
"""
|
|
cfg = self._cfg.get_effect("capture")
|
|
flash_color = WS_Color(cfg.color[0], cfg.color[1], cfg.color[2])
|
|
off = WS_Color(0, 0, 0)
|
|
|
|
# Calcul de la luminosite du flash
|
|
# Si auto_brightness activee et lux connu, on adapte. Sinon boost a 255.
|
|
if self._cfg.auto_brightness and self._ambient_lux >= 0:
|
|
lux = self._ambient_lux
|
|
if lux < 50:
|
|
flash_brightness = 255 # salle sombre → flash plein
|
|
elif lux < 180:
|
|
# lineaire : 255 a 50 lux → 120 a 180 lux
|
|
flash_brightness = int(255 - (lux - 50) / 130 * 135)
|
|
else:
|
|
flash_brightness = 80 # salle tres claire → flash reduit
|
|
logger.debug("Flash brightness auto: %d (lux=%.1f)", flash_brightness, lux)
|
|
else:
|
|
flash_brightness = 255
|
|
|
|
if HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(flash_brightness)
|
|
|
|
# Le flash démarre pre_flash_advance secondes AVANT la photo.
|
|
# La durée ON doit couvrir ce délai + la durée d'exposition.
|
|
# Sinon les LEDs s'éteignent avant que la photo soit prise.
|
|
effective_on_time = self._cfg.pre_flash_advance + cfg.flash_duration
|
|
|
|
try:
|
|
for i in range(cfg.flashes):
|
|
if self._cmd_event.is_set():
|
|
break
|
|
self._fill(flash_color)
|
|
# Premier flash : durée étendue pour couvrir le délai + expo
|
|
# Flashs suivants : durée normale (décoratif post-capture)
|
|
on_time = effective_on_time if i == 0 else cfg.flash_duration
|
|
time.sleep(on_time)
|
|
self._fill(off)
|
|
if i < cfg.flashes - 1:
|
|
time.sleep(cfg.flash_duration * 0.4)
|
|
finally:
|
|
# Toujours restaurer la luminosite normale meme en cas d'erreur
|
|
if HAS_WS281X and self._strip:
|
|
self._strip.setBrightness(self._cfg.brightness)
|
|
self._fill(off)
|
|
|
|
def _effect_solid(self):
|
|
with self._lock:
|
|
current = self._current_effect
|
|
cfg = self._cfg.get_effect(current)
|
|
self._fill(WS_Color(*cfg.color))
|
|
self._cmd_event.wait()
|
|
|
|
def _effect_finished(self):
|
|
cfg = self._cfg.get_effect("finished")
|
|
self._fill(WS_Color(*cfg.color))
|
|
self._cmd_event.wait(timeout=cfg.duration)
|
|
|
|
def _effect_spin(self):
|
|
cfg = self._cfg.get_effect("printing")
|
|
c = cfg.color
|
|
n = self._cfg.count
|
|
tail = 6
|
|
|
|
pos = 0
|
|
while not self._cmd_event.is_set() and not self._stop_event.is_set():
|
|
for i in range(n):
|
|
dist = (i - pos) % n
|
|
if dist < tail:
|
|
factor = (tail - dist) / tail
|
|
self._strip.setPixelColor(i, WS_Color(
|
|
int(c[0] * factor),
|
|
int(c[1] * factor),
|
|
int(c[2] * factor),
|
|
))
|
|
else:
|
|
self._strip.setPixelColor(i, WS_Color(0, 0, 0))
|
|
self._strip.show()
|
|
pos = (pos + 1) % n
|
|
time.sleep(cfg.speed)
|
|
|
|
def _effect_error(self):
|
|
cfg = self._cfg.get_effect("error")
|
|
red = WS_Color(cfg.color[0], cfg.color[1], cfg.color[2])
|
|
off = WS_Color(0, 0, 0)
|
|
for _ in range(cfg.flashes):
|
|
if self._cmd_event.is_set():
|
|
break
|
|
self._fill(red)
|
|
time.sleep(0.15)
|
|
self._fill(off)
|
|
time.sleep(0.15)
|
|
|
|
def _effect_off(self):
|
|
self._fill(WS_Color(0, 0, 0))
|
|
self._cmd_event.wait()
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _fill(self, color: WS_Color):
|
|
if not HAS_WS281X:
|
|
logger.debug("LED mock fill: %s", color)
|
|
return
|
|
for i in range(self._cfg.count):
|
|
self._strip.setPixelColor(i, color)
|
|
self._strip.show()
|
|
|
|
def _all_off(self):
|
|
try:
|
|
self._fill(WS_Color(0, 0, 0))
|
|
except Exception:
|
|
pass
|
|
|
|
def _notify_change(self, effect: str):
|
|
if self._on_change_cb and self._loop:
|
|
asyncio.run_coroutine_threadsafe(
|
|
self._on_change_cb({"type": "led_effect", "effect": effect}),
|
|
self._loop,
|
|
)
|
|
|
|
@property
|
|
def current_effect(self) -> str:
|
|
with self._lock:
|
|
return self._current_effect
|