feat: add complete AOI production footprint

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))

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.
+    """Return scales allowed to materialize physical tiles in coverage mode."""
-
-    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.
-    """
     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,6 +316,13 @@ def create_production_plan(
         maxx, maxy = lonlat_to_mercator(east, north)
         aoi_box_m = box(minx, miny, maxx, maxy)
 
+    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:
@@ -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:
+    extra_old_jobs = sorted(old_jobs - new_jobs)
-        missing_old = len(new_jobs - old_jobs)
+    added_new_jobs = sorted(new_jobs - old_jobs)
-        extra_old = len(old_jobs - new_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,