from __future__ import annotations from collections import Counter, defaultdict import json from pathlib import Path from typing import Any import mapbox_vector_tile SPRITE_PROPERTY_FIELDS = ( "chart_icon_image", "chart_fill_pattern", "chart_line_pattern", ) def ev(expr: Any, props: dict[str, Any]): if not isinstance(expr, list): return expr if not expr: return None op = expr[0] if op == "get": return props.get(expr[1]) if op == "has": return expr[1] in props if op == "!": return not bool(ev(expr[1], props)) if op == "all": return all(bool(ev(x, props)) for x in expr[1:]) if op == "any": return any(bool(ev(x, props)) for x in expr[1:]) if op == "==": return ev(expr[1], props) == ev(expr[2], props) if op == "!=": return ev(expr[1], props) != ev(expr[2], props) if op == "slice": value = ev(expr[1], props) if value is None: return None start = int(ev(expr[2], props)) end = int(ev(expr[3], props)) if len(expr) > 3 else None return str(value)[start:end] if op == "to-number": value = ev(expr[1], props) try: num = float(value) return int(num) if num.is_integer() else num except Exception: return 0 if op == "literal": return expr[1] if op == "match": value = ev(expr[1], props) pairs = expr[2:-1] default = expr[-1] for i in range(0, len(pairs), 2): label = pairs[i] output = pairs[i + 1] if isinstance(label, list) and label and label[0] == "literal": labels = label[1] if value in labels: return ev(output, props) elif isinstance(label, list): if value in label: return ev(output, props) elif value == label: return ev(output, props) return ev(default, props) raise RuntimeError(f"unsupported style filter op: {op}") def audit_pbf_style_and_sprite( pbf_root: Path, style_path: Path, sprite_json_path: Path | None = None, ) -> dict[str, Any]: pbf_root = Path(pbf_root) style = json.loads(Path(style_path).read_text(encoding="utf-8")) sprite_keys = None if sprite_json_path is not None: sprite_obj = json.loads(Path(sprite_json_path).read_text(encoding="utf-8")) sprite_keys = set(sprite_obj) style_by_source_layer = defaultdict(list) for layer in style.get("layers", []): if layer.get("source") == "navsea_delivery" and layer.get("source-layer"): style_by_source_layer[layer["source-layer"]].append(layer) files = sorted(pbf_root.rglob("*.pbf")) contract_unmatched = Counter() zoom_hidden = Counter() visible = Counter() sprite_missing = Counter() sprite_values = Counter() total_features = 0 for path in files: rel = path.relative_to(pbf_root) if len(rel.parts) < 3: raise RuntimeError(f"PBF path must end with z/x/y.pbf: {path}") z = int(rel.parts[-3]) decoded = mapbox_vector_tile.decode(path.read_bytes()) for source_layer, layer in decoded.items(): style_layers = style_by_source_layer.get(source_layer, []) for feature in layer.get("features", []): total_features += 1 props = feature.get("properties") or {} filter_matches = [] for sl in style_layers: flt = sl.get("filter") if flt is None or bool(ev(flt, props)): filter_matches.append(sl) key = (z, source_layer) if not filter_matches: contract_unmatched[key] += 1 else: zoom_matches = [] for sl in filter_matches: minz = sl.get("minzoom") maxz = sl.get("maxzoom") if minz is not None and z < minz: continue if maxz is not None and z >= maxz: continue zoom_matches.append(sl) if zoom_matches: visible[key] += 1 else: zoom_hidden[key] += 1 if sprite_keys is not None: for field in SPRITE_PROPERTY_FIELDS: value = props.get(field) if isinstance(value, str) and value: sprite_values[(field, value)] += 1 if value not in sprite_keys: sprite_missing[(field, value)] += 1 return { "schema": "navsea.cm95.style-sprite-contract-audit/v1", "pbf_file_count": len(files), "feature_count": total_features, "style_source_layer_count": len(style_by_source_layer), "contract_unmatched_count": sum(contract_unmatched.values()), "zoom_hidden_count": sum(zoom_hidden.values()), "visible_count": sum(visible.values()), "sprite_key_count": None if sprite_keys is None else len(sprite_keys), "sprite_missing_count": sum(sprite_missing.values()), "contract_unmatched": [ {"z": z, "source_layer": sl, "count": count} for (z, sl), count in sorted(contract_unmatched.items()) ], "zoom_hidden": [ {"z": z, "source_layer": sl, "count": count} for (z, sl), count in sorted(zoom_hidden.items()) ], "sprite_missing": [ {"field": field, "sprite_key": key, "count": count} for (field, key), count in sorted(sprite_missing.items()) ], "sprite_values": [ {"field": field, "sprite_key": key, "count": count} for (field, key), count in sorted(sprite_values.items()) ], "overall": ( "PASS" if sum(contract_unmatched.values()) == 0 and sum(sprite_missing.values()) == 0 else "FAIL" ), }