verify_v2_artifacts.py

#!/usr/bin/env python3
"""Verify that every v2 artifact is in place and self-consistent.

Checks performed
----------------
1. Top-level docs (LICENSE, NOTICE.md, DATA_CARD.md, CITATION.cff,
   SUMMARY_v2.md) exist and are non-empty.
2. Per-subset LICENSE files exist (Piraeus, Norway). DMA + NOAA are
   verified against their symlink-target LICENSE.
3. Historical OSM PBFs exist with the documented sizes.
4. stage_17 + publish artifacts exist for every subset × track (A + B).
5. The dual-subset row counts add up:  total = consistent + inconsistent (+ uncheckable).
6. The flag side-car CSV's header is the expected schema.
7. Piraeus historical-OSM context tree exists once the 2019 build is
   done (presence-only check).

Outputs a JSON report at v2_verification_report.json and prints a
human summary.
"""
from __future__ import annotations

import os
import csv
import gzip
import hashlib
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent.parent


SUBSETS_A = {
    "DMA":     Path(os.environ.get("DMA_ROOT", "data/dma")),
    "NOAA":    Path(os.environ.get("NOAA_ROOT", "data/noaa")),
    "Piraeus": ROOT / "Piraeus_ship_trajectory_datasets",
    "Norway":  ROOT / "norway_ship_trajectory_datasets",
}
SUBSETS_B = {ds: ROOT / "track_b" / ds for ds in ("DMA","NOAA","Piraeus","Norway")}

REQUIRED_FLAG_COLS = {
    "sample_id", "n_hist_points", "n_fut_points",
    "n_hist_inland", "n_fut_inland", "n_inland_points",
    "n_uncheckable_points", "max_inland_depth_m", "max_consec_inland_run",
    "osm_temporal_consistent",
}
REQUIRED_NEW_CSV_COLS = {
    "osm_temporal_consistent", "osm_max_inland_depth_m",
    "osm_n_inland_points", "osm_max_consec_inland_run",
}


def check_file(path: Path, min_size: int = 1) -> dict:
    if not path.exists():
        return {"path": str(path), "exists": False}
    sz = path.stat().st_size
    return {"path": str(path), "exists": True, "size": sz, "ok": sz >= min_size}


def check_pbf(path: Path, expected_size: int | None = None) -> dict:
    out = check_file(path, 1_000_000)
    if expected_size and out.get("size") != expected_size:
        out["size_mismatch"] = expected_size
    return out


def check_flag_csv(path: Path) -> dict:
    if not path.exists():
        return {"path": str(path), "exists": False}
    with path.open(newline="") as fh:
        r = csv.reader(fh)
        try:
            header = next(r)
        except StopIteration:
            return {"path": str(path), "exists": True, "empty": True}
        cols = set(header)
        missing = REQUIRED_FLAG_COLS - cols
        n_rows = sum(1 for _ in r)
    return {
        "path": str(path),
        "exists": True,
        "header_ok": not missing,
        "missing_cols": sorted(missing),
        "n_rows": n_rows,
    }


def check_pub_csv(path: Path) -> dict:
    if not path.exists():
        return {"path": str(path), "exists": False}
    with gzip.open(path, "rt") as fh:
        r = csv.reader(fh)
        try:
            header = next(r)
        except StopIteration:
            return {"path": str(path), "exists": True, "empty": True}
        cols = set(header)
        new_cols_present = REQUIRED_NEW_CSV_COLS <= cols
        n_rows = sum(1 for _ in r)
    return {
        "path": str(path),
        "exists": True,
        "has_flag_cols": new_cols_present,
        "n_rows": n_rows,
    }


def main():
    report = {"top_level_docs": {}, "licenses": {}, "pbfs": {}, "stage17": {}, "pub": {}}

    # 1. Top-level docs
    for f in ["LICENSE", "NOTICE.md", "DATA_CARD.md", "CITATION.cff", "SUMMARY_v2.md", "REPORT.md"]:
        report["top_level_docs"][f] = check_file(ROOT / f, 500)

    # 2. Per-subset LICENSE
    for ds, root in SUBSETS_A.items():
        report["licenses"][ds] = check_file(root / "LICENSE", 500)

    # 3. Historical PBFs
    report["pbfs"]["greece-200101"] = check_pbf(
        ROOT / "Piraeus_ship_trajectory_datasets/data_raw/historical_osm/greece-200101.osm.pbf",
        185245296)
    report["pbfs"]["denmark-260101"] = check_pbf(
        Path(os.environ.get("DMA_ROOT", "data/dma")) / "data_raw/dma/historical_osm/denmark-260101.osm.pbf",
        480395981)
    report["pbfs"]["norway-260101"] = check_pbf(
        ROOT / "norway_ship_trajectory_datasets/data_raw/historical_osm/norway-260101.osm.pbf",
        1357070732)

    # 4. Stage 17 flags
    for track, sset in (("A", SUBSETS_A), ("B", SUBSETS_B)):
        for ds, root in sset.items():
            key = f"{ds}_{track}"
            base = root / "multi_type_mini_bench_build/standard_track_v1"
            ftr = base / "osm_temporal_consistency"
            report["stage17"][key] = {
                "train": check_flag_csv(ftr / "train_flags.csv"),
                "val":   check_flag_csv(ftr / "val_flags.csv"),
                "test":  check_flag_csv(ftr / "test_flags.csv"),
                "summary": check_file(ftr / "summary.json", 100),
            }

    # 4b. Piraeus 2019 OSM stage 17
    ftr_19 = (SUBSETS_A["Piraeus"] / "multi_type_mini_bench_build/standard_track_v1"
              / "osm_temporal_consistency_2019osm")
    report["stage17"]["Piraeus_2019osm"] = {
        "train": check_flag_csv(ftr_19 / "train_flags.csv"),
        "val":   check_flag_csv(ftr_19 / "val_flags.csv"),
        "test":  check_flag_csv(ftr_19 / "test_flags.csv"),
        "summary": check_file(ftr_19 / "summary.json", 100),
        "context_dir": check_file(SUBSETS_A["Piraeus"] / "multi_type_mini_bench_build/standard_track_v1/context_v1_2019osm/summary.json", 100),
    }

    # 5. Pub CSVs
    for track, sset in (("A", SUBSETS_A), ("B", SUBSETS_B)):
        for ds, root in sset.items():
            key = f"{ds}_{track}"
            base = root / "multi_type_mini_bench_build"
            for sub in ("standard_track_v1_all", "standard_track_v1_filtered"):
                for split in ("train", "val", "test"):
                    p = base / sub / split / "part-000.csv.gz"
                    report["pub"].setdefault(key, {})[f"{sub}/{split}"] = check_pub_csv(p)

    # Sanity check rollup
    issues = []
    def walk(d, prefix=""):
        for k, v in d.items():
            if isinstance(v, dict):
                if v.get("exists") is False:
                    issues.append(f"{prefix}{k}: MISSING ({v['path']})")
                elif v.get("ok") is False:
                    issues.append(f"{prefix}{k}: too small ({v['path']})")
                elif v.get("header_ok") is False:
                    issues.append(f"{prefix}{k}: missing_cols={v.get('missing_cols')}")
                elif v.get("has_flag_cols") is False:
                    issues.append(f"{prefix}{k}: missing flag cols ({v['path']})")
                if not all(kk in ("exists","size","ok","path","size_mismatch","header_ok","missing_cols","n_rows","empty","has_flag_cols","context_dir","summary","train","val","test")
                           for kk in v):
                    walk(v, prefix + k + "/")
            elif isinstance(v, dict) is False:
                pass
    walk(report)

    summary = {
        "n_total_checks": sum(1 for k, sd in report.items() for _ in (sd.values() if isinstance(sd, dict) else [])),
        "n_issues": len(issues),
        "issues":   issues[:30],
    }
    out = {"summary": summary, "report": report}
    (ROOT / "v2_verification_report.json").write_text(json.dumps(out, indent=2))
    print(f"[verify] checks={summary['n_total_checks']}  issues={summary['n_issues']}")
    if issues:
        print(f"[verify] first issues:")
        for i in issues[:10]:
            print(f"  - {i}")
    else:
        print(f"[verify] OK — all v2 artifacts present and well-formed")
    sys.exit(0 if not issues else 1)


if __name__ == "__main__":
    main()