room_glitch/tools/pi_gpio/pi_gpio_bridge.py

#!/usr/bin/env python3
"""Local Godot -> gpiozero bridge. BCM pins; run on the Pi, not the PC."""
import argparse
import json
import signal
import threading
import time
import hashlib
import sqlite3
from pathlib import Path
from http.server import BaseHTTPRequestHandler, HTTPServer


class Outputs:
    def __init__(self, factory, controller, allowed, clock=time.monotonic, ledger=":memory:"):
        self.factory, self.controller, self.allowed = factory, controller, set(allowed)
        self.clock = clock
        self.last_seen = clock()
        self.pins = {}
        self.lock = threading.RLock()
        self.ledger = sqlite3.connect(ledger, check_same_thread=False)
        self.ledger.execute("CREATE TABLE IF NOT EXISTS requests (id TEXT PRIMARY KEY, digest TEXT NOT NULL, result TEXT)")

    def heartbeat(self):
        with self.lock:
            self.last_seen = self.clock()

    def command(self, data):
        if not isinstance(data, dict): raise ValueError("Expected a JSON object")
        request_id = data.get("request_id")
        if request_id is None: return self._apply_command(data)
        if not isinstance(request_id, str) or not 1 <= len(request_id) <= 100:
            raise ValueError("Invalid request_id")
        digest = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
        with self.lock:
            row = self.ledger.execute("SELECT digest, result FROM requests WHERE id=?", (request_id,)).fetchone()
            if row:
                if row[0] != digest: raise ValueError("request_id reused with a different output")
                return json.loads(row[1]) if row[1] else {"ok": False, "error": "Prior output outcome uncertain; refusing to pulse again"}
            # Commit before touching the pin. A crash can miss a reward, but cannot replay it.
            with self.ledger:
                self.ledger.execute("INSERT INTO requests VALUES (?, ?, NULL)", (request_id, digest))
            try:
                result = self._apply_command(data)
            except Exception as exc:
                result = {"ok": False, "error": str(exc)}
            with self.ledger:
                self.ledger.execute("UPDATE requests SET result=? WHERE id=?", (json.dumps(result), request_id))
            return result

    def _apply_command(self, data):
        if not isinstance(data, dict):
            raise ValueError("Expected a JSON object")
        if data.get("controller") != self.controller:
            raise ValueError("Wrong controller_id for this Pi")
        pin = data.get("pin")
        if type(pin) is not int or pin not in self.allowed:
            raise ValueError("BCM pin is not enabled in the bridge --pins list")
        polarity = data.get("active_high")
        if type(polarity) is not bool:
            raise ValueError("active_high must be boolean")
        mode = data.get("mode")
        if mode not in ("pulse", "set"):
            raise ValueError("mode must be pulse or set")
        duration = data.get("duration_ms", 0)
        if mode == "pulse" and (type(duration) is not int or not 1 <= duration <= 10000):
            raise ValueError("Pulse duration must be 1..10000 ms")
        active = data.get("active", True)
        if type(active) is not bool:
            raise ValueError("active must be boolean")
        with self.lock:
            if pin not in self.pins:
                device = self.factory(pin, active_high=polarity, initial_value=False)
                self.pins[pin] = {"device": device, "polarity": polarity, "deadline": None}
            entry = self.pins[pin]
            if entry["polarity"] != polarity:
                raise ValueError("Pin polarity differs from its earlier configuration; restart bridge to change")
            self.last_seen = self.clock()
            if mode == "pulse":
                entry["device"].on()
                # Latest pulse restarts the duration; an older timer cannot switch it off.
                entry["deadline"] = self.clock() + duration / 1000
            else:
                entry["device"].on() if active else entry["device"].off()
                entry["deadline"] = None
            return {"ok": True, "pin": pin, "active": mode == "pulse" or active}

    def tick(self):
        with self.lock:
            now = self.clock()
            for entry in self.pins.values():
                if now - self.last_seen > 3 or (entry["deadline"] is not None and now >= entry["deadline"]):
                    entry["device"].off()
                    entry["deadline"] = None

    def close(self):
        with self.lock:
            for entry in self.pins.values():
                try:
                    entry["device"].off()
                finally:
                    entry["device"].close()
            self.ledger.close()


def handler_for(outputs, feedback=None):
    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            try:
                if self.headers.get("Origin"):
                    raise ValueError("Browser requests are not supported")
                length = int(self.headers.get("Content-Length", "0"))
                if not 0 <= length <= 4096:
                    raise ValueError("Request is too large")
                data = json.loads(self.rfile.read(length) or b"{}")
                if self.path == "/heartbeat":
                    outputs.heartbeat()
                    result = {"ok": True, "feedback": feedback.snapshot() if feedback else {"devices": [], "events": []}}
                elif self.path == "/output":
                    result = outputs.command(data)
                    print("GPIO:", data, flush=True)
                else:
                    raise ValueError("Unknown endpoint")
                code = 200
            except Exception as exc:
                result, code = {"ok": False, "error": str(exc)}, 400
                print("GPIO ERROR:", exc, flush=True)
            body = json.dumps(result).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def log_message(self, *_args):
            pass
    return Handler


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--controller", default="cabinet_01_gpio")
    parser.add_argument("--pins", default="17", help="BCM pin allowlist, comma separated")
    parser.add_argument("--ledger", default=str(Path.home() / ".local/state/arcade_gpio/requests.sqlite3"))
    parser.add_argument("--devices", help="Explicit PCB feedback JSON inventory")
    parser.add_argument("--rfid-helper", help="Built arcade-rfid helper executable")
    args = parser.parse_args()
    pins = [int(n) for n in args.pins.split(",") if n.strip()]
    if len(set(pins)) != len(pins) or any(pin < 0 or pin > 27 for pin in pins):
        parser.error("Use BCM header GPIO numbers 0..27")
    from gpiozero import OutputDevice, Button
    from device_feedback import Feedback, load_devices
    definitions = load_devices(args.devices, pins) if args.devices else []
    Path(args.ledger).parent.mkdir(parents=True, exist_ok=True)
    outputs = Outputs(OutputDevice, args.controller, pins, ledger=args.ledger)
    feedback = Feedback(definitions, Button, outputs, args.rfid_helper)
    server = HTTPServer(("127.0.0.1", 8765), handler_for(outputs, feedback))
    server.timeout = .05
    stop = threading.Event()
    signal.signal(signal.SIGTERM, lambda *_: stop.set())
    signal.signal(signal.SIGINT, lambda *_: stop.set())

    def watchdog():
        while not stop.wait(.02):
            outputs.tick()
            feedback.tick()

    worker = threading.Thread(target=watchdog, daemon=True)
    worker.start()
    print(f"GPIO bridge ready: controller={args.controller}, BCM pins={pins}, localhost:8765", flush=True)
    try:
        while not stop.is_set():
            server.handle_request()
    finally:
        stop.set()
        worker.join()
        feedback.close()
        outputs.close()
        server.server_close()


if __name__ == "__main__":
    main()