CM95/tools/cm95_pbf_output_audit_v1.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations

import argparse
from collections import Counter, defaultdict
import json
from pathlib import Path

import mapbox_vector_tile
import yaml


DELIVERY_ALLOWLIST = {
    "canonical_object_type",
    "class_code",
    "chart_fill_pattern",
    "chart_fill_style",
    "chart_icon_image",
    "chart_label_position_code",
    "chart_label_subtext",
    "chart_label_text",
    "chart_line_color",
    "chart_line_width",
    "chart_symbol_code",
    "chart_text_color",
    "chart_text_style",
    "clearance_height_m",
    "depth_value_m",
    "display_code",
    "least_depth_m",
    "light_color_code",
    "light_sector_mode",
    "arc_id",
    "arc_source_code",
    "name_ja",
    "place_name_en",
    "shape_class_code",
}

SPRITE_FIELDS = {"chart_icon_image", "chart_fill_pattern", "chart_line_pattern", "arc_id"}


def parse_args():
    p = argparse.ArgumentParser()
    p.add_argument("--pbf-root", required=True)
    p.add_argument("--style", required=True)
    p.add_argument("--sprite-json", required=True)
    p.add_argument("--policy", default="config/pbf/cm95_tile_policy_v1.yaml")
    p.add_argument("--profile", choices=["delivery", "engineering"], default="delivery")
    p.add_argument("--output", default="cm95_pbf_output_audit_v1.json")
    return p.parse_args()


def main():
    a = parse_args()
    root = Path(a.pbf_root)
    style = json.loads(Path(a.style).read_text(encoding="utf-8"))
    sprite = json.loads(Path(a.sprite_json).read_text(encoding="utf-8"))
    policy = yaml.safe_load(Path(a.policy).read_text(encoding="utf-8"))

    style_layers = {
        layer["source-layer"]
        for layer in style.get("layers", [])
        if layer.get("source") == "navsea_delivery" and layer.get("source-layer")
    }
    sprite_keys = set(sprite)

    layer_counts = Counter()
    extent_counts = Counter()
    unknown_layers = Counter()
    property_failures = []
    sprite_failures = []
    decode_failures = []
    feature_ids = defaultdict(set)
    duplicate_ids = []

    files = sorted(root.rglob("*.pbf"))
    for path in files:
        try:
            z = int(path.relative_to(root).parts[0])
            decoded = mapbox_vector_tile.decode(path.read_bytes())
        except Exception as exc:
            decode_failures.append({"file": str(path), "error": str(exc)})
            continue

        expected_extent = int(
            policy.get("extent_by_zoom", {}).get(str(z), policy.get("default_extent", 4096))
        )

        for layer_name, payload in decoded.items():
            features = payload.get("features", [])
            layer_counts[layer_name] += len(features)
            extent = int(payload.get("extent") or 4096)
            extent_counts[f"z{z}:{extent}"] += 1

            if extent != expected_extent:
                decode_failures.append(
                    {
                        "file": str(path),
                        "layer": layer_name,
                        "error": f"extent {extent} != expected {expected_extent}",
                    }
                )

            if layer_name not in style_layers and not (
                a.profile == "engineering" and layer_name == "cm95_unmapped"
            ):
                unknown_layers[layer_name] += len(features)

            for f in features:
                fid = f.get("id")
                if fid is not None:
                    key = (str(path), layer_name)
                    if fid in feature_ids[key]:
                        duplicate_ids.append(
                            {"file": str(path), "layer": layer_name, "id": fid}
                        )
                    feature_ids[key].add(fid)

                props = f.get("properties") or {}
                if a.profile == "delivery":
                    extras = sorted(set(props) - DELIVERY_ALLOWLIST)
                    if extras:
                        property_failures.append(
                            {
                                "file": str(path),
                                "layer": layer_name,
                                "id": fid,
                                "extra_properties": extras,
                            }
                        )

                for field in SPRITE_FIELDS:
                    value = props.get(field)
                    if value and value not in sprite_keys:
                        # arc_id may intentionally reference arc sprite; same JSON key contract applies.
                        sprite_failures.append(
                            {
                                "file": str(path),
                                "layer": layer_name,
                                "id": fid,
                                "field": field,
                                "value": value,
                            }
                        )

    failure_count = (
        len(decode_failures)
        + sum(unknown_layers.values())
        + len(property_failures)
        + len(sprite_failures)
        + len(duplicate_ids)
    )

    result = {
        "schema": "navsea.cm95.pbf-output-audit/v1",
        "profile": a.profile,
        "pbf_file_count": len(files),
        "feature_count": sum(layer_counts.values()),
        "layer_counts": dict(layer_counts),
        "extent_counts": dict(extent_counts),
        "unknown_layers": dict(unknown_layers),
        "decode_or_extent_failures": decode_failures,
        "delivery_property_failures": property_failures,
        "sprite_failures": sprite_failures,
        "duplicate_feature_ids_within_layer_tile": duplicate_ids,
        "failure_count": failure_count,
        "overall": "PASS" if failure_count == 0 else "FAIL",
    }

    Path(a.output).write_text(
        json.dumps(result, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    print(f"PBF files       : {result['pbf_file_count']}")
    print(f"Features        : {result['feature_count']}")
    print(f"Layers          : {len(layer_counts)}")
    print(f"Failures        : {failure_count}")
    print(f"OVERALL         : {result['overall']}")


if __name__ == "__main__":
    main()