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/") 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/") 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/") 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/") 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)