118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
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)
|