248 lines
8.6 KiB
Python
248 lines
8.6 KiB
Python
"""Service de gestion du bouton physique GPIO23 + relay GPIO12.
|
|
|
|
- Détection multi-clic (1 à 4 clics) avec timer
|
|
- Détection long appui (>1500ms)
|
|
- Contrôle du relay (désactive/active le bouton 12V)
|
|
|
|
Mode réel : gpiozero (Raspberry Pi)
|
|
Mode mock : aucune action GPIO, simulation possible via API
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import threading
|
|
import time
|
|
|
|
from backend.services.config_service import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
from gpiozero import Button, OutputDevice
|
|
HAS_GPIO = True
|
|
except (ImportError, Exception):
|
|
HAS_GPIO = False
|
|
logger.warning("gpiozero non disponible — mode mock bouton activé")
|
|
|
|
|
|
class ButtonService:
|
|
def __init__(self, config: Config, led_service, photobooth_service, ws_manager, loop: asyncio.AbstractEventLoop):
|
|
self._cfg = config.button
|
|
self._btn_actions = config.button_actions
|
|
self._led = led_service
|
|
self._pb = photobooth_service
|
|
self._ws = ws_manager
|
|
self._loop = loop
|
|
|
|
self._button = None
|
|
self._relay = None
|
|
|
|
self._click_count = 0
|
|
self._press_time: float = 0.0
|
|
self._click_timer: threading.Timer | None = None
|
|
self._long_timer: threading.Timer | None = None # ← fix: init explicite
|
|
self._long_press_fired = False
|
|
self._relay_enabled = True
|
|
self._lock = threading.Lock()
|
|
|
|
# ── Lifecycle ─────────────────────────────────────────────────────────────
|
|
|
|
def start(self):
|
|
if not HAS_GPIO:
|
|
logger.info("Mode mock bouton — GPIO non disponible")
|
|
return
|
|
|
|
try:
|
|
self._relay = OutputDevice(
|
|
self._cfg.relay_pin,
|
|
active_high=True,
|
|
initial_value=True,
|
|
)
|
|
self._button = Button(
|
|
self._cfg.pin,
|
|
pull_up=True,
|
|
bounce_time=self._cfg.debounce_ms / 1000,
|
|
)
|
|
self._button.when_pressed = self._on_pressed
|
|
self._button.when_released = self._on_released
|
|
logger.info("Bouton GPIO%d, Relay GPIO%d initialisés", self._cfg.pin, self._cfg.relay_pin)
|
|
except Exception as e:
|
|
logger.error("Erreur init GPIO: %s", e)
|
|
|
|
def stop(self):
|
|
if self._click_timer:
|
|
self._click_timer.cancel()
|
|
if self._button:
|
|
try:
|
|
self._button.close()
|
|
except Exception:
|
|
pass
|
|
if self._relay:
|
|
try:
|
|
self._relay.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Relay ─────────────────────────────────────────────────────────────────
|
|
|
|
def relay_on(self):
|
|
"""Active le relay (bouton 12V allumé)."""
|
|
self._relay_enabled = True
|
|
if self._relay:
|
|
self._relay.on()
|
|
self._led.play("idle")
|
|
logger.debug("Relay ON")
|
|
|
|
def relay_off(self):
|
|
"""Désactive le relay (bouton 12V éteint pendant la capture)."""
|
|
self._relay_enabled = False
|
|
if self._relay:
|
|
self._relay.off()
|
|
self._led.play("disabled")
|
|
logger.debug("Relay OFF")
|
|
|
|
# ── Simulation (pour mode mock / tests) ───────────────────────────────────
|
|
|
|
def simulate_click(self, count: int):
|
|
asyncio.run_coroutine_threadsafe(self._handle_click(count), self._loop)
|
|
|
|
def simulate_long_press(self):
|
|
asyncio.run_coroutine_threadsafe(self._handle_long_press(), self._loop)
|
|
|
|
def update_timings(
|
|
self,
|
|
double_click_ms: int | None = None,
|
|
long_press_ms: int | None = None,
|
|
debounce_ms: int | None = None,
|
|
max_clicks: int | None = None,
|
|
):
|
|
"""Met à jour les timings à chaud (sans redémarrage du service)."""
|
|
if double_click_ms is not None:
|
|
self._cfg.double_click_ms = double_click_ms
|
|
if long_press_ms is not None:
|
|
self._cfg.long_press_ms = long_press_ms
|
|
if max_clicks is not None:
|
|
self._cfg.max_clicks = max(1, min(4, max_clicks))
|
|
if debounce_ms is not None:
|
|
self._cfg.debounce_ms = debounce_ms
|
|
if self._button:
|
|
try:
|
|
self._button.bounce_time = debounce_ms / 1000
|
|
except Exception:
|
|
pass
|
|
logger.info(
|
|
"Timings bouton mis à jour : double_click=%dms, long_press=%dms, debounce=%dms, max_clicks=%d",
|
|
self._cfg.double_click_ms, self._cfg.long_press_ms,
|
|
self._cfg.debounce_ms, self._cfg.max_clicks,
|
|
)
|
|
|
|
# ── Callbacks GPIO ────────────────────────────────────────────────────────
|
|
|
|
def _on_pressed(self):
|
|
self._press_time = time.time()
|
|
self._long_press_fired = False
|
|
|
|
# Lance un timer pour détecter le long appui
|
|
if self._click_timer:
|
|
self._click_timer.cancel()
|
|
|
|
long_ms = self._cfg.long_press_ms / 1000
|
|
self._long_timer = threading.Timer(long_ms, self._on_long_press_timer)
|
|
self._long_timer.start()
|
|
|
|
def _on_released(self):
|
|
if hasattr(self, "_long_timer") and self._long_timer:
|
|
self._long_timer.cancel()
|
|
|
|
if self._long_press_fired:
|
|
return # Long press déjà traité
|
|
|
|
# Compte un clic
|
|
with self._lock:
|
|
self._click_count += 1
|
|
count = self._click_count
|
|
|
|
if self._click_timer:
|
|
self._click_timer.cancel()
|
|
|
|
if count >= self._cfg.max_clicks:
|
|
# Dispatch immédiat si max atteint
|
|
self._click_timer = threading.Timer(0.05, self._dispatch_clicks)
|
|
else:
|
|
# Attente pour éventuel prochain clic
|
|
self._click_timer = threading.Timer(
|
|
self._cfg.double_click_ms / 1000,
|
|
self._dispatch_clicks,
|
|
)
|
|
self._click_timer.start()
|
|
|
|
def _on_long_press_timer(self):
|
|
self._long_press_fired = True
|
|
asyncio.run_coroutine_threadsafe(self._handle_long_press(), self._loop)
|
|
|
|
def _dispatch_clicks(self):
|
|
with self._lock:
|
|
count = self._click_count
|
|
self._click_count = 0
|
|
self._click_timer = None
|
|
|
|
asyncio.run_coroutine_threadsafe(self._handle_click(count), self._loop)
|
|
|
|
# ── Handlers async ────────────────────────────────────────────────────────
|
|
|
|
async def _handle_click(self, count: int):
|
|
if count < 1:
|
|
return
|
|
|
|
count = min(count, self._cfg.max_clicks)
|
|
logger.info("Bouton: %d clic(s)", count)
|
|
|
|
# Récupère l'index de l'action photobooth
|
|
action_info = self._btn_actions.get(count) or self._btn_actions.get(str(count))
|
|
if not action_info:
|
|
logger.warning("Aucune action mappée pour %d clic(s)", count)
|
|
return
|
|
|
|
pb_index = action_info.get("photobooth_index", 0)
|
|
label = action_info.get("label", f"Action {count}")
|
|
|
|
await self._ws.broadcast({
|
|
"type": "button_event",
|
|
"clicks": count,
|
|
"action": label,
|
|
"photobooth_index": pb_index,
|
|
})
|
|
|
|
# Désactive le relay + LED disabled
|
|
self.relay_off()
|
|
|
|
# Déclenche l'action photobooth-app
|
|
try:
|
|
await self._pb.trigger_image_action(pb_index)
|
|
except Exception as e:
|
|
logger.error("Erreur déclenchement action %d: %s", pb_index, e)
|
|
self._led.play("error")
|
|
await asyncio.sleep(1)
|
|
self.relay_on()
|
|
|
|
async def _handle_long_press(self):
|
|
logger.info("Bouton: long appui")
|
|
await self._ws.broadcast({"type": "button_event", "clicks": 0, "action": "long_press"})
|
|
|
|
if not self._cfg.print_enabled:
|
|
return
|
|
|
|
# Déclenche la demande d'impression sur la dernière photo
|
|
try:
|
|
self._led.play("printing")
|
|
await self._pb.trigger_share_latest(0)
|
|
except Exception as e:
|
|
logger.error("Erreur long press impression: %s", e)
|
|
self._led.play("error")
|
|
|
|
@property
|
|
def relay_state(self) -> bool:
|
|
return self._relay_enabled
|