CM95/web/kyushu_compare_server.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

RELEASE_ID = "cm95-d1-62a85bd40d3a-1c27f0a66f6b"


def parse_args():
    parser = argparse.ArgumentParser(description="Kyushu S-52 PBF side-by-side comparison server")
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--port", type=int, default=18888)
    parser.add_argument("--release-id", default=RELEASE_ID)
    return parser.parse_args()


def main():
    args = parse_args()
    web_root = Path(__file__).resolve().parent
    project_root = web_root.parent
    release_dir = project_root / "build" / "releases-s52-v2" / args.release_id
    pbf_root = release_dir / "pbf" / "delivery"
    report_root = release_dir / "reports" / "delivery"
    manifest_path = release_dir / "release_manifest_v1.json"
    html_path = web_root / "kyushu_compare.html"
    assets_root = project_root / "build" / "fukuoka_s52_web" / "site" / "assets"
    s52_style_path = project_root / "build" / "s52_style_compiler_release_1c27f0a66f6b" / "style.s52.json"
    s52_sprite_root = project_root / "build" / "s52_resources_v2" / "final"

    required = [
        pbf_root,
        manifest_path,
        html_path,
        assets_root / "maplibre-gl.js",
        assets_root / "maplibre-gl.css",
        s52_style_path,
        s52_sprite_root / "s52-sprite-full.json",
        s52_sprite_root / "s52-sprite-full.png",
        s52_sprite_root / "s52-sprite-full@2x.json",
        s52_sprite_root / "s52-sprite-full@2x.png",
    ]
    missing = [str(path) for path in required if not path.exists()]
    if missing:
        raise SystemExit("Missing required path(s):\n" + "\n".join(missing))

    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    class Handler(BaseHTTPRequestHandler):
        def send_bytes(self, data: bytes, content_type: str, status: int = 200):
            self.send_response(status)
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(len(data)))
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            self.wfile.write(data)

        def do_GET(self):
            path = urlparse(self.path).path

            if path in ("/", "/index.html", "/compare"):
                self.send_bytes(html_path.read_bytes(), "text/html; charset=utf-8")
                return

            if path == "/health":
                payload = {
                    "status": "ok",
                    "release_id": args.release_id,
                    "release_status": manifest.get("status"),
                    "pbf_file_count": manifest.get("pbf_file_count"),
                    "decoded_pbf_features": manifest.get("decoded_pbf_features"),
                    "s52_runtime_contract": (manifest.get("s52_runtime_contract") or {}).get("status"),
                    "s52_light_presentation_contract": (manifest.get("s52_light_presentation_contract") or {}).get("status"),
                    "s52_presentation_conservation_contract": (manifest.get("s52_presentation_conservation_contract") or {}).get("status"),
                }
                self.send_bytes(json.dumps(payload, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8")
                return

            if path == "/release-manifest.json":
                self.send_bytes(manifest_path.read_bytes(), "application/json; charset=utf-8")
                return

            if path == "/s52/full-style.json":
                self.send_bytes(s52_style_path.read_bytes(), "application/json; charset=utf-8")
                return

            if path in (
                "/s52/sprite.json",
                "/s52/sprite.png",
                "/s52/sprite@2x.json",
                "/s52/sprite@2x.png",
            ):
                suffix = ".json" if path.endswith(".json") else ".png"
                scale = "@2x" if "@2x" in path else ""
                sprite_path = s52_sprite_root / f"s52-sprite-full{scale}{suffix}"
                content_type = "application/json; charset=utf-8" if suffix == ".json" else "image/png"
                self.send_bytes(sprite_path.read_bytes(), content_type)
                return

            if path in ("/assets/maplibre-gl.js", "/assets/maplibre-gl.css"):
                filename = path.rsplit("/", 1)[-1]
                file_path = assets_root / filename
                content_type = "application/javascript; charset=utf-8" if filename.endswith(".js") else "text/css; charset=utf-8"
                self.send_bytes(file_path.read_bytes(), content_type)
                return

            if path.startswith("/pbf/") and path.endswith(".pbf"):
                parts = path[len("/pbf/"):].split("/")
                if len(parts) != 3:
                    self.send_error(404)
                    return
                z, x, yfile = parts
                y = yfile[:-4]
                if not (z.isdigit() and x.isdigit() and y.isdigit()):
                    self.send_error(404)
                    return
                tile = pbf_root / z / x / yfile
                if not tile.is_file():
                    report = report_root / z / x / f"{y}.json"
                    if report.is_file():
                        try:
                            report_obj = json.loads(report.read_text(encoding="utf-8"))
                        except Exception:
                            report_obj = {}
                        if (
                            int(report_obj.get("planned_output_feature_count") or 0) == 0
                            and int(report_obj.get("output_feature_count") or 0) == 0
                            and not (report_obj.get("hard_gate_unmapped") or [])
                            and not (report_obj.get("encoder_dropped_features") or [])
                        ):
                            self.send_bytes(b"", "application/vnd.mapbox-vector-tile", status=204)
                            return
                    self.send_error(404, "PBF tile not found")
                    return
                self.send_bytes(tile.read_bytes(), "application/vnd.mapbox-vector-tile")
                return

            self.send_error(404)

        def log_message(self, fmt, *args):
            print(f"{self.address_string()} - {fmt % args}")

    server = ThreadingHTTPServer((args.host, args.port), Handler)
    print(f"Kyushu compare: http://{args.host}:{args.port}/")
    print(f"Release: {release_dir}")
    print(f"PBF: {pbf_root}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()