From c03a5ac788ef64df1dd916a0409fdddc4def10c0 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Fri, 18 Sep 2026 23:56:51 +0800 Subject: [PATCH] feat: add complete AOI production footprint --- src/cm95pbf/cli.py | 10 ++- src/cm95pbf/production.py | 126 +++++++++++++++++++++++++++++++++----- 2 files changed, 120 insertions(+), 16 deletions(-) diff --git a/src/cm95pbf/cli.py b/src/cm95pbf/cli.py index 80bcfa7..b9bc7fd 100644 --- a/src/cm95pbf/cli.py +++ b/src/cm95pbf/cli.py @@ -139,7 +139,13 @@ def build_parser() -> argparse.ArgumentParser: prod_plan.add_argument("--s52-boundary-table", choices=["Plain", "Symbolized"], default="Plain") prod_plan.add_argument("--s52-mariner-mode", choices=["runtime", "reference"], default="runtime") prod_plan.add_argument("--aoi-name", default=None, help="可审计 AOI 名称") - prod_plan.add_argument("--aoi-bbox", type=_parse_aoi_bbox, default=None, metavar="W,S,E,N", help="只枚举与 WGS84 bbox 相交的 coverage/tile") + prod_plan.add_argument("--aoi-bbox", type=_parse_aoi_bbox, default=None, metavar="W,S,E,N", help="WGS84 AOI bbox；aoi-complete 模式下枚举 bbox 内完整 XYZ 金字塔") + prod_plan.add_argument( + "--tile-enumeration-mode", + choices=["auto", "coverage", "aoi-complete"], + default="auto", + help="auto=有 AOI 时使用 aoi-complete，否则 coverage；正式区域/全国生产应使用 aoi-complete", + ) prod_build = sub.add_parser( "build-production", @@ -231,6 +237,7 @@ def main() -> None: s52_mariner_mode=args.s52_mariner_mode, aoi_name=args.aoi_name, aoi_bbox_wgs84=args.aoi_bbox, + tile_enumeration_mode=args.tile_enumeration_mode, ) print(json.dumps({ "plan_id": result["plan_id"], @@ -239,6 +246,7 @@ def main() -> None: "production_pipeline_version": result.get("production_pipeline_version"), "s52": result.get("s52"), "aoi": result.get("aoi"), + "tile_enumeration_mode": result.get("tile_enumeration_mode"), "source_stats": result["source_stats"], "output": str(Path(args.output)), }, ensure_ascii=False, indent=2)) diff --git a/src/cm95pbf/production.py b/src/cm95pbf/production.py index 4c18578..f3f617a 100644 --- a/src/cm95pbf/production.py +++ b/src/cm95pbf/production.py @@ -35,8 +35,9 @@ from .policy import ( PRODUCTION_PLAN_SCHEMA_V1 = "navsea.cm95.production-plan/v1" PRODUCTION_PLAN_SCHEMA = "navsea.cm95.production-plan/v2" SUPPORTED_PRODUCTION_PLAN_SCHEMAS = {PRODUCTION_PLAN_SCHEMA_V1, PRODUCTION_PLAN_SCHEMA} -PRODUCTION_PIPELINE_VERSION = "cm95pbf-production/2.0-s52-aoi" +PRODUCTION_PIPELINE_VERSION = "cm95pbf-production/2.1-s52-aoi-complete" PRODUCTION_RUN_SCHEMA = "navsea.cm95.production-run/v1" +TILE_ENUMERATION_MODES = {"auto", "coverage", "aoi-complete"} def utc_now() -> str: @@ -98,18 +99,58 @@ def _scale_ladder(policy: dict[str, Any], z: int) -> list[int]: def _enumeration_scales(policy: dict[str, Any], z: int) -> list[int]: - """Return scales allowed to materialize physical tiles at a zoom. - - This is intentionally distinct from scale_ladders: coarse fallback scales may - be read *inside* an already materialized tile, but they must not expand the - high-zoom production footprint. - """ + """Return scales allowed to materialize physical tiles in coverage mode.""" raw = (policy.get("enumeration_scales") or {}).get(str(z)) if raw: return [int(v) for v in raw] return _scale_ladder(policy, z) +def _resolve_tile_enumeration_mode( + requested: str, + aoi_bbox_wgs84: tuple[float, float, float, float] | None, +) -> str: + mode = str(requested or "auto").strip().lower() + if mode not in TILE_ENUMERATION_MODES: + raise ValueError(f"unsupported tile enumeration mode: {requested}") + if mode == "auto": + return "aoi-complete" if aoi_bbox_wgs84 is not None else "coverage" + if mode == "aoi-complete" and aoi_bbox_wgs84 is None: + raise ValueError("aoi-complete requires --aoi-bbox") + return mode + + +def _job_key_set(rows: Iterable[dict[str, Any]]) -> set[tuple[int, int, int]]: + return { + (int(row["z"]), int(row["x"]), int(row["y"])) + for row in rows + } + + +def _validate_aoi_complete_plan_jobs(plan: dict[str, Any]) -> None: + if str(plan.get("tile_enumeration_mode") or "coverage") != "aoi-complete": + return + bbox = (plan.get("aoi") or {}).get("bbox_wgs84") + if not bbox or len(bbox) != 4: + raise RuntimeError("aoi-complete plan missing aoi.bbox_wgs84") + west, south, east, north = (float(v) for v in bbox) + minx, miny = lonlat_to_mercator(west, south) + maxx, maxy = lonlat_to_mercator(east, north) + aoi_box_m = box(minx, miny, maxx, maxy) + expected: set[tuple[int, int, int]] = set() + for z in parse_zooms(plan.get("zooms") or []): + expected |= _tiles_intersecting_geom(z, aoi_box_m) + actual = _job_key_set(plan.get("jobs") or []) + if actual != expected: + missing = sorted(expected - actual) + extra = sorted(actual - expected) + raise RuntimeError( + "aoi-complete job footprint mismatch: " + f"missing={len(missing)} extra={len(extra)} " + f"missing_examples={missing[:8]} extra_examples={extra[:8]}" + ) + + def _mercator_geom_from_wkb(value: Any): if value is None: return None @@ -241,6 +282,7 @@ def create_production_plan( s52_mariner_mode: str = "runtime", aoi_name: str | None = None, aoi_bbox_wgs84: tuple[float, float, float, float] | None = None, + tile_enumeration_mode: str = "auto", ) -> dict[str, Any]: if profile not in {"delivery", "engineering"}: raise ValueError("profile must be delivery or engineering") @@ -274,7 +316,14 @@ def create_production_plan( maxx, maxy = lonlat_to_mercator(east, north) aoi_box_m = box(minx, miny, maxx, maxy) - all_scales = sorted({s for z in zooms for s in _enumeration_scales(policy, z)}) + resolved_tile_enumeration_mode = _resolve_tile_enumeration_mode( + tile_enumeration_mode, + aoi_bbox_wgs84, + ) + if resolved_tile_enumeration_mode == "aoi-complete": + all_scales = sorted({s for z in zooms for s in _scale_ladder(policy, z)}) + else: + all_scales = sorted({s for z in zooms for s in _enumeration_scales(policy, z)}) with _connect(db) as conn: try: @@ -292,7 +341,11 @@ def create_production_plan( ladder = _scale_ladder(policy, z) enumeration = _enumeration_scales(policy, z) enumeration_set = set(enumeration) - raw_z_pieces = [(cid, scale, geom, src) for cid, scale, geom, src in pieces if scale in enumeration_set] + raw_z_pieces = [ + (cid, scale, geom, src) + for cid, scale, geom, src in pieces + if scale in enumeration_set + ] z_pieces = [] for cid, scale, geom, src in raw_z_pieces: if aoi_box_m is not None: @@ -302,19 +355,57 @@ def create_production_plan( if geom.is_empty: continue z_pieces.append((cid, scale, geom, src)) - z_jobs: set[tuple[int, int, int]] = set() + + coverage_jobs: set[tuple[int, int, int]] = set() fallback_pieces = 0 for _cid, _scale, geom, source in z_pieces: if source == "cell_bbox_fallback": fallback_pieces += 1 - z_jobs |= _tiles_intersecting_geom(z, geom) + coverage_jobs |= _tiles_intersecting_geom(z, geom) + + expected_aoi_jobs: set[tuple[int, int, int]] = set() + ladder_coverage_jobs: set[tuple[int, int, int]] = set() + ladder_pieces = [] + if resolved_tile_enumeration_mode == "aoi-complete": + assert aoi_box_m is not None + expected_aoi_jobs = _tiles_intersecting_geom(z, aoi_box_m) + ladder_set = set(ladder) + raw_ladder_pieces = [ + (cid, scale, geom, src) + for cid, scale, geom, src in pieces + if scale in ladder_set + ] + for cid, scale, geom, src in raw_ladder_pieces: + if not geom.intersects(aoi_box_m): + continue + clipped = geom.intersection(aoi_box_m) + if clipped.is_empty: + continue + ladder_pieces.append((cid, scale, clipped, src)) + ladder_coverage_jobs |= _tiles_intersecting_geom(z, clipped) + uncovered = sorted(expected_aoi_jobs - ladder_coverage_jobs) + if uncovered: + raise RuntimeError( + f"aoi-complete source coverage incomplete at z{z}: " + f"missing={len(uncovered)} examples={uncovered[:12]}; " + "fix scale_ladders/AOI before production" + ) + z_jobs = expected_aoi_jobs + else: + z_jobs = coverage_jobs + jobs |= z_jobs source_stats[str(z)] = { + "tile_enumeration_mode": resolved_tile_enumeration_mode, "enumeration_scales": enumeration, "scale_ladder": ladder, "coverage_pieces_before_aoi": len(raw_z_pieces), "coverage_pieces": len(z_pieces), "cell_bbox_fallback_pieces": fallback_pieces, + "coverage_tile_count": len(coverage_jobs), + "ladder_coverage_pieces": len(ladder_pieces), + "ladder_coverage_tile_count": len(ladder_coverage_jobs), + "expected_aoi_tile_count": len(expected_aoi_jobs), "tile_count": len(z_jobs), } @@ -341,6 +432,7 @@ def create_production_plan( "dataset_fingerprint": fingerprint, "profile": profile, "zooms": zooms, + "tile_enumeration_mode": resolved_tile_enumeration_mode, "mapping_sha256": sha256_file(mapping_path), "policy_sha256": sha256_file(policy_path), "builder_contract_version": BUILDER_CONTRACT_VERSION, @@ -372,6 +464,7 @@ def load_plan(path: Path) -> dict[str, Any]: obj = json.loads(Path(path).read_text(encoding="utf-8")) if obj.get("schema") not in SUPPORTED_PRODUCTION_PLAN_SCHEMAS: raise RuntimeError(f"unsupported plan schema: {obj.get('schema')}") + _validate_aoi_complete_plan_jobs(obj) return obj @@ -795,11 +888,12 @@ def migrate_production_release( (profile,), ).fetchall()) old_jobs = {(profile, int(r["z"]), int(r["x"]), int(r["y"])) for r in old_rows} - if old_jobs != new_jobs: - missing_old = len(new_jobs - old_jobs) - extra_old = len(old_jobs - new_jobs) + extra_old_jobs = sorted(old_jobs - new_jobs) + added_new_jobs = sorted(new_jobs - old_jobs) + if extra_old_jobs: raise RuntimeError( - f"old/new job set mismatch: missing_old={missing_old} extra_old={extra_old}" + "new plan does not contain all old jobs; " + f"extra_old={len(extra_old_jobs)} examples={extra_old_jobs[:8]}" ) changed_classes = set(_load_delivery_changed_classes(Path(new_plan["mapping_path"]))) @@ -890,6 +984,8 @@ def migrate_production_release( "production_pipeline_version": new_plan.get("production_pipeline_version"), "s52": new_plan.get("s52"), "aoi": new_plan.get("aoi"), + "tile_enumeration_mode": new_plan.get("tile_enumeration_mode"), + "added_new_jobs": len(added_new_jobs), "changed_delivery_classes": sorted(changed_classes), "reused_pass_jobs": reused, "reused_pbf_files": reused_pbf,