Spaces:
Running on Zero
Running on Zero
| """Phase 2 — Member 4 data preparation (deterministic, read-only on originals). | |
| Produces lightweight processed artifacts for Phase 3+: | |
| data/processed/near_miss/ | |
| dtc_fm_combined.csv positional-merged features+labels with sample_index | |
| duplicate_report.csv feature-row duplicate groups + label agreement | |
| label_mapping.json neutral class_0/class_1 (semantics unconfirmed) | |
| feature_policy.json keep/flag decision + reason per feature | |
| data/processed/accident/ | |
| train_index.parquet one row per image (split, source, sha256, dims, | |
| validation_index.parquet annotations) — NO image bytes copied | |
| test_index.parquet | |
| duplicate_report.csv train exact-byte duplicate groups w/ annotation diff | |
| split_integrity_report.json | |
| label_mapping.json | |
| corruption_report.json (empty list when all sampled images decode) | |
| Originals under dataset/ are never modified. No randomness is used anywhere. | |
| """ | |
| import hashlib | |
| import json | |
| import struct | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import pyarrow as pa | |
| import pyarrow.parquet as pq | |
| DATASET_DIR = Path("dataset") | |
| OUT_NM = Path("data/processed/near_miss") | |
| OUT_ACC = Path("data/processed/accident") | |
| REPORT_DIR = Path("reports") | |
| FEATURES_CSV = DATASET_DIR / "filtered_features_DTC_FM.csv" | |
| LABELS_CSV = DATASET_DIR / "target_labels_DTC_FM.csv" | |
| ACCIDENT_SPLITS = { | |
| "train": [ | |
| "train-00000-of-00002.parquet", | |
| "train-00001-of-00002.parquet", | |
| ], | |
| "validation": ["validation-00000-of-00001.parquet"], | |
| "test": ["test-00000-of-00001.parquet"], | |
| } | |
| EXPECTED_DTC_ROWS = 2508 | |
| DECODE_SAMPLE_STRIDE = 100 # decode every Nth image as a representative sample | |
| # -------------------------------------------------------------------------- | |
| # Near-miss (DTC-FM) | |
| # -------------------------------------------------------------------------- | |
| def build_near_miss() -> None: | |
| feats = pd.read_csv(FEATURES_CSV) | |
| labels = pd.read_csv(LABELS_CSV) | |
| # --- alignment verification (positional; no sorting, no pre-drops) --- | |
| assert len(feats) == EXPECTED_DTC_ROWS, f"features rows {len(feats)} != {EXPECTED_DTC_ROWS}" | |
| assert len(labels) == EXPECTED_DTC_ROWS, f"label rows {len(labels)} != {EXPECTED_DTC_ROWS}" | |
| assert list(labels.columns) == ["event_type"] | |
| assert labels["event_type"].isin([0.0, 1.0]).all(), "unexpected label values" | |
| combined = feats.copy() | |
| combined.insert(0, "sample_index", np.arange(len(combined), dtype=np.int64)) | |
| combined["event_type"] = labels["event_type"].astype(np.int64).values | |
| # --- duplicate analysis (report only; nothing dropped) --- | |
| dup_mask = combined.duplicated( | |
| subset=[c for c in feats.columns], keep=False | |
| ) | |
| groups = combined.loc[dup_mask, feats.columns.tolist()].apply( | |
| lambda r: "|".join(map(str, r)), axis=1 | |
| ) | |
| group_ids = pd.factorize(groups)[0] | |
| report = pd.DataFrame({ | |
| "duplicate_group": group_ids, | |
| "sample_index": combined.loc[dup_mask, "sample_index"].values, | |
| "event_type": combined.loc[dup_mask, "event_type"].values, | |
| }).sort_values(["duplicate_group", "sample_index"]) | |
| label_conflicts = ( | |
| report.groupby("duplicate_group")["event_type"].nunique() > 1 | |
| ) | |
| report.attrs["n_groups"] = int(report["duplicate_group"].nunique()) | |
| report.attrs["n_conflicting_groups"] = int(label_conflicts.sum()) | |
| report.to_csv(OUT_NM / "duplicate_report.csv", index=False) | |
| # --- cleaned combined dataset (original order preserved) --- | |
| combined.to_csv(OUT_NM / "dtc_fm_combined.csv", index=False) | |
| # exact-float companion (CSV round-trip can shift the last ULP of ~23 values) | |
| combined.to_parquet(OUT_NM / "dtc_fm_combined.parquet", index=False) | |
| write_near_miss_label_mapping() | |
| write_feature_policy() | |
| return report | |
| def write_near_miss_label_mapping() -> None: | |
| payload = { | |
| "dataset": "DTC-FM", | |
| "target_column": "event_type", | |
| "mapping_status": "unconfirmed", | |
| "class_0": "class_0 (semantics unconfirmed; evidence from class-conditional " | |
| "statistics is consistent with a 'non-conflict / normal' class, " | |
| "NOT confirmed by source documentation)", | |
| "class_1": "class_1 (semantics unconfirmed; evidence from class-conditional " | |
| "statistics is consistent with a 'conflict / near-miss' class, " | |
| "NOT confirmed by source documentation)", | |
| "source": "No label map present in the CSV files. Web documentation search " | |
| "performed in Phase 3 (2026-09-02): no public dataset, paper or " | |
| "documentation named 'DTC-FM' could be located; no source " | |
| "documentation available locally. Neutral class_0/class_1 " | |
| "terminology MUST be used until the mapping is confirmed.", | |
| "class_distribution": { | |
| "class_0": {"count": 1867, "pct": 74.44}, | |
| "class_1": {"count": 641, "pct": 25.56}, | |
| }, | |
| "note": "sample_index in dtc_fm_combined.csv is an internal processing " | |
| "index (row position 0..2507), NOT a real dataset identifier.", | |
| } | |
| (OUT_NM / "label_mapping.json").write_text(json.dumps(payload, indent=2)) | |
| def write_feature_policy() -> None: | |
| policy = { | |
| "policy_version": "phase2", | |
| "decisions": [ | |
| { | |
| "feature": "speed_object_1_kph", "type": "original", | |
| "decision": "keep", "reason": "base physical measurement (kph)", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "speed_object_2_kph", "type": "original", | |
| "decision": "keep", "reason": "base physical measurement (kph)", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "angle_degrees", "type": "original", | |
| "decision": "keep", "reason": "encounter geometry, computable pre-conflict", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "PET", "type": "original", | |
| "decision": "keep_with_flag", | |
| "reason": "core surrogate-safety measure; retained for offline " | |
| "modeling but flagged: post-encroachment time can only be " | |
| "measured AFTER the interaction completes", | |
| "leakage_risk": "none_statistical", | |
| "deployment_availability": "POST_EVENT_ONLY — not available in a " | |
| "real-time system at decision time; " | |
| "deployment/design issue, not target leakage", | |
| }, | |
| { | |
| "feature": "acceleration_obj1_mps2", "type": "original", | |
| "decision": "keep", "reason": "base kinematic measurement", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "acceleration_obj2_mps2", "type": "original", | |
| "decision": "keep", "reason": "base kinematic measurement", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "min_dist_dual_check_m", "type": "original", | |
| "decision": "keep", "reason": "minimum separation distance (m)", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "object_count_at_frame1", "type": "original", | |
| "decision": "keep", "reason": "scene context", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "target_dist_px", "type": "original", | |
| "decision": "keep_with_flag", | |
| "reason": "raw pixel distance; camera-geometry dependent", | |
| "leakage_risk": "none", | |
| "deployment_availability": "camera_dependent — may not transfer " | |
| "across camera geometries", | |
| }, | |
| { | |
| "feature": "class_object_1", "type": "original_categorical", | |
| "decision": "keep", "reason": "object type; requires encoding in Phase 3", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "class_object_2", "type": "original_categorical", | |
| "decision": "keep", "reason": "object type; requires encoding in Phase 3", | |
| "leakage_risk": "none", "deployment_availability": "available_live", | |
| }, | |
| { | |
| "feature": "FE_inv_PET", "type": "engineered_verified", | |
| "formula": "1 / PET", | |
| "decision": "retain_temporarily_flag_redundant", | |
| "reason": "pure mathematical transform, verified; exact duplicate of " | |
| "PET information (r=0.92) — statistical redundancy, NOT leakage", | |
| "leakage_risk": "none_statistical_redundancy_only", | |
| "deployment_availability": "inherits PET post-event limitation", | |
| }, | |
| { | |
| "feature": "FE_dist_squared", "type": "engineered_verified", | |
| "formula": "target_dist_px ** 2", | |
| "decision": "retain_temporarily_flag_redundant", | |
| "reason": "verified pure transform; heavy-tailed scale instability; " | |
| "statistical redundancy, NOT leakage", | |
| "leakage_risk": "none_statistical_redundancy_only", | |
| "deployment_availability": "camera_dependent", | |
| }, | |
| { | |
| "feature": "FE_log_mesafe", "type": "engineered_verified", | |
| "formula": "ln(1 + target_dist_px)", | |
| "decision": "retain_temporarily_flag_redundant", | |
| "reason": "verified pure transform; monotone copy of target_dist_px; " | |
| "statistical redundancy, NOT leakage", | |
| "leakage_risk": "none_statistical_redundancy_only", | |
| "deployment_availability": "camera_dependent", | |
| }, | |
| { | |
| "feature": "FE_safety_index", "type": "engineered_verified_phase3", | |
| "formula": "(speed_object_1_kph + speed_object_2_kph) / " | |
| "(2 * max(target_dist_px, 1e-06))", | |
| "decision": "keep_with_flag", | |
| "reason": "PROVENANCE ESTABLISHED in Phase 3: exact deterministic " | |
| "transform of speed_object_1_kph, speed_object_2_kph and " | |
| "target_dist_px (verified to 3 decimals on all 2,457 " | |
| "dist>0 rows; the 51 dist==0 rows use a 1e-6 floor, " | |
| "producing quantized extremes up to 2.85e7). It is an " | |
| "inverse-pixel-distance x combined-speed composite with " | |
| "inconsistent units (kph/px) — redundant with existing " | |
| "features and numerically unstable, but NOT target leakage", | |
| "leakage_risk": "none (pure input-feature transform); " | |
| "redundancy + instability only", | |
| "deployment_availability": "available_live (speeds + pixel distance), " | |
| "camera_dependent", | |
| "action": "Phase 3 trains with/without it (configs A/B) to measure " | |
| "its effect; no test-based keep/drop decision", | |
| }, | |
| ], | |
| "principles": [ | |
| "derived != leakage: transforms are flagged for redundancy, not accused of leakage", | |
| "PET concern is deployment availability (post-event), not statistical leakage", | |
| "no feature deleted; decisions are advisory for Phase 3", | |
| ], | |
| } | |
| (OUT_NM / "feature_policy.json").write_text(json.dumps(policy, indent=2)) | |
| # -------------------------------------------------------------------------- | |
| # Accident (Parquet) | |
| # -------------------------------------------------------------------------- | |
| def png_ihdr(data: bytes) -> tuple[str, int, int] | None: | |
| """Return (format, width, height) from a PNG header, or None if invalid.""" | |
| if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n": | |
| return None | |
| w, h = struct.unpack(">II", data[16:24]) | |
| return "PNG", w, h | |
| def scan_accident_shards() -> dict[str, pd.DataFrame]: | |
| """Stream every shard once: hash, dims, annotations -> per-split index frames.""" | |
| indexes: dict[str, list[dict]] = {s: [] for s in ACCIDENT_SPLITS} | |
| for split, files in ACCIDENT_SPLITS.items(): | |
| for fname in files: | |
| pf = pq.ParquetFile(DATASET_DIR / fname) | |
| local_row = 0 | |
| for rg in range(pf.metadata.num_row_groups): | |
| tbl = pf.read_row_group(rg) | |
| images = tbl.column("image").to_pylist() | |
| objects = tbl.column("objects").to_pylist() | |
| for row_in_rg, (img, obj) in enumerate(zip(images, objects)): | |
| data = (img or {}).get("bytes") or b"" | |
| fmt, w, h = png_ihdr(data) if data else ("EMPTY", 0, 0) | |
| cats = list((obj or {}).get("category") or []) | |
| boxes = [list(map(float, b)) for b in (obj or {}).get("bbox") or []] | |
| indexes[split].append({ | |
| "split": split, | |
| "source_file": fname, | |
| "row_index": local_row, | |
| # row_index is 0-based WITHIN source_file (internal | |
| # reference = source_file + row_index; not a dataset ID) | |
| "image_sha256": hashlib.sha256(data).hexdigest() if data else "", | |
| "image_format": fmt, | |
| "width": w, | |
| "height": h, | |
| "n_objects": len(cats), | |
| "categories": cats, | |
| "bbox": boxes, | |
| }) | |
| local_row += 1 | |
| out = {} | |
| for split, rows in indexes.items(): | |
| schema = pa.schema([ | |
| ("split", pa.string()), ("source_file", pa.string()), | |
| ("row_index", pa.int64()), ("image_sha256", pa.string()), | |
| ("image_format", pa.string()), ("width", pa.int32()), | |
| ("height", pa.int32()), ("n_objects", pa.int32()), | |
| ("categories", pa.list_(pa.int64())), | |
| ("bbox", pa.list_(pa.list_(pa.float32(), 4))), | |
| ]) | |
| tbl = pa.Table.from_pylist(rows, schema=schema) | |
| pq.write_table(tbl, OUT_ACC / f"{split}_index.parquet") | |
| out[split] = tbl.to_pandas() | |
| return out | |
| def accident_label_mapping() -> None: | |
| payload = { | |
| "dataset": "traffic-accident-cctv-object-detection", | |
| "task": "accident + object detection", | |
| "label_column": "objects.category", | |
| "mapping": {"0": "accident", "1": "non_accident"}, | |
| "source": "dataset documentation (Hugging Face dataset card README, " | |
| "justjuu/traffic-accident-cctv-object-detection); class map is " | |
| "NOT embedded in the parquet files themselves", | |
| "bbox_format": "COCO-style [x, y, w, h] in absolute pixels on 640x640 images", | |
| } | |
| (OUT_ACC / "label_mapping.json").write_text(json.dumps(payload, indent=2)) | |
| def accident_duplicate_report(idx: dict[str, pd.DataFrame]) -> pd.DataFrame: | |
| """Group train images by exact-byte hash; compare labels+annotations.""" | |
| rows = [] | |
| train = idx["train"] | |
| for h, grp in train.groupby("image_sha256"): | |
| if len(grp) < 2: | |
| continue | |
| ref_cats = list(grp["categories"].iloc[0]) | |
| ref_bbox = [list(b) for b in grp["bbox"].iloc[0]] | |
| cats_identical = all(list(c) == ref_cats for c in grp["categories"]) | |
| bbox_identical = all( | |
| [list(b) for b in boxes] == ref_bbox for boxes in grp["bbox"] | |
| ) | |
| for _, r in grp.iterrows(): | |
| rows.append({ | |
| "image_sha256": h, | |
| "source_file": r["source_file"], | |
| "row_index": r["row_index"], | |
| "categories": json.dumps([int(c) for c in r["categories"]]), | |
| "annotations_identical": cats_identical and bbox_identical, | |
| "label_conflict": not cats_identical, | |
| }) | |
| rep = pd.DataFrame(rows).sort_values(["image_sha256", "row_index"]) | |
| rep.to_csv(OUT_ACC / "duplicate_report.csv", index=False) | |
| return rep | |
| def split_integrity_report(idx: dict[str, pd.DataFrame]) -> dict: | |
| sets = {s: set(idx[s]["image_sha256"]) for s in idx} | |
| tv = sets["train"] & sets["validation"] | |
| tt = sets["train"] & sets["test"] | |
| vt = sets["validation"] & sets["test"] | |
| n_cross = len(tv) + len(tt) + len(vt) | |
| payload = { | |
| "method": "SHA-256 of embedded PNG bytes (content hash, not filename)", | |
| "train_images": len(idx["train"]), | |
| "validation_images": len(idx["validation"]), | |
| "test_images": len(idx["test"]), | |
| "unique_images": len(sets["train"] | sets["validation"] | sets["test"]), | |
| "cross_split_duplicates": n_cross, | |
| "detail": { | |
| "train_and_validation": len(tv), | |
| "train_and_test": len(tt), | |
| "validation_and_test": len(vt), | |
| }, | |
| "result": "PASS" if n_cross == 0 else "FAIL", | |
| } | |
| (OUT_ACC / "split_integrity_report.json").write_text(json.dumps(payload, indent=2)) | |
| return payload | |
| def fetch_image_bytes(source_file: str, row_index: int) -> bytes: | |
| """Resolve a split-global row_index to its row group and fetch image bytes.""" | |
| pf = pq.ParquetFile(DATASET_DIR / source_file) | |
| offset = 0 | |
| for rg in range(pf.metadata.num_row_groups): | |
| n = pf.metadata.row_group(rg).num_rows | |
| if offset + n > row_index: | |
| tbl = pf.read_row_group(rg, columns=["image"]) | |
| return tbl.column("image")[row_index - offset].as_py()["bytes"] | |
| offset += n | |
| raise IndexError(f"row_index {row_index} beyond {source_file}") | |
| def validate_sample_images(idx: dict[str, pd.DataFrame]) -> list[dict]: | |
| """Decode a representative sample; verify dims + bounds on the full index.""" | |
| corrupt: list[dict] = [] | |
| try: | |
| from PIL import Image | |
| have_pil = True | |
| except ImportError: | |
| have_pil = False | |
| print("WARNING: Pillow unavailable — decode check skipped (IHDR checks only)") | |
| for split, df in idx.items(): | |
| bad_dims = df[(df["width"] != 640) | (df["height"] != 640)] | |
| for _, r in bad_dims.iterrows(): | |
| corrupt.append({"source_file": r["source_file"], "row_index": int(r["row_index"]), | |
| "issue": f"unexpected dims {r['width']}x{r['height']}"}) | |
| empty = df[df["image_format"] == "EMPTY"] | |
| for _, r in empty.iterrows(): | |
| corrupt.append({"source_file": r["source_file"], "row_index": int(r["row_index"]), | |
| "issue": "empty image bytes"}) | |
| for _, r in df.iloc[::DECODE_SAMPLE_STRIDE].iterrows(): | |
| data = fetch_image_bytes(r["source_file"], int(r["row_index"])) | |
| if have_pil: | |
| try: | |
| im = Image.open(__import__("io").BytesIO(data)) | |
| im.load() | |
| if im.size != (r["width"], r["height"]): | |
| corrupt.append({"source_file": r["source_file"], | |
| "row_index": int(r["row_index"]), | |
| "issue": "decoded dims != IHDR dims"}) | |
| except Exception as exc: # noqa: BLE001 - report any decode failure | |
| corrupt.append({"source_file": r["source_file"], | |
| "row_index": int(r["row_index"]), | |
| "issue": f"decode failed: {exc}"}) | |
| # bbox bounds check on the full index (annotation references valid image) | |
| for split, df in idx.items(): | |
| for i, boxes in enumerate(df["bbox"]): | |
| for b in boxes: | |
| x, y, w, h = b | |
| if x < 0 or y < 0 or x + w > 640.5 or y + h > 640.5: | |
| corrupt.append({"source_file": df["source_file"].iloc[i], | |
| "row_index": int(df["row_index"].iloc[i]), | |
| "issue": f"bbox out of bounds: {b}"}) | |
| (OUT_ACC / "corruption_report.json").write_text( | |
| json.dumps({"sampled_stride": DECODE_SAMPLE_STRIDE, | |
| "issues_found": len(corrupt), "corrupt_records": corrupt}, indent=2)) | |
| return corrupt | |
| def class_balance(idx: dict[str, pd.DataFrame]) -> dict: | |
| labels = pd.read_csv(LABELS_CSV)["event_type"].value_counts().sort_index() | |
| payload = { | |
| "near_miss_DTC_FM": { | |
| "class_0": {"count": int(labels[0.0]), "pct": round(labels[0.0] / 2508 * 100, 2)}, | |
| "class_1": {"count": int(labels[1.0]), "pct": round(labels[1.0] / 2508 * 100, 2)}, | |
| "verdict": "MODERATELY_IMBALANCED", | |
| }, | |
| "accident_object_level": {}, | |
| "accident_image_level": {}, | |
| } | |
| for split, df in idx.items(): | |
| all_cats = [c for cats in df["categories"] for c in cats] | |
| n0 = sum(1 for c in all_cats if c == 0) | |
| n1 = sum(1 for c in all_cats if c == 1) | |
| n_img_acc = sum(1 for cats in df["categories"] if 0 in cats) | |
| payload["accident_object_level"][split] = { | |
| "accident": n0, "non_accident": n1, "total": n0 + n1, | |
| "accident_pct": round(n0 / max(n0 + n1, 1) * 100, 2), | |
| } | |
| payload["accident_image_level"][split] = { | |
| "images": len(df), | |
| "images_with_accident_object": n_img_acc, | |
| "pct": round(n_img_acc / len(df) * 100, 2), | |
| } | |
| return payload | |
| def main() -> None: | |
| OUT_NM.mkdir(parents=True, exist_ok=True) | |
| OUT_ACC.mkdir(parents=True, exist_ok=True) | |
| REPORT_DIR.mkdir(exist_ok=True) | |
| dup_rep = build_near_miss() | |
| print(f"near-miss: combined dataset written; duplicate groups=" | |
| f"{dup_rep.attrs['n_groups']}, conflicting-label groups=" | |
| f"{dup_rep.attrs['n_conflicting_groups']}") | |
| idx = scan_accident_shards() | |
| accident_label_mapping() | |
| acc_dup = accident_duplicate_report(idx) | |
| integrity = split_integrity_report(idx) | |
| corrupt = validate_sample_images(idx) | |
| balance = class_balance(idx) | |
| print(f"accident: index rows per split = " | |
| f"{ {s: len(d) for s, d in idx.items()} }") | |
| print(f"accident: train duplicate images = " | |
| f"{acc_dup['image_sha256'].nunique()}, conflicting annotations = " | |
| f"{int((~acc_dup['annotations_identical']).groupby(acc_dup['image_sha256']).any().sum())}") | |
| print(f"accident: split integrity = {integrity['result']}") | |
| print(f"accident: corrupt/invalid records found = {len(corrupt)}") | |
| (REPORT_DIR / "class_balance_phase2.json").write_text(json.dumps(balance, indent=2)) | |
| print("class balance written to reports/class_balance_phase2.json") | |
| # ---------------- post-preparation validation ---------------- | |
| combined = pd.read_csv(OUT_NM / "dtc_fm_combined.csv") | |
| combined_exact = pd.read_parquet(OUT_NM / "dtc_fm_combined.parquet") | |
| orig_f = pd.read_csv(FEATURES_CSV) | |
| orig_l = pd.read_csv(LABELS_CSV)["event_type"].astype(np.int64).values | |
| num_cols = orig_f.select_dtypes(include=[np.number]).columns | |
| def csv_roundtrip_ok() -> bool: | |
| if not combined[orig_f.columns].dtypes.equals(orig_f.dtypes): | |
| if not combined[orig_f.select_dtypes(exclude=[np.number]).columns].equals( | |
| orig_f[orig_f.select_dtypes(exclude=[np.number]).columns]): | |
| return False | |
| return bool(np.allclose( | |
| combined[num_cols].to_numpy(dtype=float), | |
| orig_f[num_cols].to_numpy(dtype=float), | |
| rtol=0, atol=1e-9, equal_nan=True)) | |
| checks = { | |
| "near_miss_rows": len(combined) == EXPECTED_DTC_ROWS, | |
| "near_miss_features_exact_in_parquet": | |
| combined_exact[orig_f.columns].equals(orig_f), | |
| "near_miss_feature_values_unchanged_csv_tol1e-9": csv_roundtrip_ok(), | |
| "near_miss_labels_unchanged": (combined["event_type"].values == orig_l).all(), | |
| "near_miss_sample_index_ok": | |
| (combined["sample_index"].values == np.arange(EXPECTED_DTC_ROWS)).all(), | |
| "near_miss_no_new_missing": int(combined.isna().sum().sum()) == 0, | |
| } | |
| for split, df in idx.items(): | |
| pq_rows = pq.ParquetFile(OUT_ACC / f"{split}_index.parquet").metadata.num_rows | |
| checks[f"{split}_index_count_matches_scan"] = pq_rows == len(df) | |
| checks["accident_total_images"] = sum(len(d) for d in idx.values()) == 2763 | |
| checks["accident_no_cross_split_contamination"] = integrity["result"] == "PASS" | |
| print("\nvalidation:") | |
| for k, v in checks.items(): | |
| print(f" {'PASS' if v else 'FAIL'}: {k}") | |
| if not all(checks.values()): | |
| raise SystemExit("VALIDATION FAILED — inspect outputs before proceeding") | |
| if __name__ == "__main__": | |
| main() |