photobooth/backend/relais.py

"""Pilotage module relais USB HID (LCUS 5131:2007).

4 canaux, protocole LCUS :
  ON  : 0x00 0xA0 <ch> 0x01 <checksum> 0x00 0x00 0x00 0x00
  OFF : 0x00 0xA0 <ch> 0x00 <checksum> 0x00 0x00 0x00 0x00
  Status : 0x00 0xD1 ...

Assignation :
  1 = Projecteur gauche (NO)
  2 = Projecteur droit  (NO)
  3 = Canon alim         (NF — OFF=alimenté, ON=coupé)
  4 = Libre
"""

import ctypes
import ctypes.util
import logging
import threading
import time

log = logging.getLogger("photobooth.relais")

VID = 0x5131
PID = 0x2007

CANAUX = {
    "projecteur_gauche": {"relay": 1, "mode": "NO"},
    "projecteur_droit":  {"relay": 2, "mode": "NO"},
    "canon":             {"relay": 3, "mode": "NF"},
}

_lib = None
_dev = None
_lock = threading.Lock()


def _charger_hidapi():
    global _lib
    if _lib is not None:
        return _lib
    for name in ("hidapi-hidraw", "hidapi-libusb", "hidapi"):
        path = ctypes.util.find_library(name)
        if path:
            _lib = ctypes.CDLL(path)
            _lib.hid_init()
            _lib.hid_open.restype = ctypes.c_void_p
            _lib.hid_read_timeout.restype = ctypes.c_int
            log.info(f"hidapi chargé : {path}")
            return _lib
    try:
        _lib = ctypes.CDLL("libhidapi-hidraw.so.0")
        _lib.hid_init()
        _lib.hid_open.restype = ctypes.c_void_p
        _lib.hid_read_timeout.restype = ctypes.c_int
        return _lib
    except OSError:
        log.warning("hidapi introuvable — relais désactivés")
        return None


def connecter():
    global _dev
    lib = _charger_hidapi()
    if not lib:
        return False
    with _lock:
        if _dev:
            return True
        _dev = lib.hid_open(VID, PID, None)
        if not _dev:
            log.warning("Module relais non trouvé (5131:2007)")
            _dev = None
            return False
        log.info("Module relais connecté")
        return True


def deconnecter():
    global _dev
    with _lock:
        if _dev and _lib:
            _lib.hid_close(_dev)
        _dev = None


def est_connecte():
    return _dev is not None


def _ecrire(data: list[int]):
    if not _dev or not _lib:
        return False
    buf = (ctypes.c_ubyte * 9)(*data)
    with _lock:
        res = _lib.hid_write(_dev, buf, 9)
    return res > 0


def _relay_set(num: int, on: bool):
    state = 0x01 if on else 0x00
    checksum = (0xA0 + num + state) & 0xFF
    ok = _ecrire([0x00, 0xA0, num, state, checksum, 0x00, 0x00, 0x00, 0x00])
    if ok:
        log.info(f"Relais {num} → {'ON' if on else 'OFF'}")
    else:
        log.error(f"Échec écriture relais {num}")
    return ok


def status():
    if not _dev or not _lib:
        return None
    _ecrire([0x00, 0xD1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
    time.sleep(0.1)
    rbuf = (ctypes.c_ubyte * 64)()
    with _lock:
        n = _lib.hid_read_timeout(_dev, rbuf, 64, 500)
    if n < 4:
        return None
    return {
        "relay_1": rbuf[1] == 1,
        "relay_2": rbuf[2] == 1,
        "relay_3": rbuf[3] == 1,
    }


def activer(nom: str):
    cfg = CANAUX.get(nom)
    if not cfg:
        return False
    return _relay_set(cfg["relay"], True)


def desactiver(nom: str):
    cfg = CANAUX.get(nom)
    if not cfg:
        return False
    return _relay_set(cfg["relay"], False)


def projecteurs(on: bool):
    _relay_set(1, on)
    _relay_set(2, on)
    log.info(f"Projecteurs {'allumés' if on else 'éteints'}")


def power_cycle_canon(duree: float = 3.0):
    log.info(f"Power-cycle Canon ({duree}s)")
    _relay_set(3, True)   # NF : ON = coupé
    time.sleep(duree)
    _relay_set(3, False)  # NF : OFF = alimenté
    log.info("Canon ré-alimenté")


def etat_complet():
    raw = status()
    if raw is None:
        return {"connecte": False}
    result = {"connecte": True}
    for nom, cfg in CANAUX.items():
        relay_on = raw.get(f"relay_{cfg['relay']}", False)
        if cfg["mode"] == "NF":
            result[nom] = not relay_on
        else:
            result[nom] = relay_on
    return result