initial commit

app.py

@@ -0,0 +1,117 @@
+import os
+from typing import Any, Dict, Tuple
+
+import requests
+from bottle import Bottle, HTTPError, TEMPLATE_PATH, request, response, run, static_file, template
+
+APP_DIR = os.path.dirname(os.path.abspath(__file__))
+TEMPLATE_PATH.insert(0, os.path.join(APP_DIR, "templates"))
+
+API_BASE_URL = os.getenv("SIGNAL_LIGHT_API_URL", "http://127.0.0.1:8000").rstrip("/")
+REQUEST_TIMEOUT_SECONDS = float(os.getenv("SIGNAL_LIGHT_TIMEOUT", "3"))
+
+app = Bottle()
+
+
+@app.hook("after_request")
+def add_security_headers() -> None:
+    response.set_header("X-Content-Type-Options", "nosniff")
+    response.set_header("X-Frame-Options", "DENY")
+    response.set_header("Referrer-Policy", "no-referrer")
+
+
+@app.get("/")
+def index():
+    return template("index", api_base_url=API_BASE_URL)
+
+@app.get("/favicon.ico")
+def serve_favicon():
+    return static_file("favicon.ico", root=os.path.join(APP_DIR, "static"))
+
+
+@app.get("/static/<filepath:path>")
+def serve_static(filepath: str):
+    return static_file(filepath, root=os.path.join(APP_DIR, "static"))
+
+
+def call_signal_api(method: str, path: str) -> Tuple[Dict[str, Any], int]:
+    """Call the signal-light service and normalize errors for the browser UI."""
+    url = f"{API_BASE_URL}{path}"
+
+    try:
+        upstream = requests.request(
+            method=method,
+            url=url,
+            timeout=REQUEST_TIMEOUT_SECONDS,
+        )
+    except requests.RequestException as exc:
+        return {
+            "ok": False,
+            "error": "Signal-light service is unavailable",
+            "detail": str(exc),
+            "upstream_url": url,
+        }, 502
+
+    try:
+        payload = upstream.json()
+    except ValueError:
+        payload = {
+            "ok": False,
+            "error": "Signal-light service returned a non-JSON response",
+            "detail": upstream.text[:500],
+        }
+
+    if not isinstance(payload, dict):
+        payload = {"data": payload}
+
+    payload.setdefault("ok", upstream.ok)
+    return payload, upstream.status_code
+
+
+def proxy(method: str, path: str):
+    payload, status_code = call_signal_api(method, path)
+    response.content_type = "application/json"
+    response.status = status_code
+    return payload
+
+
+@app.get("/ui-api/status")
+def ui_status():
+    return proxy("GET", "/status")
+
+
+@app.post("/ui-api/color/<color>")
+def ui_color(color: str):
+    if color not in {"green", "yellow", "red"}:
+        raise HTTPError(400, "Unsupported color")
+    return proxy("POST", f"/{color}")
+
+
+@app.post("/ui-api/alert/<state>")
+def ui_alert(state: str):
+    if state == "on":
+        return proxy("POST", "/alert")
+    if state == "off":
+        return proxy("POST", "/alert/off")
+    raise HTTPError(400, "Unsupported alert state")
+
+
+@app.post("/ui-api/demo/<state>")
+def ui_demo(state: str):
+    if state == "on":
+        return proxy("POST", "/demo")
+    if state == "off":
+        return proxy("POST", "/demo/off")
+    raise HTTPError(400, "Unsupported demo state")
+
+
+@app.post("/ui-api/off")
+def ui_off():
+    return proxy("POST", "/off")
+
+
+if __name__ == "__main__":
+    host = os.getenv("BOTTLE_HOST", "0.0.0.0")
+    port = int(os.getenv("BOTTLE_PORT", "8080"))
+    debug = os.getenv("BOTTLE_DEBUG", "false").lower() == "true"
+    run(app=app, host=host, port=port, debug=debug, reloader=debug)