zk_door.py

"""Communication with the ZKTeco access device.

Wraps the pyzk library and exposes safe functions, open_door() and
ping_device(). Device settings are read at call time from device_store
(in-app overrides layered over the .env defaults), so the device can be
reconfigured from the browser without a restart.

The device IP is resolved per call: normally device_store's "ip". If
auto-discovery is enabled and no IP is cached, the app scans the configured
subnets for a responding ZKTeco (off the request path, on a background thread)
and caches the first match. Discovery only succeeds when this machine is on the
same subnet as the device.
"""

import logging
import threading
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout, as_completed
from contextlib import contextmanager

from zk import ZK

from config import settings
from device_store import load_device_config

log = logging.getLogger(__name__)

# Cache the auto-discovered device IP so we do not rescan on every request.
# A scan can take many seconds, so it runs in a background thread and requests
# only ever read this cache. Re-scan is allowed once the retry window passes.
_discovery_cache = {"ts": 0.0, "ip": None}
_discovery_lock = threading.Lock()
_discovery_scanning = False  # guarded by _discovery_lock
_discovery_gen = 0  # bumped on reset; a scan started under an old gen won't cache
_DISCOVERY_RETRY_SECONDS = 60


def _udp_options(force_udp):
    """Which transports to try. If UDP is forced, only UDP; else TCP then UDP."""
    return [True] if force_udp else [False, True]


def _probe_host(ip, port, password, timeout, force_udp):
    """Try to talk to a ZKTeco at ip. Returns (ip, proto, name) or None."""
    for udp in _udp_options(force_udp):
        zk = ZK(ip, port=port, timeout=timeout, password=password, force_udp=udp, ommit_ping=True)
        conn = None
        try:
            conn = zk.connect()
            name = conn.get_device_name()
            return ip, ("UDP" if udp else "TCP"), name
        except Exception:
            continue
        finally:
            if conn is not None:
                try:
                    conn.disconnect()
                except Exception:
                    pass
    return None


def discover_device(subnets=None, timeout=None, max_workers=64, budget=None):
    """Scan subnets for a ZKTeco device. Returns (ip, proto, name) or None.

    subnets is a list of /24 prefixes (first three octets). Each is scanned
    across hosts .1-.254 in parallel; the first responder wins. `budget` caps
    the total wall-clock (seconds) so a no-device scan can't block its caller
    indefinitely; None means no overall cap (background use).
    """
    cfg = load_device_config()
    subnets = subnets if subnets is not None else cfg["discovery_subnets"]
    timeout = timeout if timeout is not None else settings.DEVICE_DISCOVERY_TIMEOUT
    port, password, force_udp = cfg["port"], cfg["password"], cfg["force_udp"]
    candidates = [f"{net}.{host}" for net in subnets for host in range(1, 255)]
    log.info("Scanning %d address(es) across %s for a ZKTeco device", len(candidates), subnets)

    found = None
    executor = ThreadPoolExecutor(max_workers=max_workers)
    try:
        futures = {
            executor.submit(_probe_host, ip, port, password, timeout, force_udp): ip
            for ip in candidates
        }
        for future in as_completed(futures, timeout=budget):
            result = future.result()
            if result:
                found = result
                break
    except FuturesTimeout:
        log.info("Discovery budget of %ss elapsed before a device answered", budget)
    finally:
        # Cancel queued probes; ones already started finish in the background.
        executor.shutdown(wait=False, cancel_futures=True)
    if found:
        log.info("Discovered ZKTeco at %s via %s (%r)", *found)
    return found


def _run_in_background(target):
    """Start `target` on a daemon thread. Indirection point for tests."""
    threading.Thread(target=target, daemon=True).start()


def _background_discover():
    """Scan for the device and cache the result. Runs off the request path."""
    global _discovery_scanning
    with _discovery_lock:
        gen = _discovery_gen
    try:
        found = discover_device()
        if found:
            with _discovery_lock:
                # Discard the result if settings changed (reset) mid-scan, so a
                # stale IP found under old config is never cached.
                if gen == _discovery_gen:
                    _discovery_cache["ip"] = found[0]
        else:
            log.warning("Auto-discovery found no ZKTeco device on the configured subnets")
    finally:
        with _discovery_lock:
            _discovery_scanning = False


def _invalidate_discovery():
    """Forget the cached IP and start the failure backoff window (on failure)."""
    with _discovery_lock:
        _discovery_cache["ip"] = None
        _discovery_cache["ts"] = time.time()


def reset_discovery():
    """Drop any cached discovery result and backoff so a fresh scan can run now.

    Called when device settings change so a stale discovered IP isn't reused.
    Bumps the generation so an in-flight scan started under the old settings
    won't write its (now stale) result into the cache.
    """
    global _discovery_gen
    with _discovery_lock:
        _discovery_cache["ip"] = None
        _discovery_cache["ts"] = 0.0
        _discovery_gen += 1


def _resolve_device_ip():
    """Pick the IP to connect to. Never blocks: if discovery is needed it is
    kicked off in the background and the configured IP is used until a scan
    populates the cache."""
    cfg = load_device_config()
    if not cfg["autodiscover"]:
        return cfg["ip"]

    global _discovery_scanning
    start_scan = False
    with _discovery_lock:
        if _discovery_cache["ip"]:
            return _discovery_cache["ip"]
        now = time.time()
        if not _discovery_scanning and now - _discovery_cache["ts"] >= _DISCOVERY_RETRY_SECONDS:
            _discovery_cache["ts"] = now
            _discovery_scanning = True
            start_scan = True

    if start_scan:
        try:
            _run_in_background(_background_discover)
        except Exception:
            with _discovery_lock:
                _discovery_scanning = False
            log.exception("Failed to start device discovery thread")
    # Non-blocking fallback while a scan runs (or during the backoff window).
    return cfg["ip"]


@contextmanager
def _device_connection():
    """Open a short-lived connection to the device and always close it."""
    cfg = load_device_config()
    zk = ZK(
        _resolve_device_ip(),
        port=cfg["port"],
        timeout=cfg["timeout"],
        password=cfg["password"],
        force_udp=cfg["force_udp"],
        ommit_ping=cfg["ommit_ping"],
    )
    conn = None
    try:
        conn = zk.connect()
        yield conn
    finally:
        if conn is not None:
            try:
                conn.disconnect()
            except Exception:
                log.warning("Error while disconnecting from device", exc_info=True)


def open_door():
    """Trigger the door relay.

    Returns a tuple of (ok, message).
    ok is True on success, False on failure. message is human readable.
    """
    cfg = load_device_config()
    try:
        with _device_connection() as conn:
            conn.unlock(cfg["door_open_seconds"])
        log.info("Door open command sent successfully")
        return True, "Door opened"
    except Exception as exc:  # noqa: BLE001 - we want to surface any failure
        if cfg["autodiscover"]:
            _invalidate_discovery()
        log.error("Failed to open door: %s", exc, exc_info=True)
        return False, f"Could not reach the device: {exc}"


def ping_device():
    """Check whether the device is reachable. Returns (ok, message)."""
    try:
        with _device_connection() as conn:
            name = conn.get_device_name()
        return True, f"Connected to {name or 'device'}"
    except Exception as exc:  # noqa: BLE001
        if load_device_config()["autodiscover"]:
            _invalidate_discovery()
        return False, str(exc)


if __name__ == "__main__":
    # Standalone helper: scan the configured subnets and report the device IP.
    #   python zk_door.py
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    _cfg = load_device_config()
    print(
        f"Scanning {_cfg['discovery_subnets']} on port {_cfg['port']} "
        f"(timeout {settings.DEVICE_DISCOVERY_TIMEOUT}s/host)..."
    )
    hit = discover_device()
    if hit:
        ip, proto, name = hit
        print(f"\nFound ZKTeco at {ip} via {proto}: {name!r}")
        print(f"Set the device IP to {ip} in the app's Device panel (or DEVICE_IP in .env).")
    else:
        print(
            "\nNo ZKTeco found. Make sure this machine is on the SAME subnet as the "
            "device, then check the discovery subnets / port."
        )