1206 lines
42 KiB
Python
1206 lines
42 KiB
Python
"""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"
|
|
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 = {
|
|
"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 = (
|
|
"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(
|
|
"""
|
|
SELECT last_observed_make, last_observed_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 safe_mart
|
|
WHERE eligible_returning_target
|
|
AND target_nonpass IN (0, 1)
|
|
AND last_observed_make IS NOT NULL
|
|
AND last_observed_model IS NOT NULL
|
|
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, last_observed_make, last_observed_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,
|
|
"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())
|