publish_dual_subset.py

#!/usr/bin/env python3
"""DEPRECATED in v2 final (2026-06-17).
Publish the dual (`_all`, `_filtered`) standard_track subsets after
running stage 17.

The v2 final HF release ships a SINGLE directory per (subset, track)
with the 4 OSM-temporal-consistency flag columns merged inline into
the main CSVs. The `_all` and `_filtered` sibling directories this
script produces are no longer part of the released layout. Users can
recover them with one line of pandas:

    df = pd.read_csv("<subset>_track_v1/<split>/part-000.csv.gz")
    df_filtered = df[df["osm_temporal_consistent"] == "true"]

This file is preserved for historical reproducibility of intermediate
v2 commits and for studies that prefer the pre-merge layout. Do NOT
run it as part of the canonical reproduction pipeline.

----------------------------------------------------------------------
Original docstring:

`standard_track_v1_all/` is identical to `standard_track_v1/` except
that every CSV row gains four extra columns merged from the stage-17
flag side-car:

  osm_temporal_consistent       — "true" | "false" | "" (uncheckable)
  osm_max_inland_depth_m        — float
  osm_n_inland_points           — int
  osm_max_consec_inland_run     — int

`standard_track_v1_filtered/` is the subset of `_all` where
`osm_temporal_consistent == "true"`. Uncheckable rows are *excluded*
from the filtered subset to keep training tables tidy (the rare
uncheckable case is logged in the side-car summary). This subset is
the recommended default for paper main tables.

Both subsets reuse the rasters/SDFs/social context by reference — no
duplication of heavy artefacts. They are emitted as new gzipped CSVs
in the standard `train/val/test/part-000.csv.gz` layout so existing
loaders work unchanged.

Usage
-----
python publish_dual_subset.py \
    --track-root <standard_track_v1>            \
    --flags-root <osm_temporal_consistency>     \
    [--splits train val test]
"""
from __future__ import annotations

import argparse
import csv
import gzip
import json
from pathlib import Path


def load_flags(flags_csv: Path) -> dict[str, dict]:
    out = {}
    with flags_csv.open(newline="") as fh:
        r = csv.DictReader(fh)
        for row in r:
            out[row["sample_id"]] = row
    return out


def publish_split(track_root: Path, flags_root: Path, split: str) -> dict:
    src = track_root / split / "part-000.csv.gz"
    flag_csv = flags_root / f"{split}_flags.csv"
    if not src.exists():
        return {"split": split, "skipped": True, "reason": "no src CSV"}
    if not flag_csv.exists():
        return {"split": split, "skipped": True, "reason": "no flag CSV"}

    all_dir = track_root.parent / (track_root.name + "_all") / split
    flt_dir = track_root.parent / (track_root.name + "_filtered") / split
    all_dir.mkdir(parents=True, exist_ok=True)
    flt_dir.mkdir(parents=True, exist_ok=True)

    flags = load_flags(flag_csv)

    n_total = 0
    n_consistent = 0
    n_inconsistent = 0
    n_uncheckable = 0

    new_cols = [
        "osm_temporal_consistent",
        "osm_max_inland_depth_m",
        "osm_n_inland_points",
        "osm_max_consec_inland_run",
    ]

    with gzip.open(src, "rt", encoding="utf-8", newline="") as fh:
        reader = csv.DictReader(fh)
        base_fields = reader.fieldnames or []
        out_fields = base_fields + new_cols

        with gzip.open(all_dir / "part-000.csv.gz", "wt", encoding="utf-8", newline="") as a_fh, \
             gzip.open(flt_dir / "part-000.csv.gz", "wt", encoding="utf-8", newline="") as f_fh:
            a_w = csv.DictWriter(a_fh, fieldnames=out_fields)
            f_w = csv.DictWriter(f_fh, fieldnames=out_fields)
            a_w.writeheader()
            f_w.writeheader()
            for row in reader:
                n_total += 1
                sid = row["sample_id"]
                f = flags.get(sid)
                if f is None:
                    consistent_str = ""
                    n_uncheckable += 1
                    row_with_flags = dict(row)
                    row_with_flags.update({
                        "osm_temporal_consistent":   "",
                        "osm_max_inland_depth_m":    "",
                        "osm_n_inland_points":       "",
                        "osm_max_consec_inland_run": "",
                    })
                    a_w.writerow(row_with_flags)
                    continue
                consistent_str = f["osm_temporal_consistent"]
                if consistent_str == "true":
                    n_consistent += 1
                elif consistent_str == "false":
                    n_inconsistent += 1
                else:
                    n_uncheckable += 1
                row_with_flags = dict(row)
                row_with_flags.update({
                    "osm_temporal_consistent":   consistent_str,
                    "osm_max_inland_depth_m":    f.get("max_inland_depth_m", ""),
                    "osm_n_inland_points":       f.get("n_inland_points", ""),
                    "osm_max_consec_inland_run": f.get("max_consec_inland_run", ""),
                })
                a_w.writerow(row_with_flags)
                if consistent_str == "true":
                    f_w.writerow(row_with_flags)

    return {
        "split":         split,
        "total":         n_total,
        "consistent":    n_consistent,
        "inconsistent":  n_inconsistent,
        "uncheckable":   n_uncheckable,
        "all_path":      str(all_dir / "part-000.csv.gz"),
        "filtered_path": str(flt_dir / "part-000.csv.gz"),
    }


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--track-root", type=Path, required=True,
                   help="standard_track_v1 root")
    p.add_argument("--flags-root", type=Path, required=True,
                   help="osm_temporal_consistency root containing <split>_flags.csv")
    p.add_argument("--splits", nargs="+", default=["train", "val", "test"])
    args = p.parse_args()

    print(f"[pub] track={args.track_root}")
    print(f"[pub] flags={args.flags_root}")
    all_stats = {}
    for split in args.splits:
        s = publish_split(args.track_root, args.flags_root, split)
        all_stats[split] = s
        print(f"[pub]   {s}")
    out = {
        "track_root": str(args.track_root),
        "flags_root": str(args.flags_root),
        "splits":     all_stats,
    }
    summary_path = args.track_root.parent / "standard_track_v1_filtered.summary.json"
    summary_path.write_text(json.dumps(out, indent=2))
    print(f"[pub] DONE — summary → {summary_path}")


if __name__ == "__main__":
    main()