app.py

#!/usr/bin/env python3
"""Configurable Raspberry Pi GPIO/UDP gateway."""

from __future__ import annotations

import copy
import json
import logging
from logging.handlers import RotatingFileHandler
import os
from pathlib import Path
import socket
import threading
import time
from collections import deque
from typing import Any

from flask import Flask, jsonify, redirect, render_template_string, request, url_for

BASE_DIR = Path(__file__).resolve().parent
CONFIG_PATH = Path(os.environ.get("PI_GPIO_GATEWAY_CONFIG", BASE_DIR / "config.json"))
EXAMPLE_PATH = BASE_DIR / "config.example.json"
LOG_PATH = Path(os.environ.get("PI_GPIO_GATEWAY_LOG", BASE_DIR / "pi-gpio-network-gateway.log"))
MOCK_GPIO = os.environ.get("PI_GPIO_GATEWAY_MOCK_GPIO", "").lower() in {"1", "true", "yes"}


class MockButton:
    def __init__(self, pin: int, **_: Any):
        self.pin = pin
        self.when_pressed = None

    @property
    def is_pressed(self) -> bool:
        return False

    def close(self) -> None:
        pass


class MockOutput:
    def __init__(self, pin: int, active_high: bool = True, initial_value: bool = False):
        self.pin = pin
        self.active_high = active_high
        self.value = bool(initial_value)

    def on(self) -> None:
        self.value = True

    def off(self) -> None:
        self.value = False

    def toggle(self) -> None:
        self.value = not self.value

    def close(self) -> None:
        pass


if not MOCK_GPIO:
    try:
        from gpiozero import Button as GpioButton, DigitalOutputDevice as GpioOutput
    except Exception:
        MOCK_GPIO = True

if MOCK_GPIO:
    GpioButton, GpioOutput = MockButton, MockOutput


class Gateway:
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.events: deque[dict[str, str]] = deque(maxlen=250)
        self.inputs: list[Any] = []
        self.outputs: list[Any] = []
        self.input_status: list[dict[str, Any]] = []
        self.output_status: list[dict[str, Any]] = []
        self.stop_event = threading.Event()
        self.udp_thread: threading.Thread | None = None
        self.config = self.load_config()
        self.logger = self.make_logger()
        self.configure_gpio()

    def make_logger(self) -> logging.Logger:
        logger = logging.getLogger("pi-gpio-network-gateway")
        logger.setLevel(logging.INFO)
        if not logger.handlers:
            LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
            handler = RotatingFileHandler(LOG_PATH, maxBytes=5 * 1024 * 1024, backupCount=5)
            handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
            logger.addHandler(handler)
            logger.addHandler(logging.StreamHandler())
        return logger

    def load_config(self) -> dict[str, Any]:
        source = CONFIG_PATH if CONFIG_PATH.exists() else EXAMPLE_PATH
        data = json.loads(source.read_text(encoding="utf-8"))
        self.validate_config(data)
        if not CONFIG_PATH.exists():
            self.save_config(data)
        return data

    def save_config(self, data: dict[str, Any]) -> None:
        CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
        temp = CONFIG_PATH.with_suffix(CONFIG_PATH.suffix + ".tmp")
        temp.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
        temp.replace(CONFIG_PATH)

    @staticmethod
    def validate_config(data: dict[str, Any]) -> None:
        if not isinstance(data.get("gpis"), list) or not isinstance(data.get("gpos"), list):
            raise ValueError("Configuration must contain gpis and gpos arrays")
        used: dict[int, str] = {}
        for kind in ("gpis", "gpos"):
            for index, channel in enumerate(data[kind], 1):
                if not channel.get("enabled"):
                    continue
                pin = int(channel["pin"])
                if pin < 0 or pin > 27:
                    raise ValueError(f"{kind[:-1].upper()} {index}: BCM pin must be 0-27")
                if pin in used:
                    raise ValueError(f"GPIO {pin} is assigned to both {used[pin]} and {kind[:-1].upper()} {index}")
                used[pin] = f"{kind[:-1].upper()} {index}"
        listener_port = int(data["udp_listener"]["port"])
        web_port = int(data["web"]["port"])
        if not 1 <= listener_port <= 65535 or not 1 <= web_port <= 65535:
            raise ValueError("Ports must be between 1 and 65535")

    def event(self, message: str, level: str = "info") -> None:
        item = {"time": time.strftime("%Y-%m-%d %H:%M:%S"), "message": message}
        with self.lock:
            self.events.appendleft(item)
        getattr(self.logger, level, self.logger.info)(message)

    def close_gpio(self) -> None:
        for device in self.inputs + self.outputs:
            try:
                device.close()
            except Exception:
                pass
        self.inputs, self.outputs = [], []

    def configure_gpio(self) -> None:
        with self.lock:
            self.close_gpio()
            self.input_status = []
            self.output_status = []
            for index, cfg in enumerate(self.config["gpis"]):
                status = {"count": 0, "last_trigger": "Never", "last_result": "Not sent"}
                self.input_status.append(status)
                if not cfg.get("enabled"):
                    self.inputs.append(None)
                    continue
                button = GpioButton(
                    int(cfg["pin"]),
                    pull_up=bool(cfg.get("active_low", True)),
                    bounce_time=max(0, int(cfg.get("debounce_ms", 100))) / 1000,
                )
                button.when_pressed = lambda idx=index: self.on_input(idx)
                self.inputs.append(button)
            for cfg in self.config["gpos"]:
                status = {"state": bool(cfg.get("default_on", False)), "last_action": "Configured"}
                self.output_status.append(status)
                if not cfg.get("enabled"):
                    self.outputs.append(None)
                    continue
                output = GpioOutput(
                    int(cfg["pin"]),
                    active_high=bool(cfg.get("active_high", True)),
                    initial_value=bool(cfg.get("default_on", False)),
                )
                self.outputs.append(output)
        self.event(f"GPIO configured ({'mock' if MOCK_GPIO else 'hardware'} mode)")

    def on_input(self, index: int) -> None:
        with self.lock:
            cfg = copy.deepcopy(self.config["gpis"][index])
            status = self.input_status[index]
            now = time.monotonic()
            last = status.get("last_monotonic", 0.0)
            if now - last < int(cfg.get("lockout_ms", 250)) / 1000:
                return
            status["last_monotonic"] = now
            status["count"] += 1
            status["last_trigger"] = time.strftime("%Y-%m-%d %H:%M:%S")
        threading.Thread(target=self.send_udp, args=(index, cfg), daemon=True).start()

    def send_udp(self, index: int, cfg: dict[str, Any]) -> tuple[bool, str]:
        try:
            payload = str(cfg["message"]).encode("utf-8")
            target = (str(cfg["target_host"]), int(cfg["target_port"]))
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.sendto(payload, target)
            result = f"Sent {len(payload)} bytes to {target[0]}:{target[1]}"
            ok = True
            self.event(f"GPI {index + 1}: {result}")
        except Exception as exc:
            result, ok = f"UDP send failed: {exc}", False
            self.event(f"GPI {index + 1}: {result}", "error")
        with self.lock:
            self.input_status[index]["last_result"] = result
        return ok, result

    def output_action(self, index: int, action: str, source: str = "web") -> tuple[bool, str]:
        with self.lock:
            if index < 0 or index >= len(self.config["gpos"]):
                return False, "Unknown GPO"
            cfg = self.config["gpos"][index]
            output = self.outputs[index]
            if not cfg.get("enabled") or output is None:
                return False, f"GPO {index + 1} is disabled"
            action = action.upper()
            if action == "ON":
                output.on()
            elif action == "OFF":
                output.off()
            elif action == "TOGGLE":
                output.toggle()
            elif action == "PULSE":
                output.on()
                duration = max(1, int(cfg.get("pulse_ms", 250))) / 1000
                threading.Timer(duration, self.finish_pulse, args=(index,)).start()
            else:
                return False, "Unknown action"
            self.output_status[index]["state"] = bool(output.value)
            self.output_status[index]["last_action"] = f"{action} from {source}"
        message = f"GPO {index + 1}: {action} from {source}"
        self.event(message)
        return True, message

    def finish_pulse(self, index: int) -> None:
        with self.lock:
            output = self.outputs[index]
            if output is not None:
                output.off()
                self.output_status[index]["state"] = False
                self.output_status[index]["last_action"] = "Pulse complete"
        self.event(f"GPO {index + 1}: pulse complete -> OFF")

    def handle_command(self, message: str, source_ip: str) -> tuple[bool, str]:
        command = message.strip()
        with self.lock:
            configs = copy.deepcopy(self.config["gpos"])
        for index, cfg in enumerate(configs):
            if not cfg.get("enabled"):
                continue
            allowed = str(cfg.get("allowed_source", "")).strip()
            commands = {
                str(cfg.get("pulse_command", "")): "PULSE",
                str(cfg.get("on_command", "")): "ON",
                str(cfg.get("off_command", "")): "OFF",
                str(cfg.get("toggle_command", "")): "TOGGLE",
            }
            if command in commands:
                if allowed and allowed != source_ip:
                    result = f"GPO {index + 1}: ignored command from unauthorized source {source_ip}"
                    self.event(result, "warning")
                    return False, result
                return self.output_action(index, commands[command], f"UDP {source_ip}")
        result = f"Unknown UDP command from {source_ip}: {command!r}"
        self.event(result, "warning")
        return False, result

    def udp_loop(self) -> None:
        while not self.stop_event.is_set():
            cfg = self.config["udp_listener"]
            try:
                with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                    sock.bind((str(cfg.get("host", "0.0.0.0")), int(cfg["port"])))
                    sock.settimeout(1)
                    self.event(f"UDP listener ready on {cfg.get('host', '0.0.0.0')}:{cfg['port']}")
                    while not self.stop_event.is_set():
                        try:
                            payload, address = sock.recvfrom(4096)
                        except socket.timeout:
                            continue
                        message = payload.decode("utf-8", errors="replace").strip()
                        self.event(f"UDP RX {address[0]}:{address[1]} {message!r}")
                        self.handle_command(message, address[0])
            except OSError as exc:
                self.event(f"UDP listener error: {exc}; retrying", "error")
                self.stop_event.wait(2)

    def start(self) -> None:
        self.udp_thread = threading.Thread(target=self.udp_loop, name="udp-listener", daemon=True)
        self.udp_thread.start()


app = Flask(__name__)
gateway = Gateway()

PAGE = r"""
<!doctype html><html><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Pi GPIO Network Gateway</title><style>
body{font:15px system-ui;background:#111827;color:#e5e7eb;margin:0}main{max-width:1200px;margin:auto;padding:24px}h1,h2{color:#fff}.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(320px,1fr));gap:16px}.card{background:#1f2937;padding:16px;border-radius:12px;box-shadow:0 4px 14px #0005}label{display:block;margin:8px 0}input,select{width:100%;box-sizing:border-box;padding:7px;background:#111827;color:#fff;border:1px solid #4b5563;border-radius:6px}input[type=checkbox]{width:auto}button{padding:8px 12px;margin:3px;background:#2563eb;color:#fff;border:0;border-radius:6px;cursor:pointer}.ok{color:#4ade80}.off{color:#9ca3af}.error{color:#f87171}pre{white-space:pre-wrap;background:#0b1020;padding:12px;border-radius:8px;max-height:360px;overflow:auto}.flash{padding:10px;background:#374151;border-radius:6px}</style></head>
<body><main><h1>Raspberry Pi GPIO Network Gateway</h1>{% if error %}<p class="flash error">{{error}}</p>{% endif %}
<form method="post" action="{{url_for('save')}}"><h2>Network services</h2><div class="grid"><div class="card"><label>Web host<input name="web.host" value="{{cfg.web.host}}"></label><label>Web port<input type="number" name="web.port" value="{{cfg.web.port}}"></label></div><div class="card"><label>UDP listener host<input name="udp_listener.host" value="{{cfg.udp_listener.host}}"></label><label>UDP listener port<input type="number" name="udp_listener.port" value="{{cfg.udp_listener.port}}"></label></div></div>
<h2>GPIO inputs → UDP</h2><div class="grid">{% for c in cfg.gpis %}<div class="card"><h3>{{c.name}} <span class="{{'ok' if c.enabled else 'off'}}">{{'Enabled' if c.enabled else 'Disabled'}}</span></h3><label><input type="checkbox" name="gpis.{{loop.index0}}.enabled" {% if c.enabled %}checked{% endif %}> Enabled</label><label>Name<input name="gpis.{{loop.index0}}.name" value="{{c.name}}"></label><label>BCM GPIO<input type="number" name="gpis.{{loop.index0}}.pin" value="{{c.pin}}"></label><label><input type="checkbox" name="gpis.{{loop.index0}}.active_low" {% if c.active_low %}checked{% endif %}> Active LOW / pull-up</label><label>Debounce ms<input type="number" name="gpis.{{loop.index0}}.debounce_ms" value="{{c.debounce_ms}}"></label><label>Lockout ms<input type="number" name="gpis.{{loop.index0}}.lockout_ms" value="{{c.lockout_ms}}"></label><label>Destination<input name="gpis.{{loop.index0}}.target_host" value="{{c.target_host}}"></label><label>Port<input type="number" name="gpis.{{loop.index0}}.target_port" value="{{c.target_port}}"></label><label>Message<input name="gpis.{{loop.index0}}.message" value="{{c.message}}"></label><p>Count {{input_status[loop.index0].count}} · Last {{input_status[loop.index0].last_trigger}}<br>{{input_status[loop.index0].last_result}}</p><button type="submit" formaction="{{url_for('test_input', index=loop.index0)}}" formmethod="post">Test UDP</button></div>{% endfor %}</div>
<h2>UDP → GPIO outputs</h2><div class="grid">{% for c in cfg.gpos %}{% set gpo_index=loop.index0 %}<div class="card"><h3>{{c.name}} <span class="{{'ok' if output_status[gpo_index].state else 'off'}}">{{'ON' if output_status[gpo_index].state else 'OFF'}}</span></h3><label><input type="checkbox" name="gpos.{{gpo_index}}.enabled" {% if c.enabled %}checked{% endif %}> Enabled</label><label>Name<input name="gpos.{{gpo_index}}.name" value="{{c.name}}"></label><label>BCM GPIO<input type="number" name="gpos.{{gpo_index}}.pin" value="{{c.pin}}"></label><label><input type="checkbox" name="gpos.{{gpo_index}}.active_high" {% if c.active_high %}checked{% endif %}> Active HIGH</label><label><input type="checkbox" name="gpos.{{gpo_index}}.default_on" {% if c.default_on %}checked{% endif %}> Default ON</label><label>Pulse ms<input type="number" name="gpos.{{gpo_index}}.pulse_ms" value="{{c.pulse_ms}}"></label><label>Allowed source IP<input name="gpos.{{gpo_index}}.allowed_source" value="{{c.allowed_source}}"></label>{% for field in ['pulse_command','on_command','off_command','toggle_command'] %}<label>{{field|replace('_',' ')|title}}<input name="gpos.{{gpo_index}}.{{field}}" value="{{c[field]}}"></label>{% endfor %}<p>{{output_status[gpo_index].last_action}}</p>{% for a in ['PULSE','ON','OFF','TOGGLE'] %}<button type="submit" formaction="{{url_for('gpo_action', index=gpo_index, action=a)}}" formmethod="post">{{a|title}}</button>{% endfor %}</div>{% endfor %}</div><p><button type="submit">Save and apply configuration</button></p></form>
<h2>Recent events</h2><pre>{% for event in events %}{{event.time}}  {{event.message}}
{% endfor %}</pre></main></body></html>
"""


@app.get("/")
def index():
    with gateway.lock:
        return render_template_string(PAGE, cfg=gateway.config, input_status=gateway.input_status,
                                      output_status=gateway.output_status, events=list(gateway.events), error=request.args.get("error"))


def update_from_form(config: dict[str, Any]) -> dict[str, Any]:
    data = copy.deepcopy(config)
    scalar_types = {"port": int, "pin": int, "debounce_ms": int, "lockout_ms": int,
                    "target_port": int, "pulse_ms": int}
    for section in ("web", "udp_listener"):
        for key in data[section]:
            value = request.form.get(f"{section}.{key}")
            if value is not None:
                data[section][key] = scalar_types.get(key, str)(value)
    for section in ("gpis", "gpos"):
        checkbox_keys = {"enabled", "active_low", "active_high", "default_on"}
        for index, channel in enumerate(data[section]):
            for key in channel:
                field = f"{section}.{index}.{key}"
                if key in checkbox_keys:
                    channel[key] = field in request.form
                elif field in request.form:
                    channel[key] = scalar_types.get(key, str)(request.form[field])
    return data


@app.post("/save")
def save():
    try:
        data = update_from_form(gateway.config)
        gateway.validate_config(data)
        with gateway.lock:
            gateway.save_config(data)
            gateway.config = data
            gateway.configure_gpio()
        gateway.event("Configuration saved and GPIO reapplied")
        return redirect(url_for("index"))
    except Exception as exc:
        return redirect(url_for("index", error=str(exc)))


@app.post("/gpi/<int:index>/test")
def test_input(index: int):
    if 0 <= index < len(gateway.config["gpis"]):
        gateway.send_udp(index, copy.deepcopy(gateway.config["gpis"][index]))
    return redirect(url_for("index"))


@app.post("/gpo/<int:index>/<action>")
def gpo_action(index: int, action: str):
    ok, message = gateway.output_action(index, action, "web")
    return redirect(url_for("index", error="" if ok else message))


@app.get("/api/status")
def api_status():
    with gateway.lock:
        return jsonify(inputs=gateway.input_status, outputs=gateway.output_status, events=list(gateway.events), mock_gpio=MOCK_GPIO)


if __name__ == "__main__":
    gateway.start()
    web = gateway.config["web"]
    app.run(host=str(web.get("host", "0.0.0.0")), port=int(web.get("port", 8080)), threaded=True)