94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
#!/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
|
|
|
|
BATHY_BUILD_ID = "cm95-bathymetry-north-kyushu-v1"
|
|
|
|
|
|
def parse_args():
|
|
p = argparse.ArgumentParser(description="CM95 independent bathymetry PBF test server")
|
|
p.add_argument("--host", default="0.0.0.0")
|
|
p.add_argument("--port", type=int, default=18890)
|
|
p.add_argument("--bathymetry-build-id", default=BATHY_BUILD_ID)
|
|
return p.parse_args()
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
web_root = Path(__file__).resolve().parent
|
|
project_root = web_root.parent
|
|
html_path = web_root / "bathymetry_test.html"
|
|
bathy_root = project_root / "build" / "bathymetry" / args.bathymetry_build_id
|
|
bathy_pbf_root = bathy_root / "pbf"
|
|
bathy_manifest = bathy_root / "build_manifest.json"
|
|
|
|
required = [html_path, bathy_pbf_root, bathy_manifest]
|
|
missing = [str(path) for path in required if not path.exists()]
|
|
if missing:
|
|
raise SystemExit("Missing required path(s):\n" + "\n".join(missing))
|
|
manifest_obj = json.loads(bathy_manifest.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-store")
|
|
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"):
|
|
self.send_bytes(html_path.read_bytes(), "text/html; charset=utf-8")
|
|
return
|
|
if path == "/bathymetry-manifest.json":
|
|
self.send_bytes(
|
|
json.dumps(manifest_obj, ensure_ascii=False, indent=2).encode("utf-8"),
|
|
"application/json; charset=utf-8",
|
|
)
|
|
return
|
|
if path.startswith("/bathymetry/") and path.endswith(".pbf"):
|
|
relative = path[len("/bathymetry/"):]
|
|
parts = relative.split("/")
|
|
if len(parts) != 3:
|
|
self.send_error(404)
|
|
return
|
|
z, x, yfile = parts
|
|
if not (z.isdigit() and x.isdigit() and yfile.endswith(".pbf")):
|
|
self.send_error(404)
|
|
return
|
|
y = yfile[:-4]
|
|
if not y.isdigit():
|
|
self.send_error(404)
|
|
return
|
|
tile_path = bathy_pbf_root / z / x / yfile
|
|
if not tile_path.is_file():
|
|
self.send_bytes(b"", "application/vnd.mapbox-vector-tile", status=204)
|
|
return
|
|
self.send_bytes(tile_path.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"Bathymetry test: http://{args.host}:{args.port}/")
|
|
print(f"Bathymetry PBF: {bathy_pbf_root}")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|