astrape-dashboard/astrape_dashboard/scripts/daemons/web_daemon.py

from __future__ import annotations

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from importlib import import_module, reload
from os import environ
from pathlib import Path
import json
from urllib.parse import parse_qs, urlparse

from astrape_ems.classes.env import EnvLoader

EnvLoader().load()

HOST = environ.get("ASTRAPE_WEB_HOST", "0.0.0.0")
PORT = int(environ.get("ASTRAPE_WEB_PORT", "8080"))
PROJECT_ROOT = Path(__file__).resolve().parents[2]
WATCHED_PATHS = [
    PROJECT_ROOT / "gibil" / "classes" / "models.py",
    PROJECT_ROOT / "gibil" / "classes" / "webui.py",
    PROJECT_ROOT / "gibil" / "classes" / "weather" / "display.py",
    PROJECT_ROOT / "gibil" / "classes" / "weather" / "station_display.py",
    PROJECT_ROOT / "gibil" / "classes" / "oracle" / "display.py",
    PROJECT_ROOT / "gibil" / "classes" / "oracle" / "quality_display.py",
    PROJECT_ROOT / "gibil" / "classes" / "weather" / "store.py",
    PROJECT_ROOT / "gibil" / "classes" / "oracle" / "store.py",
    PROJECT_ROOT / "gibil" / "classes" / "oracle" / "builder.py",
    PROJECT_ROOT / "gibil" / "classes" / "oracle" / "config.py",
    PROJECT_ROOT / "gibil" / "classes" / "sigen" / "store.py",
]


class AstrapeWebHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        path = parsed.path

        if path in {"/", "/oracle"}:
            self._send_html(self._webui().render_page("oracle"))
            return

        if path == "/weather":
            self._send_html(self._webui().render_page("weather"))
            return

        if path == "/weather-station":
            self._send_html(self._webui().render_page("weather_station"))
            return

        if path == "/quality":
            self._send_html(self._webui().render_page("quality"))
            return

        if path == "/api/weather":
            self._send_json_text(self._webui().weather_payload())
            return

        if path == "/api/weather-station":
            params = parse_qs(parsed.query)
            lookback_hours = self._float_param(params, "lookback_hours", 168)
            self._send_json_text(
                self._webui().weather_station_payload(
                    lookback_hours=lookback_hours
                )
            )
            return

        if path == "/api/oracle":
            self._send_json_text(self._webui().oracle_payload())
            return

        if path == "/api/oracle-quality":
            params = parse_qs(parsed.query)
            lookback_hours = self._float_param(params, "lookback_hours", 168)
            self._send_json_text(
                self._webui().oracle_quality_payload(
                    lookback_hours=lookback_hours
                )
            )
            return

        if path == "/api/ui-version":
            self._send_json_text(json.dumps({"version": self._ui_version()}))
            return

        self.send_error(404)

    def log_message(self, format: str, *args: object) -> None:
        print(f"{self.address_string()} - {format % args}")

    def _webui(self):
        models_module = import_module("gibil.classes.models")
        weather_store_module = import_module("astrape_oracle.classes.weather.store")
        sigen_store_module = import_module("astrape_ems.classes.store")
        oracle_store_module = import_module("astrape_oracle.classes.oracle_store")
        oracle_config_module = import_module("gibil.classes.forecast_config")
        oracle_builder_module = import_module("gibil.classes.oracle_builder")
        oracle_display_module = import_module("astrape_dashboard.classes.displays.oracle")
        oracle_quality_display_module = import_module(
            "astrape_dashboard.classes.displays.oracle_quality"
        )
        weather_display_module = import_module("astrape_oracle.classes.datasets")
        weather_station_display_module = import_module(
            "astrape_oracle.classes.datasets"
        )
        webui_module = import_module("astrape_dashboard.classes.webui")
        reload(models_module)
        reload(weather_display_module)
        reload(weather_station_display_module)
        reload(weather_store_module)
        reload(sigen_store_module)
        reload(oracle_store_module)
        reload(oracle_config_module)
        reload(oracle_builder_module)
        reload(oracle_display_module)
        reload(oracle_quality_display_module)
        reload(webui_module)
        return webui_module.WebUI()

    def _float_param(
        self,
        params: dict[str, list[str]],
        key: str,
        default: float,
    ) -> float:
        values = params.get(key)
        if not values:
            return default
        try:
            return float(values[0])
        except ValueError:
            return default

    def _ui_version(self) -> str:
        mtimes = [
            str(path.stat().st_mtime_ns)
            for path in WATCHED_PATHS
            if path.exists()
        ]
        return ".".join(mtimes)

    def _send_html(self, body: str) -> None:
        encoded = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def _send_json_text(self, body: str) -> None:
        encoded = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)


def main() -> None:
    server = ThreadingHTTPServer((HOST, PORT), AstrapeWebHandler)
    print(f"Astrape web UI listening on http://{HOST}:{PORT}")
    server.serve_forever()


if __name__ == "__main__":
    main()