"""Publish privacy-reviewed aggregate JSON assets for the dashboard. The exporter never loads a model or emits row-level predictions. It reads only the train, tune, and calibration portions of the private feature mart. The 2025 locked-test partition is outside every aggregation query. """ from __future__ import annotations import argparse import hashlib import json import math import os import re import shutil import sys import tempfile from dataclasses import dataclass from pathlib import Path from typing import Dict, Iterable, List, Mapping, Optional, Sequence, Tuple import duckdb PROJECT_ROOT = Path(__file__).resolve().parents[1] PUBLIC_DATA_ROOT = (PROJECT_ROOT / "dashboard/public/data").resolve() SCHEMA_VERSION = "dashboard_data_v1" PUBLICATION_POLICY_VERSION = "dashboard_publication_policy_v1" ALLOWED_PARTITIONS = ("train", "tune", "calibrate") PARTITION_ORDER = {name: index for index, name in enumerate(ALLOWED_PARTITIONS)} LOCKED_PARTITION_ALIASES = ("test", "locked_test", "locked-test") MIN_SUPPORT = 100 MIN_CLASS_COUNT = 10 SUPPORT_ROUNDING = 100 MAX_SCORECARDS = 200 ROW_ASSET_SCHEMAS = { "overview_period_county.json": ( "year", "quarter", "public_county", "support_rounded", "nonpass_rate", ), "age_risk_curve.json": ( "age_band", "support_rounded", "nonpass_rate", ), "cohort_scorecard.json": ( "prior_make", "prior_model", "support_rounded", "nonpass_rate", ), "model_diagnostics.json": ( "model", "partition", "average_precision", "brier", "log_loss", "roc_auc", "top_10_capture", ), "coverage_quality.json": ( "year", "source_era", "support_rounded", "labeled_rate", "overall_result_share", "utah_obd_proxy_share", ), } PUBLIC_ASSET_NAMES = tuple( sorted( list(ROW_ASSET_SCHEMAS) + ["filter_catalog.json", "data_manifest.json"] ) ) SHA256_MANIFEST_NAME = "sha256_manifest.json" PUBLIC_OUTPUT_NAMES = frozenset(PUBLIC_ASSET_NAMES + (SHA256_MANIFEST_NAME,)) REQUIRED_MART_COLUMNS = { "vehicle_token", "is_vin_audit", "episode_number", "episode_start", "first_outcome", "target_outcome_label_source", "target_nonpass", "eligible_returning_target", "temporal_partition", "public_county", "source_era", "vehicle_age", "last_observed_make", "last_observed_model", } DENIED_EXACT_KEYS = { "credential", "credentials", "inspector", "inspector_id", "password", "secret", "technician", "technician_id", "vehicle_token", "vehicle_bucket", "is_vin_audit", "episode_number", "episode_start", "first_outcome", "target_nonpass", "target_outcome_label_source", "outcome_label_source", "internal_event_id", "vin", "plate", "zip", "zipcode", "station", "station_id", "raw_json", "row_prediction", "prediction", "probability", "pass_count", "nonpass_count", "row_count", } DENIED_KEY_FRAGMENTS = ( "credential", "inspector", "password", "secret", "technician", "vehicle_token", "internal_event", "license_plate", "exact_timestamp", "row_prediction", "raw_", "label_source", ) EXACT_TIMESTAMP_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}") VIN_LIKE_PATTERN = re.compile(r"^[A-HJ-NPR-Z0-9]{17}$") SAFE_CATEGORY_PATTERN = re.compile(r"^[^\x00-\x1f\x7f]{1,80}$") SAFE_MODEL_NAME_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$") class DashboardExportError(Exception): """Base class for expected publication failures.""" class ManifestError(DashboardExportError): """Raised when private input provenance cannot be verified.""" class PublicSchemaError(DashboardExportError): """Raised when an asset violates the public-data contract.""" @dataclass(frozen=True) class ModelBundle: manifest: Path metrics: Path def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--mart", required=True, type=Path) parser.add_argument("--mart-manifest", type=Path) parser.add_argument( "--model-manifest", required=True, action="append", type=Path, help="Private model manifest; repeat once per model bundle.", ) parser.add_argument( "--model-metrics", required=True, action="append", type=Path, help="Metrics JSON paired by position with --model-manifest.", ) parser.add_argument("--output-dir", type=Path, default=PUBLIC_DATA_ROOT) parser.add_argument("--development-preview", action="store_true") parser.add_argument("--overwrite", action="store_true") return parser.parse_args(argv) def _reject_json_constant(value: str) -> object: raise ManifestError("JSON contains a non-finite numeric constant: " + value) def _unique_object(pairs: Iterable[Tuple[str, object]]) -> Dict[str, object]: result: Dict[str, object] = {} for key, value in pairs: if key in result: raise ManifestError("JSON contains duplicate key: " + key) result[key] = value return result def load_json(path: Path) -> object: try: if path.stat().st_size > 20 * 1024 * 1024: raise ManifestError("Refusing an unexpectedly large JSON input") return json.loads( path.read_text(encoding="utf-8"), object_pairs_hook=_unique_object, parse_constant=_reject_json_constant, ) except OSError as exc: raise ManifestError("Cannot read required JSON input") from exc except UnicodeError as exc: raise ManifestError("JSON input is not valid UTF-8") from exc except json.JSONDecodeError as exc: raise ManifestError("JSON input is malformed") from exc def file_sha256(path: Path) -> str: digest = hashlib.sha256() try: with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) except OSError as exc: raise ManifestError("Cannot read a required private input") from exc return digest.hexdigest() def _require_mapping(value: object, label: str) -> Mapping[str, object]: if not isinstance(value, dict): raise ManifestError(label + " must be a JSON object") return value def _mart_manifest_path(mart: Path, explicit: Optional[Path]) -> Path: if explicit is not None: return explicit return mart.with_name(mart.name + ".manifest.json") def validate_mart_manifest( mart: Path, manifest_path: Path, development_preview: bool, ) -> Tuple[Mapping[str, object], bool, str]: manifest = _require_mapping(load_json(manifest_path), "Mart manifest") if manifest.get("build_kind") != "private_leakage_safe_inspection_feature_mart": raise ManifestError("Input is not the approved private feature mart") classification = manifest.get("classification") if not isinstance(classification, str) or not classification.startswith("private_"): raise ManifestError("Feature mart classification is not private") expected_digest = manifest.get("parquet_sha256") if not isinstance(expected_digest, str) or not re.fullmatch( r"[0-9a-f]{64}", expected_digest ): raise ManifestError("Mart manifest has no valid Parquet digest") actual_digest = file_sha256(mart) if actual_digest != expected_digest: raise ManifestError("Feature mart digest does not match its manifest") population_allowed = manifest.get("population_estimate_allowed") if not isinstance(population_allowed, bool): raise ManifestError("Mart population-estimate flag must be boolean") source_kind = manifest.get("source_data_kind") if not isinstance(source_kind, str): raise ManifestError("Mart source-data kind is missing") source_kind_lower = source_kind.lower() development_source = ( not population_allowed or "development" in source_kind_lower or "sample" in source_kind_lower ) if development_source and not development_preview: raise ManifestError( "Development-sample marts require --development-preview" ) if not development_preview and not population_allowed: raise ManifestError("Population publication requires an approved mart") columns = manifest.get("columns") if not isinstance(columns, list): raise ManifestError("Mart manifest has no column contract") declared_names = { item.get("name") for item in columns if isinstance(item, dict) and isinstance(item.get("name"), str) } missing = sorted(REQUIRED_MART_COLUMNS - declared_names) if missing: raise ManifestError("Mart manifest is missing required columns") return manifest, development_source, actual_digest def load_model_diagnostics( bundles: Sequence[ModelBundle], mart_digest: str, partition_vehicle_support: Mapping[str, Tuple[int, int]], ) -> Tuple[List[Dict[str, object]], List[Tuple[str, str]]]: diagnostics: List[Dict[str, object]] = [] model_provenance: List[Tuple[str, str]] = [] seen = set() for bundle in bundles: manifest = _require_mapping(load_json(bundle.manifest), "Model manifest") if manifest.get("locked_test_evaluated") is not False: raise ManifestError( "Every model manifest must state locked_test_evaluated=false" ) classification = manifest.get("classification") if not isinstance(classification, str) or "no_row_predictions" not in classification: raise ManifestError("Model artifact classification is not publishable") if manifest.get("input_sha256") != mart_digest: raise ManifestError("Model metrics were not built from this mart") expected_metrics_digest = manifest.get("metrics_sha256") if not isinstance(expected_metrics_digest, str) or not re.fullmatch( r"[0-9a-f]{64}", expected_metrics_digest ): raise ManifestError("Model manifest has no valid metrics digest") metrics_digest = file_sha256(bundle.metrics) if metrics_digest != expected_metrics_digest: raise ManifestError("Model metrics digest does not match its manifest") model_version = manifest.get("model_version") if not isinstance(model_version, str) or not SAFE_MODEL_NAME_PATTERN.fullmatch( model_version ): raise ManifestError("Model manifest has an unsafe model version") model_provenance.append((model_version, metrics_digest)) payload = load_json(bundle.metrics) if isinstance(payload, dict): payload = payload.get("metrics") if not isinstance(payload, list): raise ManifestError("Model metrics must be a JSON array") for raw in payload: if not isinstance(raw, dict): raise ManifestError("Each model metric must be an object") partition = raw.get("partition") if partition not in ALLOWED_PARTITIONS: raise ManifestError( "Metrics contain a locked, shadow, or unknown partition" ) if raw.get("cohort") != "non_audit": continue episodes = raw.get("episodes") nonpass = raw.get("nonpass") vehicles = raw.get("vehicles") if ( isinstance(episodes, bool) or not isinstance(episodes, int) or isinstance(nonpass, bool) or not isinstance(nonpass, int) or isinstance(vehicles, bool) or not isinstance(vehicles, int) or vehicles < 0 or vehicles > episodes or nonpass < 0 or nonpass > episodes ): raise ManifestError( "Model metrics lack valid private suppression counts" ) class_vehicle_support = partition_vehicle_support.get(partition) if class_vehicle_support is None or not _publishable( episodes, nonpass, vehicles, class_vehicle_support[0], class_vehicle_support[1], ): continue model = raw.get("model") if not isinstance(model, str) or not SAFE_MODEL_NAME_PATTERN.fullmatch(model): raise ManifestError("Model metric has an unsafe model name") row: Dict[str, object] = {"model": model, "partition": partition} for key in ( "average_precision", "brier", "log_loss", "roc_auc", "top_10_capture", ): value = raw.get(key) if isinstance(value, bool) or not isinstance(value, (int, float)): raise ManifestError("Model metric is missing a numeric value") number = float(value) if not math.isfinite(number) or number < 0.0: raise ManifestError("Model metric is non-finite or negative") if key != "log_loss" and number > 1.0: raise ManifestError("Bounded model metric exceeds one") row[key] = round(number, 4) identity = (model, partition) if identity in seen: raise ManifestError("Duplicate model/partition diagnostic") seen.add(identity) diagnostics.append(row) if not diagnostics: raise ManifestError("No approved train/tune/calibrate diagnostics found") diagnostics.sort( key=lambda row: (str(row["model"]), PARTITION_ORDER[str(row["partition"])]) ) model_provenance.sort() return diagnostics, model_provenance def _round_support(value: int) -> int: if value < MIN_SUPPORT: raise PublicSchemaError("Unsuppressed support is below the threshold") return int(math.floor((value + SUPPORT_ROUNDING / 2) / SUPPORT_ROUNDING)) * SUPPORT_ROUNDING def _rate(numerator: int, denominator: int) -> float: if denominator <= 0 or numerator < 0 or numerator > denominator: raise PublicSchemaError("Invalid aggregate rate inputs") return round(numerator / denominator, 3) def _publishable( n: int, nonpass: int, vehicles: int, pass_vehicles: int, nonpass_vehicles: int, ) -> bool: return ( n >= MIN_SUPPORT and nonpass >= MIN_CLASS_COUNT and n - nonpass >= MIN_CLASS_COUNT and vehicles >= MIN_SUPPORT and pass_vehicles >= MIN_CLASS_COUNT and nonpass_vehicles >= MIN_CLASS_COUNT ) def _safe_category(value: object, label: str) -> str: if not isinstance(value, str): raise PublicSchemaError(label + " category is not text") normalized = value.strip() if not SAFE_CATEGORY_PATTERN.fullmatch(normalized): raise PublicSchemaError(label + " category contains unsafe characters") compact = re.sub(r"[^A-Za-z0-9]", "", normalized).upper() if VIN_LIKE_PATTERN.fullmatch(compact): raise PublicSchemaError(label + " category resembles a VIN") return normalized def _create_safe_mart(connection: duckdb.DuckDBPyConnection, mart: Path) -> None: description = connection.execute( "DESCRIBE SELECT * FROM read_parquet(?)", [str(mart)] ).fetchall() actual_columns = {str(row[0]) for row in description} if not REQUIRED_MART_COLUMNS.issubset(actual_columns): raise ManifestError("Parquet schema does not match the mart contract") boundary_errors = connection.execute( """ SELECT count(*) FROM read_parquet(?) WHERE (temporal_partition IN ('train', 'tune', 'calibrate') AND episode_start >= TIMESTAMP '2025-01-01') OR (temporal_partition IN ('test', 'locked_test', 'locked-test') AND episode_start < TIMESTAMP '2025-01-01') """, [str(mart)], ).fetchone()[0] if boundary_errors: raise ManifestError("Mart partition timestamps violate the locked boundary") # The filter is applied before target fields enter this temporary table. # No subsequent query can read or evaluate a 2025 target. connection.execute( """ CREATE TEMP TABLE safe_mart AS SELECT vehicle_token, is_vin_audit, episode_number, episode_start, first_outcome, target_outcome_label_source, target_nonpass, eligible_returning_target, temporal_partition, public_county, source_era, vehicle_age, last_observed_make, last_observed_model FROM read_parquet(?) WHERE temporal_partition IN ('train', 'tune', 'calibrate') AND episode_start < TIMESTAMP '2025-01-01' """, [str(mart)], ) def _partition_vehicle_support( connection: duckdb.DuckDBPyConnection, ) -> Dict[str, Tuple[int, int]]: """Return private class-level entity supports for diagnostic suppression.""" raw = connection.execute( """ SELECT temporal_partition, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 0 )::BIGINT AS pass_vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 1 )::BIGINT AS nonpass_vehicles FROM safe_mart WHERE eligible_returning_target AND target_nonpass IN (0, 1) AND NOT is_vin_audit GROUP BY 1 """ ).fetchall() return { str(partition): (int(pass_vehicles), int(nonpass_vehicles)) for partition, pass_vehicles, nonpass_vehicles in raw } def _overview_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]: raw = connection.execute( """ SELECT year(episode_start)::INTEGER AS year, quarter(episode_start)::INTEGER AS quarter, public_county, count(*)::BIGINT AS n, sum(target_nonpass)::BIGINT AS nonpass, count(DISTINCT vehicle_token)::BIGINT AS vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 0 )::BIGINT AS pass_vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 1 )::BIGINT AS nonpass_vehicles FROM safe_mart WHERE eligible_returning_target AND target_nonpass IN (0, 1) AND public_county IS NOT NULL GROUP BY 1, 2, 3 ORDER BY 1, 2, 3 """ ).fetchall() rows = [] for ( year, quarter, county, n, nonpass, vehicles, pass_vehicles, nonpass_vehicles, ) in raw: if not _publishable( n, nonpass, vehicles, pass_vehicles, nonpass_vehicles ): continue rows.append( { "year": int(year), "quarter": int(quarter), "public_county": _safe_category(county, "County"), "support_rounded": _round_support(n), "nonpass_rate": _rate(nonpass, n), } ) return rows AGE_BAND_SQL = """CASE WHEN vehicle_age BETWEEN 0 AND 3 THEN '0-3' WHEN vehicle_age BETWEEN 4 AND 7 THEN '4-7' WHEN vehicle_age BETWEEN 8 AND 11 THEN '8-11' WHEN vehicle_age BETWEEN 12 AND 15 THEN '12-15' WHEN vehicle_age BETWEEN 16 AND 20 THEN '16-20' WHEN vehicle_age >= 21 THEN '21+' ELSE NULL END""" AGE_BAND_ORDER = {name: index for index, name in enumerate( ("0-3", "4-7", "8-11", "12-15", "16-20", "21+") )} def _age_risk_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]: raw = connection.execute( f""" SELECT {AGE_BAND_SQL} AS age_band, count(*)::BIGINT AS n, sum(target_nonpass)::BIGINT AS nonpass, count(DISTINCT vehicle_token)::BIGINT AS vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 0 )::BIGINT AS pass_vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 1 )::BIGINT AS nonpass_vehicles FROM safe_mart WHERE eligible_returning_target AND target_nonpass IN (0, 1) GROUP BY 1 ORDER BY 1 """ ).fetchall() rows = [] for ( age_band, n, nonpass, vehicles, pass_vehicles, nonpass_vehicles, ) in raw: if age_band is None or not _publishable( n, nonpass, vehicles, pass_vehicles, nonpass_vehicles ): continue if age_band not in AGE_BAND_ORDER: raise PublicSchemaError("Unexpected age band") rows.append( { "age_band": age_band, "support_rounded": _round_support(n), "nonpass_rate": _rate(nonpass, n), } ) rows.sort(key=lambda row: AGE_BAND_ORDER[str(row["age_band"])]) return rows def _scorecard_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]: raw = connection.execute( """ WITH legacy_cells(prior_make, prior_model) AS ( VALUES ('FORD', 'F150'), ('HONDA', 'ACCORD'), ('HONDA', 'CIVIC'), ('TOYOTA', 'CAMRY') ), make_aliases(raw_make, canonical_make) AS ( VALUES ('CHEVR', 'CHEVROLET'), ('HYUND', 'HYUNDAI'), ('NISSA', 'NISSAN'), ('SUBAR', 'SUBARU'), ('TOYOT', 'TOYOTA') ), normalized AS ( SELECT safe_mart.last_observed_make AS raw_make, safe_mart.last_observed_model AS raw_model, coalesce( make_aliases.canonical_make, upper(trim(safe_mart.last_observed_make)) ) AS normalized_make, upper(trim(safe_mart.last_observed_model)) AS normalized_model, exact_legacy.prior_make IS NOT NULL AS is_exact_legacy, display_legacy.prior_make IS NOT NULL AS collides_with_legacy, safe_mart.vehicle_token, safe_mart.target_nonpass FROM safe_mart LEFT JOIN make_aliases ON upper(trim(safe_mart.last_observed_make)) = make_aliases.raw_make LEFT JOIN legacy_cells AS exact_legacy ON safe_mart.last_observed_make = exact_legacy.prior_make AND safe_mart.last_observed_model = exact_legacy.prior_model LEFT JOIN legacy_cells AS display_legacy ON trim(safe_mart.last_observed_make) = display_legacy.prior_make AND trim(safe_mart.last_observed_model) = display_legacy.prior_model WHERE safe_mart.eligible_returning_target AND safe_mart.target_nonpass IN (0, 1) AND safe_mart.last_observed_make IS NOT NULL AND safe_mart.last_observed_model IS NOT NULL ), classified AS ( SELECT CASE -- Only the eight families introduced in this release use -- alias folding. The four legacy cells never enter this -- branch, so their published counts retain raw-key meaning. WHEN normalized_make = 'TOYOTA' AND ( normalized_model = 'COROLLA' OR normalized_model LIKE 'COROLLA %' ) AND normalized_model != 'COROLLA IM' AND normalized_model NOT LIKE 'COROLLA IM %' THEN 'toyota_corolla' WHEN normalized_make = 'TOYOTA' AND ( normalized_model = '4RUNNER' OR normalized_model LIKE '4RUNNER %' ) THEN 'toyota_4runner' WHEN normalized_make = 'TOYOTA' AND ( normalized_model = 'TACOMA' OR normalized_model LIKE 'TACOMA %' ) THEN 'toyota_tacoma' WHEN normalized_make = 'DODGE' AND ( normalized_model = 'RAM 1500' OR normalized_model LIKE 'RAM 1500 %' OR normalized_model = 'RAM PICKUP 1500' OR normalized_model LIKE 'RAM PICKUP 1500 %' OR normalized_model = 'RAM1500' OR normalized_model LIKE 'RAM1500 %' ) AND normalized_model NOT LIKE '% VAN' AND normalized_model NOT LIKE '% VAN %' THEN 'dodge_ram_1500' WHEN normalized_make = 'CHEVROLET' AND ( normalized_model = 'SILVERADO 1500' OR normalized_model LIKE 'SILVERADO 1500 %' OR normalized_model = 'SILVERADO 1500HD' OR normalized_model LIKE 'SILVERADO 1500HD %' OR normalized_model = 'SILVERADO C1500' OR normalized_model LIKE 'SILVERADO C1500 %' OR normalized_model = 'SILVERADO K1500' OR normalized_model LIKE 'SILVERADO K1500 %' OR normalized_model = 'C1500 SILVERADO' OR normalized_model LIKE 'C1500 SILVERADO %' OR normalized_model = 'K15 SILVERADO' OR normalized_model LIKE 'K15 SILVERADO %' OR normalized_model = 'K1500 SILVERADO' OR normalized_model LIKE 'K1500 SILVERADO %' ) THEN 'chevrolet_silverado_1500' WHEN normalized_make = 'NISSAN' AND ( normalized_model = 'ALTIMA' OR normalized_model LIKE 'ALTIMA %' ) THEN 'nissan_altima' WHEN normalized_make = 'HYUNDAI' AND ( normalized_model = 'ELANTRA' OR normalized_model LIKE 'ELANTRA %' ) THEN 'hyundai_elantra' WHEN normalized_make = 'SUBARU' AND ( normalized_model = 'OUTBACK' OR normalized_model LIKE 'OUTBACK %' OR normalized_model = 'LEGACY OUTBACK' OR normalized_model LIKE 'LEGACY OUTBACK %' ) THEN 'subaru_outback' ELSE NULL END AS new_family, * FROM normalized ), publication_input AS ( -- Exact legacy keys are deliberately copied from the raw columns. SELECT raw_make AS prior_make, raw_model AS prior_model, vehicle_token, target_nonpass FROM classified WHERE is_exact_legacy UNION ALL -- Preserve the original raw-key behavior for unrelated cohorts. -- A whitespace variant that would render as a legacy key is -- withheld rather than creating a duplicate public identity. SELECT raw_make AS prior_make, raw_model AS prior_model, vehicle_token, target_nonpass FROM classified WHERE NOT is_exact_legacy AND NOT collides_with_legacy AND new_family IS NULL UNION ALL SELECT CASE new_family WHEN 'chevrolet_silverado_1500' THEN 'CHEVROLET' WHEN 'dodge_ram_1500' THEN 'DODGE' WHEN 'hyundai_elantra' THEN 'HYUNDAI' WHEN 'nissan_altima' THEN 'NISSAN' WHEN 'subaru_outback' THEN 'SUBARU' WHEN 'toyota_4runner' THEN 'TOYOTA' WHEN 'toyota_corolla' THEN 'TOYOTA' WHEN 'toyota_tacoma' THEN 'TOYOTA' ELSE NULL END AS prior_make, CASE new_family WHEN 'chevrolet_silverado_1500' THEN 'SILVERADO 1500' WHEN 'dodge_ram_1500' THEN 'RAM 1500' WHEN 'hyundai_elantra' THEN 'ELANTRA' WHEN 'nissan_altima' THEN 'ALTIMA' WHEN 'subaru_outback' THEN 'OUTBACK' WHEN 'toyota_4runner' THEN '4RUNNER' WHEN 'toyota_corolla' THEN 'COROLLA' WHEN 'toyota_tacoma' THEN 'TACOMA' ELSE NULL END AS prior_model, vehicle_token, target_nonpass FROM classified WHERE new_family IS NOT NULL ) SELECT prior_make, prior_model, count(*)::BIGINT AS n, sum(target_nonpass)::BIGINT AS nonpass, count(DISTINCT vehicle_token)::BIGINT AS vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 0 )::BIGINT AS pass_vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 1 )::BIGINT AS nonpass_vehicles FROM publication_input GROUP BY 1, 2 HAVING count(*) >= 100 AND sum(target_nonpass) >= 10 AND count(*) - sum(target_nonpass) >= 10 AND count(DISTINCT vehicle_token) >= 100 AND count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 0 ) >= 10 AND count(DISTINCT vehicle_token) FILTER ( WHERE target_nonpass = 1 ) >= 10 ORDER BY n DESC, prior_make, prior_model LIMIT 200 """ ).fetchall() rows = [] for ( make, model, n, nonpass, vehicles, pass_vehicles, nonpass_vehicles, ) in raw: if not _publishable( n, nonpass, vehicles, pass_vehicles, nonpass_vehicles ): continue try: safe_make = _safe_category(make, "Make") safe_model = _safe_category(model, "Model") except PublicSchemaError: continue rows.append( { "prior_make": safe_make, "prior_model": safe_model, "support_rounded": _round_support(n), "nonpass_rate": _rate(nonpass, n), } ) rows.sort(key=lambda row: (str(row["prior_make"]), str(row["prior_model"]))) return rows def _coverage_rows(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, object]]: raw = connection.execute( """ SELECT year(episode_start)::INTEGER AS year, source_era, count(*)::BIGINT AS n, count(*) FILTER ( WHERE first_outcome IN ('pass','fail','reject','abort') )::BIGINT AS labeled, count(*) FILTER (WHERE first_outcome = 'pass')::BIGINT AS pass, count(*) FILTER ( WHERE first_outcome IN ('fail','reject','abort') )::BIGINT AS nonpass, count(*) FILTER ( WHERE target_outcome_label_source = 'overall_result' )::BIGINT AS overall_labels, count(*) FILTER ( WHERE target_outcome_label_source = 'utah_obd_proxy' )::BIGINT AS proxy_labels, count(DISTINCT vehicle_token)::BIGINT AS vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE first_outcome = 'pass' )::BIGINT AS pass_vehicles, count(DISTINCT vehicle_token) FILTER ( WHERE first_outcome IN ('fail','reject','abort') )::BIGINT AS nonpass_vehicles FROM safe_mart WHERE episode_number > 1 AND source_era IS NOT NULL GROUP BY 1, 2 ORDER BY 1, 2 """ ).fetchall() rows = [] for ( year, source, n, labeled, passes, nonpass, overall, proxy, vehicles, pass_vehicles, nonpass_vehicles, ) in raw: if not _publishable( n, nonpass, vehicles, pass_vehicles, nonpass_vehicles ) or passes < MIN_CLASS_COUNT: continue if labeled != passes + nonpass or labeled != overall + proxy: raise PublicSchemaError("Label-source quality totals are inconsistent") rows.append( { "year": int(year), "source_era": _safe_category(source, "Source era"), "support_rounded": _round_support(n), "labeled_rate": _rate(labeled, n), "overall_result_share": _rate(overall, labeled), "utah_obd_proxy_share": _rate(proxy, labeled), } ) return rows def _base_asset( development_preview: bool, population_estimate_allowed: bool, ) -> Dict[str, object]: return { "schema_version": SCHEMA_VERSION, "development_preview": development_preview, "population_estimate_allowed": population_estimate_allowed, } def _release_id( mart_digest: str, model_provenance: Sequence[Tuple[str, str]], ) -> str: material = { "schema_version": SCHEMA_VERSION, "publication_policy_version": PUBLICATION_POLICY_VERSION, "mart_sha256": mart_digest, "models": [ {"model_version": version, "metrics_sha256": metrics_digest} for version, metrics_digest in sorted(model_provenance) ], } encoded = json.dumps( material, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def build_assets( mart: Path, mart_manifest_path: Path, bundles: Sequence[ModelBundle], development_preview: bool, ) -> Dict[str, object]: mart_manifest, development_source, mart_digest = validate_mart_manifest( mart, mart_manifest_path, development_preview ) population_allowed = bool( mart_manifest.get("population_estimate_allowed") ) and not development_preview and not development_source connection = duckdb.connect(":memory:") try: _create_safe_mart(connection, mart) diagnostics, model_provenance = load_model_diagnostics( bundles, mart_digest, _partition_vehicle_support(connection), ) overview = _overview_rows(connection) age_risk = _age_risk_rows(connection) scorecards = _scorecard_rows(connection) coverage = _coverage_rows(connection) finally: connection.close() if not overview: raise PublicSchemaError("No overview cells survive publication suppression") base = _base_asset(development_preview, population_allowed) assets: Dict[str, object] = { "overview_period_county.json": {**base, "rows": overview}, "age_risk_curve.json": {**base, "rows": age_risk}, "cohort_scorecard.json": {**base, "rows": scorecards}, "model_diagnostics.json": {**base, "rows": diagnostics}, "coverage_quality.json": {**base, "rows": coverage}, } counties = sorted({str(row["public_county"]) for row in overview}) periods_by_year: Dict[int, set] = {} for row in overview: periods_by_year.setdefault(int(row["year"]), set()).add(int(row["quarter"])) periods = [ {"year": year, "quarters": sorted(quarters)} for year, quarters in sorted(periods_by_year.items()) ] make_models = [ {"prior_make": row["prior_make"], "prior_model": row["prior_model"]} for row in scorecards ] model_names = sorted({str(row["model"]) for row in diagnostics}) assets["filter_catalog.json"] = { **base, "public_counties": counties, "age_bands": [str(row["age_band"]) for row in age_risk], "prior_make_models": make_models, "periods": periods, "models": model_names, "partitions": list(ALLOWED_PARTITIONS), } source_years = [int(row["year"]) for row in overview] model_versions = sorted({version for version, _ in model_provenance}) assets["data_manifest.json"] = { **base, "release_id": _release_id(mart_digest, model_provenance), "model_versions": model_versions, "data_scope": { "first_year": min(source_years), "last_year": max(source_years), "partitions": list(ALLOWED_PARTITIONS), "model_names": model_names, }, "definitions": { "target": "first-attempt next-episode binary non-pass rate", "episode_gap_days": int(mart_manifest.get("episode_gap_days", 30)), "support_rounding": SUPPORT_ROUNDING, "suppression_min_support": MIN_SUPPORT, "suppression_min_pass": MIN_CLASS_COUNT, "suppression_min_nonpass": MIN_CLASS_COUNT, "suppression_min_distinct_vehicles": MIN_SUPPORT, "suppression_min_distinct_pass_vehicles": MIN_CLASS_COUNT, "suppression_min_distinct_nonpass_vehicles": MIN_CLASS_COUNT, "locked_test_metrics_published": False, }, "assets": sorted( name for name in PUBLIC_ASSET_NAMES if name != "data_manifest.json" ), } for name, asset in assets.items(): validate_public_asset(name, asset) return assets def _validate_flags(asset: Mapping[str, object]) -> None: if asset.get("schema_version") != SCHEMA_VERSION: raise PublicSchemaError("Asset schema version is invalid") if not isinstance(asset.get("development_preview"), bool): raise PublicSchemaError("Preview flag must be boolean") if not isinstance(asset.get("population_estimate_allowed"), bool): raise PublicSchemaError("Population flag must be boolean") if asset["development_preview"] and asset["population_estimate_allowed"]: raise PublicSchemaError("Preview assets cannot permit population estimates") def _scan_public_value(value: object, path: str = "root") -> None: if isinstance(value, dict): for key, child in value.items(): normalized = key.lower() if normalized in DENIED_EXACT_KEYS or any( fragment in normalized for fragment in DENIED_KEY_FRAGMENTS ): raise PublicSchemaError("Denied public key at " + path) _scan_public_value(child, path + "." + key) elif isinstance(value, list): for index, child in enumerate(value): _scan_public_value(child, path + "[{}]".format(index)) elif isinstance(value, str): if EXACT_TIMESTAMP_PATTERN.match(value): raise PublicSchemaError("Exact timestamp in public asset") elif isinstance(value, float): if not math.isfinite(value): raise PublicSchemaError("Non-finite public number") elif value is not None and not isinstance(value, (bool, int)): raise PublicSchemaError("Unsupported public JSON value") def _validate_rate(value: object, label: str) -> None: if isinstance(value, bool) or not isinstance(value, (int, float)): raise PublicSchemaError(label + " must be numeric") number = float(value) if not math.isfinite(number) or not 0.0 <= number <= 1.0: raise PublicSchemaError(label + " is outside [0,1]") def validate_public_asset(name: str, asset: object) -> None: if not isinstance(asset, dict): raise PublicSchemaError("Public asset must be an object") _validate_flags(asset) _scan_public_value(asset) if name in ROW_ASSET_SCHEMAS: if set(asset) != { "schema_version", "development_preview", "population_estimate_allowed", "rows", }: raise PublicSchemaError("Row asset has unexpected top-level keys") rows = asset.get("rows") if not isinstance(rows, list): raise PublicSchemaError("Row asset rows must be an array") required = set(ROW_ASSET_SCHEMAS[name]) for row in rows: if not isinstance(row, dict) or set(row) != required: raise PublicSchemaError("Row does not match its public schema") if "support_rounded" in row: support = row["support_rounded"] if ( isinstance(support, bool) or not isinstance(support, int) or support < MIN_SUPPORT or support % SUPPORT_ROUNDING ): raise PublicSchemaError("Published support is not safely rounded") for key, value in row.items(): if key.endswith("_rate") or key.endswith("_share") or key in { "average_precision", "brier", "roc_auc", "top_10_capture", }: _validate_rate(value, key) if key == "log_loss": if ( isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)) or float(value) < 0.0 ): raise PublicSchemaError("Invalid public log loss") return if name == "filter_catalog.json": expected = { "schema_version", "development_preview", "population_estimate_allowed", "public_counties", "age_bands", "prior_make_models", "periods", "models", "partitions", } if set(asset) != expected: raise PublicSchemaError("Filter catalog schema is invalid") if asset["partitions"] != list(ALLOWED_PARTITIONS): raise PublicSchemaError("Filter catalog exposes an invalid partition") return if name == "data_manifest.json": expected = { "schema_version", "development_preview", "population_estimate_allowed", "release_id", "model_versions", "data_scope", "definitions", "assets", } if set(asset) != expected: raise PublicSchemaError("Data manifest schema is invalid") definitions = asset.get("definitions") if not isinstance(definitions, dict) or definitions.get( "locked_test_metrics_published" ) is not False: raise PublicSchemaError("Data manifest does not lock 2025 metrics") release_id = asset.get("release_id") if not isinstance(release_id, str) or not re.fullmatch( r"[0-9a-f]{64}", release_id ): raise PublicSchemaError("Data manifest release ID is invalid") model_versions = asset.get("model_versions") if ( not isinstance(model_versions, list) or not model_versions or model_versions != sorted(set(model_versions)) or any( not isinstance(version, str) or not SAFE_MODEL_NAME_PATTERN.fullmatch(version) for version in model_versions ) ): raise PublicSchemaError("Data manifest model versions are invalid") return raise PublicSchemaError("Unexpected public asset name") def _json_bytes(value: object) -> bytes: return ( json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False, ) + "\n" ).encode("utf-8") def _validate_output_dir(output_dir: Path) -> Path: resolved = output_dir.resolve() if not resolved.is_relative_to(PUBLIC_DATA_ROOT): raise PublicSchemaError( "Dashboard assets must remain under dashboard/public/data" ) return resolved def _validate_existing_output_contract( output_dir: Path, expected_names: frozenset, require_complete: bool = False, ) -> Dict[str, Path]: if not output_dir.exists(): return {} if not output_dir.is_dir(): raise PublicSchemaError("Dashboard data output is not a directory") entries = {entry.name: entry for entry in output_dir.iterdir()} unexpected = sorted(set(entries) - expected_names) invalid_types = sorted( name for name, entry in entries.items() if entry.is_symlink() or not entry.is_file() ) if unexpected or invalid_types: raise PublicSchemaError( "Dashboard data directory contains unexpected content" ) if require_complete and set(entries) != expected_names: raise PublicSchemaError( "Dashboard data directory does not match the exact public contract" ) return entries def publish_assets( assets: Mapping[str, object], output_dir: Path, overwrite: bool, ) -> Dict[str, Path]: output_dir = _validate_output_dir(output_dir) if set(assets) != set(PUBLIC_ASSET_NAMES): raise PublicSchemaError("Asset set does not match the public contract") serialized = {name: _json_bytes(assets[name]) for name in sorted(assets)} first_asset = next(iter(assets.values())) if not isinstance(first_asset, dict): raise PublicSchemaError("Invalid public asset") sha_manifest = { "schema_version": SCHEMA_VERSION, "development_preview": first_asset["development_preview"], "population_estimate_allowed": first_asset["population_estimate_allowed"], "files": [ { "name": name, "sha256": hashlib.sha256(serialized[name]).hexdigest(), } for name in sorted(serialized) ], } _validate_flags(sha_manifest) _scan_public_value(sha_manifest) serialized[SHA256_MANIFEST_NAME] = _json_bytes(sha_manifest) targets = {name: output_dir / name for name in serialized} if set(targets) != PUBLIC_OUTPUT_NAMES: raise PublicSchemaError("Output file set does not match the public contract") existing = _validate_existing_output_contract( output_dir, PUBLIC_OUTPUT_NAMES ) if not overwrite and existing: raise FileExistsError("Refusing to overwrite existing dashboard assets") os.umask(0o022) output_dir.parent.mkdir(parents=True, exist_ok=True) staging_root = Path( tempfile.mkdtemp(prefix=".dashboard-data-", dir=str(output_dir.parent)) ) try: for name, content in serialized.items(): staged = staging_root / name with staged.open("xb") as handle: handle.write(content) handle.flush() os.fsync(handle.fileno()) staged.chmod(0o644) output_dir.mkdir(parents=True, exist_ok=True) existing = _validate_existing_output_contract( output_dir, PUBLIC_OUTPUT_NAMES ) if not overwrite and existing: raise FileExistsError("Dashboard assets appeared during publication") # Publish the checksum manifest last; consumers can treat it as the # commit marker for the individually atomic file replacements. for name in sorted(serialized): if name == SHA256_MANIFEST_NAME: continue os.replace(staging_root / name, targets[name]) os.replace( staging_root / SHA256_MANIFEST_NAME, targets[SHA256_MANIFEST_NAME], ) _validate_existing_output_contract( output_dir, PUBLIC_OUTPUT_NAMES, require_complete=True ) finally: shutil.rmtree(staging_root, ignore_errors=True) return targets def export_dashboard_data( mart: Path, mart_manifest: Path, bundles: Sequence[ModelBundle], output_dir: Path, development_preview: bool = False, overwrite: bool = False, ) -> Dict[str, Path]: assets = build_assets( mart=mart, mart_manifest_path=mart_manifest, bundles=bundles, development_preview=development_preview, ) return publish_assets(assets, output_dir, overwrite) def main(argv: Optional[Sequence[str]] = None) -> int: args = parse_args(argv) if len(args.model_manifest) != len(args.model_metrics): print( "Configuration error: pair each model manifest with one metrics file", file=sys.stderr, ) return 2 bundles = [ ModelBundle(manifest=manifest, metrics=metrics) for manifest, metrics in zip(args.model_manifest, args.model_metrics) ] try: paths = export_dashboard_data( mart=args.mart, mart_manifest=_mart_manifest_path(args.mart, args.mart_manifest), bundles=bundles, output_dir=args.output_dir, development_preview=args.development_preview, overwrite=args.overwrite, ) except (DashboardExportError, FileExistsError) as exc: print("Dashboard export refused: " + str(exc), file=sys.stderr) return 2 print("Published {} sanitized dashboard assets".format(len(paths))) print("Public data directory: " + str(args.output_dir.resolve())) return 0 if __name__ == "__main__": raise SystemExit(main())