Files
CM95/tools/cm95_bathymetry_build_national_v5.py

529 lines
20 KiB
Python

#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import math
import os
import resource
import shutil
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import mapbox_vector_tile
import numpy as np
from shapely.geometry import box, mapping
from shapely import union_all
import cm95_bathymetry_build_v1 as v1
import cm95_bathymetry_build_v3_dense as v3
import cm95_bathymetry_build_v5_adaptive as v5
ROOT = Path(__file__).resolve().parents[1]
BASE_CONFIG = ROOT / "config" / "bathymetry" / "cm95_bathymetry_north_kyushu_v5_adaptive.yaml"
PLAN_PATH = ROOT / "build" / "production" / "cm95_s52_japan_production_plan_v1.json"
PROBE_PATH = ROOT / "build" / "bathymetry" / "national_z8_probe.json"
OUTPUT_ROOT = ROOT / "build" / "bathymetry" / "cm95-bathymetry-japan-v5-adaptive"
AOI = [122.0, 20.0, 154.5, 46.0]
BUILD_ID = "cm95-bathymetry-japan-v5-adaptive"
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Build nationwide Japan v5 bathymetry PBF by z8 core chunks.")
p.add_argument("--only", default="", help="Comma separated z8 x/y pairs, e.g. 222/102,223/101")
p.add_argument("--limit", type=int, default=0)
p.add_argument("--overlap-m", type=float, default=6000.0)
p.add_argument("--resume", action="store_true")
p.add_argument("--workers", type=int, default=1)
p.add_argument("--shard-count", type=int, default=1)
p.add_argument("--shard-index", type=int, default=0)
p.add_argument("--no-finalize", action="store_true")
p.add_argument("--output-root", type=Path, default=OUTPUT_ROOT)
return p.parse_args()
def load_active_tiles() -> list[tuple[int, int, int, int]]:
plan = json.loads(PLAN_PATH.read_text(encoding="utf-8"))
plan_z8 = {(int(j["x"]), int(j["y"])) for j in plan["jobs"] if int(j["z"]) == 8}
probe = json.loads(PROBE_PATH.read_text(encoding="utf-8"))
active = []
for row in probe:
x = int(float(row["x8"]))
y = int(float(row["y8"]))
shallow = int(row.get("shallow60") or 0)
n = int(row.get("n") or 0)
if shallow > 0 and (x, y) in plan_z8:
active.append((x, y, n, shallow))
active.sort(key=lambda item: (item[1], item[0]))
return active
def mercator_aoi_box():
west, south = v1.lonlat_to_mercator(AOI[0], AOI[1])
east, north = v1.lonlat_to_mercator(AOI[2], AOI[3])
return box(west, south, east, north)
def core_box_for_z8(x: int, y: int):
tile = v1.xyz_tile_bounds(8, x, y)
return box(tile.west_m, tile.south_m, tile.east_m, tile.north_m).intersection(mercator_aoi_box())
def bbox_wgs84_for_mercator(geom) -> list[float]:
minx, miny, maxx, maxy = geom.bounds
west, south = v1.mercator_to_lonlat(minx, miny)
east, north = v1.mercator_to_lonlat(maxx, maxy)
return [float(west), float(south), float(east), float(north)]
def descendant_tiles(x8: int, y8: int, z: int):
factor = 1 << (z - 8)
for x in range(x8 * factor, (x8 + 1) * factor):
for y in range(y8 * factor, (y8 + 1) * factor):
yield x, y
def build_core_tiles(
contours: dict[float, Any],
cfg: dict[str, Any],
output_root: Path,
x8: int,
y8: int,
) -> dict[str, Any]:
pbf_root = output_root / "pbf"
pbf_root.mkdir(parents=True, exist_ok=True)
source_layer = str(cfg["source_layer"])
extent = int(cfg["extent"])
buffer_px = float(cfg.get("tile_buffer_px", 8))
build_id = BUILD_ID
major_contours = {float(v) for v in cfg.get("major_contours_m", cfg["contour_levels_m"])}
simplify_by_zoom = {int(k): float(v) for k, v in cfg.get("simplify_m_by_zoom", {}).items()}
tile_count = feature_count = total_bytes = encoder_dropped = 0
zoom_stats: dict[str, Any] = {}
for z in range(int(cfg["minzoom"]), int(cfg["maxzoom"]) + 1):
z_tiles = z_features = z_bytes = z_drop = 0
simplify_m = simplify_by_zoom.get(z, 0.0)
for x, y in descendant_tiles(x8, y8, z):
tile = v1.xyz_tile_bounds(z, x, y)
pad = tile.width_m * buffer_px / 256.0
clip_box = box(tile.west_m-pad, tile.south_m-pad, tile.east_m+pad, tile.north_m+pad)
features = []
local_index = 0
for depth, contour_geom in sorted(contours.items()):
if contour_geom.is_empty or not contour_geom.intersects(clip_box):
continue
clipped = contour_geom.intersection(clip_box)
for part in v1.line_parts(clipped):
if simplify_m > 0:
part = part.simplify(simplify_m, preserve_topology=False)
if part.is_empty or part.length <= 0:
continue
if not v1.line_survives_mvt_quantization(part, tile.mercator_tuple, extent):
continue
local_index += 1
features.append({
"id": v1.feature_id(build_id, z, x, y, depth, local_index, part),
"geometry": mapping(part),
"properties": {
"depth_m": float(depth),
"contour_class": "major" if float(depth) in major_contours else "minor",
"source_kind": str(cfg["source_kind"]),
"reliability": str(cfg["reliability"]),
"build_id": build_id,
},
})
if not features:
continue
data = mapbox_vector_tile.encode(
[{"name": source_layer, "features": features}],
default_options={"quantize_bounds": tile.mercator_tuple, "extents": extent},
)
actual = len(mapbox_vector_tile.decode(data).get(source_layer, {}).get("features", []))
dropped = max(0, len(features)-actual)
encoder_dropped += dropped
z_drop += dropped
tile_path = pbf_root / str(z) / str(x) / f"{y}.pbf"
tile_path.parent.mkdir(parents=True, exist_ok=True)
if tile_path.exists():
raise RuntimeError(f"tile ownership collision: {tile_path}")
tile_path.write_bytes(data)
tile_count += 1
feature_count += actual
total_bytes += len(data)
z_tiles += 1
z_features += actual
z_bytes += len(data)
zoom_stats[str(z)] = {"tiles": z_tiles, "features": z_features, "encoder_dropped_features": z_drop, "bytes": z_bytes}
return {
"pbf_tiles": tile_count,
"tile_feature_instances": feature_count,
"encoder_dropped_features": encoder_dropped,
"pbf_bytes": total_bytes,
"zooms": zoom_stats,
}
def extract_contours_single_pass_fast(
xs: np.ndarray,
ys: np.ndarray,
grid: np.ndarray,
water_mask,
cfg: dict[str, Any],
) -> tuple[dict[float, Any], dict[str, Any]]:
levels = sorted(float(value) for value in cfg["contour_levels_m"])
segments: dict[float, list[Any]] = {level: [] for level in levels}
fallback_counts: dict[float, int] = {level: 0 for level in levels}
v00 = grid[:-1, :-1]
v10 = grid[:-1, 1:]
v11 = grid[1:, 1:]
v01 = grid[1:, :-1]
finite_count = (
np.isfinite(v00).astype(np.uint8)
+ np.isfinite(v10).astype(np.uint8)
+ np.isfinite(v11).astype(np.uint8)
+ np.isfinite(v01).astype(np.uint8)
)
lo = np.fmin(np.fmin(v00, v10), np.fmin(v11, v01))
hi = np.fmax(np.fmax(v00, v10), np.fmax(v11, v01))
candidate_mask = (
(finite_count >= 3)
& np.isfinite(lo)
& np.isfinite(hi)
& (lo != hi)
& (hi >= levels[0])
& (lo <= levels[-1])
)
candidate_cells = np.argwhere(candidate_mask)
import bisect
import math
for iy, ix in candidate_cells:
values = (
float(grid[iy, ix]),
float(grid[iy, ix + 1]),
float(grid[iy + 1, ix + 1]),
float(grid[iy + 1, ix]),
)
finite_values = [value for value in values if math.isfinite(value)]
cell_lo = min(finite_values)
cell_hi = max(finite_values)
start = bisect.bisect_left(levels, cell_lo)
stop = bisect.bisect_right(levels, cell_hi)
finite_n = len(finite_values)
y0 = float(ys[iy])
y1 = float(ys[iy + 1])
x0 = float(xs[ix])
x1 = float(xs[ix + 1])
for level in levels[start:stop]:
cell_segments = v3._cell_contours(
x0,
x1,
y0,
y1,
values,
level,
)
segments[level].extend(cell_segments)
if finite_n == 3:
fallback_counts[level] += len(cell_segments)
contours: dict[float, Any] = {}
stats: dict[str, Any] = {}
min_segment = float(cfg["min_segment_length_m"])
simplify_m = float(cfg.get("regional_simplify_m", 0))
smooth_iterations = int(cfg.get("chaikin_iterations", 0))
import cm95_bathymetry_build_v2_smooth as v2
for level in levels:
raw = segments[level]
if not raw:
stats[str(level)] = {
"raw_segments": 0,
"triangle_fallback_segments": 0,
"line_parts": 0,
"length_m": 0.0,
}
continue
from shapely import line_merge
merged = line_merge(union_all(raw))
parts: list[Any] = []
for part in v1.line_parts(merged):
if part.length < min_segment:
continue
smoothed = v2.chaikin_smooth(part, smooth_iterations)
if simplify_m > 0:
smoothed = smoothed.simplify(
simplify_m,
preserve_topology=False,
)
clipped = smoothed.intersection(water_mask)
parts.extend(
candidate
for candidate in v1.line_parts(clipped)
if candidate.length >= min_segment
)
if not parts:
stats[str(level)] = {
"raw_segments": len(raw),
"triangle_fallback_segments": fallback_counts[level],
"line_parts": 0,
"length_m": 0.0,
}
continue
geom = union_all(parts)
contours[level] = geom
final_parts = v1.line_parts(geom)
stats[str(level)] = {
"raw_segments": len(raw),
"triangle_fallback_segments": fallback_counts[level],
"line_parts": len(final_parts),
"length_m": round(
sum(part.length for part in final_parts),
3,
),
}
stats["_performance"] = {
"candidate_cells": int(len(candidate_cells)),
"all_cells": int((len(ys) - 1) * (len(xs) - 1)),
}
return contours, stats
def clip_contours_to_core(contours: dict[float, Any], core_box) -> dict[float, Any]:
out: dict[float, Any] = {}
for depth, geom in contours.items():
clipped = geom.intersection(core_box)
parts = [part for part in v1.line_parts(clipped) if not part.is_empty and part.length > 0]
if parts:
out[depth] = union_all(parts)
return out
def build_chunk(
x8: int,
y8: int,
point_hint: int,
shallow_hint: int,
overlap_m: float,
output_root: Path,
resume: bool,
) -> dict[str, Any]:
chunk_dir = output_root / "chunks"
chunk_dir.mkdir(parents=True, exist_ok=True)
report_path = chunk_dir / f"z8_{x8}_{y8}.json"
if resume and report_path.exists():
old = json.loads(report_path.read_text(encoding="utf-8"))
if old.get("status") == "PASS":
return old
started = time.monotonic()
started_at = datetime.now(timezone.utc)
cfg = v1.load_config(BASE_CONFIG)
cfg["build_id"] = BUILD_ID
cfg["source_kind"] = "derived_sounding_adaptive_tin_v5"
core_box = core_box_for_z8(x8, y8)
if core_box.is_empty:
return {"status": "SKIP", "x8": x8, "y8": y8, "reason": "outside_aoi"}
expanded = core_box.buffer(float(overlap_m), cap_style=3, join_style=2)
cfg["bbox_wgs84"] = bbox_wgs84_for_mercator(expanded)
conn = v1.connect_db()
try:
points, sounding_stats = v1.fetch_soundings(conn, cfg)
if len(points) < 3:
result = {"status": "SKIP", "x8": x8, "y8": y8, "reason": "fewer_than_3_soundings", "duration_seconds": round(time.monotonic()-started,3)}
report_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
return result
water_mask, water_rows = v1.fetch_water_mask(conn, cfg)
finally:
conn.close()
xs, ys, raw_grid, water_grid, local_spacing_grid, surface_stats = v5.build_adaptive_tin_surface(points, water_mask, cfg)
repaired_grid, gap_stats = v3.repair_small_water_gaps(raw_grid, water_grid, cfg)
newly_filled = (~__import__("numpy").isfinite(raw_grid)) & __import__("numpy").isfinite(repaired_grid)
local_spacing_grid[newly_filled] = float(cfg["smooth_spacing_high_m"])
smooth_grid, smooth_stats = v5.adaptive_smooth_surface(repaired_grid, water_grid, local_spacing_grid, cfg)
contours, contour_stats = extract_contours_single_pass_fast(xs, ys, smooth_grid, water_mask, cfg)
core_contours = clip_contours_to_core(contours, core_box)
tile_stats = build_core_tiles(core_contours, cfg, output_root, x8, y8) if core_contours else {
"pbf_tiles": 0, "tile_feature_instances": 0, "encoder_dropped_features": 0, "pbf_bytes": 0, "zooms": {}
}
duration = round(time.monotonic()-started, 3)
peak_kib = int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)
result = {
"status": "PASS",
"x8": x8,
"y8": y8,
"point_hint": point_hint,
"shallow60_hint": shallow_hint,
"input_bbox_wgs84": cfg["bbox_wgs84"],
"core_bbox_wgs84": bbox_wgs84_for_mercator(core_box),
"overlap_m": overlap_m,
"started_at": started_at.isoformat(),
"duration_seconds": duration,
"peak_rss_kib_process": peak_kib,
"sounding_stats": sounding_stats,
"water_area_rows": water_rows,
"surface_stats": surface_stats,
"gap_repair_stats": gap_stats,
"surface_smooth_stats": smooth_stats,
"contour_levels_generated": sorted(float(k) for k in core_contours.keys()),
"contour_stats": contour_stats,
"tile_stats": tile_stats,
}
report_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
return result
def aggregate(output_root: Path, active_count: int, started_at: datetime, elapsed: float) -> dict[str, Any]:
reports = []
for path in sorted((output_root / "chunks").glob("z8_*.json")):
reports.append(json.loads(path.read_text(encoding="utf-8")))
passed = [r for r in reports if r.get("status") == "PASS"]
skipped = [r for r in reports if r.get("status") == "SKIP"]
tile_stats = {
"pbf_tiles": sum(int(r.get("tile_stats", {}).get("pbf_tiles", 0)) for r in passed),
"tile_feature_instances": sum(int(r.get("tile_stats", {}).get("tile_feature_instances", 0)) for r in passed),
"encoder_dropped_features": sum(int(r.get("tile_stats", {}).get("encoder_dropped_features", 0)) for r in passed),
"pbf_bytes": sum(int(r.get("tile_stats", {}).get("pbf_bytes", 0)) for r in passed),
}
actual_files = list((output_root / "pbf").rglob("*.pbf")) if (output_root / "pbf").exists() else []
actual_bytes = sum(p.stat().st_size for p in actual_files)
return {
"schema": "navsea.cm95.bathymetry-national-result/v1",
"build_id": BUILD_ID,
"aoi_wgs84": AOI,
"algorithm_baseline": "v5-adaptive",
"active_z8_expected": active_count,
"chunk_reports": len(reports),
"chunks_pass": len(passed),
"chunks_skip": len(skipped),
"started_at": started_at.isoformat(),
"finished_at": datetime.now(timezone.utc).isoformat(),
"elapsed_seconds": round(elapsed, 3),
"sum_chunk_seconds": round(sum(float(r.get("duration_seconds", 0)) for r in reports), 3),
"max_chunk_seconds": round(max([float(r.get("duration_seconds", 0)) for r in reports] or [0]), 3),
"tile_stats_sum": tile_stats,
"actual_pbf_files": len(actual_files),
"actual_pbf_bytes": actual_bytes,
"disk_usage_bytes": sum(p.stat().st_size for p in output_root.rglob("*") if p.is_file()),
}
def main() -> int:
args = parse_args()
output_root = args.output_root.resolve()
output_root.mkdir(parents=True, exist_ok=True)
active = load_active_tiles()
only = set()
if args.only.strip():
for token in args.only.split(","):
x, y = token.strip().split("/")
only.add((int(x), int(y)))
active = [item for item in active if (item[0], item[1]) in only]
if args.limit > 0:
active = active[: args.limit]
if args.shard_count < 1:
raise SystemExit("--shard-count must be >= 1")
if args.shard_index < 0 or args.shard_index >= args.shard_count:
raise SystemExit("--shard-index must satisfy 0 <= index < count")
active = [
item
for index, item in enumerate(active)
if index % args.shard_count == args.shard_index
]
started_at = datetime.now(timezone.utc)
started = time.monotonic()
print(f"NATIONAL_BUILD {BUILD_ID}", flush=True)
print(f"ACTIVE_Z8 {len(active)}", flush=True)
print(f"OUTPUT {output_root}", flush=True)
workers = max(1, int(args.workers))
print(f"WORKERS {workers}", flush=True)
if workers == 1:
for index, (x8, y8, n, shallow) in enumerate(active, start=1):
t0 = time.monotonic()
print(f"CHUNK_START {index}/{len(active)} z8/{x8}/{y8} points={n} shallow60={shallow}", flush=True)
result = build_chunk(x8, y8, n, shallow, args.overlap_m, output_root, args.resume)
print(
f"CHUNK_DONE {index}/{len(active)} z8/{x8}/{y8} status={result.get('status')} "
f"seconds={result.get('duration_seconds', round(time.monotonic()-t0,3))} "
f"tiles={result.get('tile_stats',{}).get('pbf_tiles',0)} "
f"bytes={result.get('tile_stats',{}).get('pbf_bytes',0)}",
flush=True,
)
else:
from concurrent.futures import ProcessPoolExecutor, as_completed
completed = 0
with ProcessPoolExecutor(max_workers=workers) as executor:
futures = {}
for x8, y8, n, shallow in active:
print(f"CHUNK_QUEUE z8/{x8}/{y8} points={n} shallow60={shallow}", flush=True)
future = executor.submit(
build_chunk,
x8,
y8,
n,
shallow,
args.overlap_m,
output_root,
args.resume,
)
futures[future] = (x8, y8, n, shallow)
for future in as_completed(futures):
x8, y8, n, shallow = futures[future]
completed += 1
try:
result = future.result()
except Exception as exc:
print(
f"CHUNK_FAIL {completed}/{len(active)} z8/{x8}/{y8} "
f"error={type(exc).__name__}:{exc}",
flush=True,
)
raise
print(
f"CHUNK_DONE {completed}/{len(active)} z8/{x8}/{y8} "
f"status={result.get('status')} "
f"seconds={result.get('duration_seconds',0)} "
f"tiles={result.get('tile_stats',{}).get('pbf_tiles',0)} "
f"bytes={result.get('tile_stats',{}).get('pbf_bytes',0)}",
flush=True,
)
if not args.no_finalize:
manifest = aggregate(
output_root,
len(active),
started_at,
time.monotonic()-started,
)
(output_root / "build_manifest.json").write_text(
json.dumps(manifest, ensure_ascii=False, indent=2),
encoding="utf-8",
)
print(
"NATIONAL_DONE",
json.dumps(manifest, ensure_ascii=False),
flush=True,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())