code/validation/src/verify_release.py

#!/usr/bin/env python3
"""Verify CI-Net release structure, metadata, catalog, weights, and privacy checks."""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
import subprocess
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import torch
import yaml
from safetensors.torch import load_file


EXPECTED_ROWS = {
    "concat": 26210,
    "hsr": 26007,
    "ci_hard": 5492,
    "hsr_valid_mask": 26007,
}
TEXT_SUFFIXES = {".py", ".yaml", ".yml", ".json", ".md", ".sh", ".txt", ".csv", ".log", ".gitignore"}
FORBIDDEN = {
    "internal mount path": re.compile("/" + "mnt" + "/"),
    "internal share path": re.compile("/" + "share" + "/"),
    "legacy executable prefix": re.compile("GK2A" + "_CI_"),
    "personal identifier": re.compile("(?i)(si" + "hyun|lsh" + "9034)"),
    "email address": re.compile(r"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b"),
    "secret-like token": re.compile(r"(?i)(api[_-]?key|access[_-]?token)\s*[:=]\s*['\"][^'\"]+"),
}


def require(condition: bool, message: str) -> None:
    if not condition:
        raise AssertionError(message)


def read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as stream:
        value = json.load(stream)
    require(isinstance(value, dict), f"JSON must be an object: {path}")
    return value


def verify_source(data_root: Path, source: str, expected_rows: int | None = None) -> np.ndarray:
    source_dir = data_root / source
    meta = read_json(source_dir / f"{source}_meta.json")
    dat_path = source_dir / f"{source}.dat"
    timestamps_path = source_dir / f"{source}_timestamps.npy"
    timestamps = np.load(timestamps_path, allow_pickle=False).astype("U12")
    row_shape = tuple(int(value) for value in meta["row_shape"])
    row_count = int(meta["row_count"])
    expected_size = row_count * int(np.prod(row_shape)) * np.dtype(meta["dtype"]).itemsize
    if expected_rows is not None:
        require(row_count == expected_rows, f"{source}: unexpected row count {row_count}")
    require(dat_path.stat().st_size == expected_size, f"{source}: byte size mismatch")
    require(len(timestamps) == row_count, f"{source}: timestamp count mismatch")
    require(len(np.unique(timestamps)) == row_count, f"{source}: duplicate timestamp")
    require(np.all(timestamps[:-1] < timestamps[1:]), f"{source}: timestamps are not strictly sorted")
    if source == "hsr_valid_mask":
        require(row_shape == (40082,), f"mask packed row shape mismatch: {row_shape}")
        require(tuple(meta["original_shape"]) == (583, 550), "mask original shape mismatch")
        require(meta.get("bitorder") == "little", "mask bitorder must be little")
    return timestamps


def sample_counts(catalog: pd.DataFrame) -> tuple[int, int]:
    rows = catalog.set_index("timestamp")
    input_complete = 0
    label_complete = 0
    for timestamp in catalog["timestamp"].astype(str):
        anchor = datetime.strptime(timestamp, "%Y%m%d%H%M")
        window = [(anchor - timedelta(minutes=value)).strftime("%Y%m%d%H%M") for value in (50, 40, 30, 20, 10, 0)]
        valid = all(
            item in rows.index
            and str(rows.at[item, "concat_status"]) == "ok"
            and str(rows.at[item, "hsr_status"]) == "ok"
            for item in window
        )
        if valid:
            input_complete += 1
            if str(rows.at[timestamp, "ci_hard_status"]) == "ok":
                label_complete += 1
    return input_complete, label_complete


def verify_catalog(data_root: Path, timestamps: dict[str, np.ndarray]) -> None:
    catalog = pd.read_csv(data_root / "catalog.csv", dtype={"timestamp": str})
    require(len(catalog) == 26496, f"catalog: expected 26496 rows, got {len(catalog)}")
    require(catalog["timestamp"].is_unique, "catalog: duplicate timestamp")
    require(catalog["timestamp"].is_monotonic_increasing, "catalog: timestamps are not sorted")
    for source, source_timestamps in timestamps.items():
        status_col = f"{source}_status"
        idx_col = f"{source}_idx"
        ok = catalog[status_col].astype(str).eq("ok")
        require(catalog.loc[ok, idx_col].notna().all(), f"{source}: ok row without index")
        indices = catalog.loc[ok, idx_col].astype(int).to_numpy()
        require(np.all((indices >= 0) & (indices < len(source_timestamps))), f"{source}: index out of range")
        require(
            np.array_equal(source_timestamps[indices], catalog.loc[ok, "timestamp"].to_numpy(dtype=str)),
            f"{source}: catalog index/timestamp mismatch",
        )
    require(sample_counts(catalog) == (24277, 5120), "catalog: inference sample counts differ from 24277/5120")


def verify_weights(root: Path) -> None:
    checkpoint_root = root / "result/training/checkpoints"
    safe = load_file(str(checkpoint_root / "model.safetensors"), device="cpu")
    payload = torch.load(checkpoint_root / "best_model.pt", map_location="cpu", weights_only=True)
    state = payload["model_state_dict"]
    require(safe.keys() == state.keys(), "weight keys differ between safetensors and pt")
    for name in safe:
        require(torch.equal(safe[name], state[name]), f"weight tensor differs: {name}")
    metadata = read_json(checkpoint_root / "checkpoint_metadata.json")
    require(int(payload["epoch"]) == int(metadata["selected_epoch"]) == 121, "selected epoch must be 121")
    require(float(metadata["validation_threshold"]) == 0.1, "validation threshold must be 0.1")
    prediction_check = read_json(checkpoint_root / "prediction_equivalence.json")
    require(bool(prediction_check.get("passed")), "checkpoint prediction equivalence failed")
    require(
        max(float(value) for value in prediction_check["max_absolute_error"].values()) <= 1e-5,
        "checkpoint prediction error exceeds 1e-5",
    )


def verify_auxiliary_checks(root: Path) -> None:
    mask_check = read_json(root / "result/final_preprocess/2025/hsr_mask_equivalence.json")
    require(bool(mask_check.get("passed")), "HSR mask equivalence failed")
    require(
        all(int(item["differing_pixels"]) == 0 for item in mask_check["checks"]),
        "HSR mask has differing pixels",
    )
    demo = read_json(root / "raw_data/20210702/demo_raw_manifest.json")
    groups = demo["groups"]
    require(int(groups["static"]["files"]) == 5, "demo must contain five fixed auxiliary files")
    require(int(groups["l1b"]["files"]) == 994, "demo must contain 994 Level-1B channel files")
    require(int(groups["l2"]["files"]) == 142, "demo must contain 142 Level-2 files")
    require(int(groups["radar_cappi"]["files"]) == 137, "demo CAPPI file count mismatch")
    require(int(groups["radar_hsr"]["files"]) == 144, "demo HSR file count mismatch")
    require(int(groups["radar_hsp"]["files"]) == 144, "demo HSP file count mismatch")


def verify_reference(root: Path) -> None:
    path = (
        root
        / "result/validation/object_validation/Model/validation_targets_2025"
        / "threshold_summary.csv"
    )
    with path.open("r", encoding="utf-8", newline="") as stream:
        rows = list(csv.DictReader(stream))
    expected_counts = {
        0.1: (5064, 3165, 8999),
        0.2: (3455, 4774, 4625),
        0.3: (2379, 5850, 2509),
        0.4: (1471, 6758, 1283),
        0.5: (808, 7421, 604),
        0.6: (354, 7875, 288),
        0.7: (170, 8059, 148),
        0.8: (94, 8135, 89),
        0.9: (23, 8206, 49),
    }
    require(len(rows) == len(expected_counts), "reference threshold count mismatch")
    indexed = {round(float(item["threshold"]), 1): item for item in rows}
    require(set(indexed) == set(expected_counts), "reference threshold values mismatch")
    for threshold, expected in expected_counts.items():
        row = indexed[threshold]
        actual = (int(row["hits"]), int(row["misses"]), int(row["falses"]))
        require(actual == expected, f"reference counts mismatch at threshold {threshold}: {actual}")
        expected_csi = expected[0] / sum(expected)
        require(
            abs(float(row["CSI"]) - expected_csi) < 1e-12,
            f"reference CSI mismatch at threshold {threshold}",
        )


def verify_inference_outputs(root: Path) -> None:
    output_root = root / "result/training/inference/2025"
    predictions = sorted(output_root.glob("*/pred_*.npy"))
    require(len(predictions) == 5120, f"inference: expected 5120 predictions, got {len(predictions)}")
    require(not any("masked" in path.name for path in predictions), "inference filenames must not contain 'masked'")
    sample = np.load(predictions[0], mmap_mode="r", allow_pickle=False)
    require(sample.shape == (1, 583, 550), f"inference sample shape mismatch: {sample.shape}")
    require(sample.dtype == np.float32, f"inference sample dtype mismatch: {sample.dtype}")


def _catalog_ok(rows: pd.DataFrame, timestamp: str, source: str) -> bool:
    if timestamp not in rows.index:
        return False
    row = rows.loc[timestamp]
    return str(row.get(f"{source}_status")) == "ok" and not pd.isna(row.get(f"{source}_idx"))


def verify_demo(root: Path) -> None:
    preprocess_root = root / "result/data_preparing/res_2km"
    l1b = sorted((preprocess_root / "L1B/20210702").glob("concat_gk2a_radar_*.npy"))
    l2 = sorted((preprocess_root / "L2/20210702").glob("l2_aii_*.npy"))
    require(len(l1b) == 142, f"demo preprocessing: expected 142 L1B rows, got {len(l1b)}")
    require(len(l2) == 142, f"demo preprocessing: expected 142 L2 rows, got {len(l2)}")

    prepared_root = root / "result/final_preprocess/demo_20210702"
    for source in ("concat", "hsr", "ci_hard", "bt", "bt_mask"):
        verify_source(prepared_root, source)

    catalog = pd.read_csv(prepared_root / "catalog.csv", dtype={"timestamp": str})
    require(catalog["timestamp"].is_unique, "demo catalog contains duplicate timestamps")
    require(catalog["timestamp"].is_monotonic_increasing, "demo catalog is not sorted")
    rows = catalog.set_index("timestamp")
    valid_samples = 0
    for timestamp in catalog["timestamp"].astype(str):
        if not ("202107020600" <= timestamp <= "202107021750"):
            continue
        anchor = datetime.strptime(timestamp, "%Y%m%d%H%M")
        input_times = [
            (anchor - timedelta(minutes=offset)).strftime("%Y%m%d%H%M")
            for offset in (50, 40, 30, 20, 10, 0)
        ]
        bt_times = [
            (anchor + timedelta(minutes=offset)).strftime("%Y%m%d%H%M")
            for offset in (10, 20, 30, 40, 50, 60)
        ]
        valid = (
            all(_catalog_ok(rows, item, "concat") and _catalog_ok(rows, item, "hsr") for item in input_times)
            and _catalog_ok(rows, timestamp, "ci_hard")
            and _catalog_ok(rows, timestamp, "bt_mask")
            and all(_catalog_ok(rows, item, "bt") for item in bt_times)
        )
        valid_samples += int(valid)
    require(valid_samples == 18, f"demo training: expected 18 valid samples, got {valid_samples}")


def verify_labeling_results(root: Path) -> None:
    labeling = root / "result/labeling"
    step1 = labeling / "step1_region_growing"
    step2 = labeling / "step2_temporal_overlap"
    step3 = labeling / "step3_mature_cloud_masking"
    require(len(list(step1.glob("2025????/*.nc"))) == 26045, "2025 step1 NetCDF count mismatch")
    require(len(list(step2.glob("2025????/*_label.nc"))) == 14208, "2025 step2 label count mismatch")
    require(len(list(step2.glob("2025????/*_links.pkl"))) == 14208, "2025 step2 links count mismatch")
    require(len(list(step2.glob("2025????/*_visited.pkl"))) == 14208, "2025 step2 visited count mismatch")
    require(len(list(step3.glob("2025????/*_label.nc"))) == 5455, "2025 step3 count mismatch")

    archives = sorted((step1 / "auxiliary_archives").glob("2025-??.tar.gz"))
    require([path.name for path in archives] == [f"2025-{month:02d}.tar.gz" for month in range(5, 11)], "step1 archive set mismatch")
    inventory = read_json(step1 / "auxiliary_archives/labeling_inventory.json")
    require(int(inventory["step1_auxiliary_members"]) == 78135, "step1 archive member count mismatch")
    archive_inventory = inventory["step1_auxiliary_archives"]
    require(sum(int(item["members"]) for item in archive_inventory) == 78135, "step1 monthly member count mismatch")
    recorded_archive_bytes = {
        f"{item['month']}.tar.gz": int(item["archive_bytes"])
        for item in archive_inventory
    }
    require(
        recorded_archive_bytes == {path.name: path.stat().st_size for path in archives},
        "step1 archive byte size differs from labeling inventory",
    )

    target_dir = root / "result/validation/validation_targets"
    target_files = sorted(target_dir.glob("*.json"))
    require(
        [path.name for path in target_files] == ["validation_targets_2025.json"],
        "validation_targets must contain only validation_targets_2025.json",
    )
    targets = read_json(target_files[0])
    needed_times: set[str] = set(key.rsplit("_", 1)[0] for key in targets)
    needed_times.update(child.rsplit("_", 1)[0] for children in targets.values() for child in children)
    available_times = {path.name[:12] for path in step2.glob("2025????/*_label.nc")}
    require(needed_times <= available_times, "validation targets reference missing step2 labels")
    require(
        not (step2 / "20250710/202507100710_label.nc").exists(),
        "unreferenced 202507100710 temporal label must not be distributed",
    )


def verify_repository_file_limits(root: Path) -> None:
    file_count = 0
    oversized_directories: list[tuple[Path, int]] = []
    directories = [root, *(path for path in root.rglob("*") if path.is_dir())]
    for directory in directories:
        entries = list(directory.iterdir())
        direct_files = sum(entry.is_file() for entry in entries)
        if direct_files >= 10_000:
            oversized_directories.append((directory.relative_to(root), direct_files))
        file_count += direct_files
    require(file_count < 100_000, f"repository has {file_count} files; expected fewer than 100000")
    require(
        not oversized_directories,
        "directories with 10000 or more direct files: "
        + ", ".join(f"{path} ({count})" for path, count in oversized_directories),
    )


def verify_layout(root: Path) -> None:
    required_directories = (
        "code/data_preparing/src",
        "code/data_preparing/run",
        "code/labeling/src",
        "code/labeling/run",
        "code/final_preprocess/src",
        "code/final_preprocess/run",
        "code/training/src",
        "code/training/run",
        "code/validation/src",
        "code/validation/run",
        "raw_data/20210702",
        "result/final_preprocess/demo_20210702",
        "result/final_preprocess/2025",
        "result/data_preparing",
        "result/labeling",
        "result/final_preprocess",
        "result/training",
        "result/validation",
        "result/validation/validation_targets",
    )
    for relative in required_directories:
        require((root / relative).is_dir(), f"required directory is missing: {relative}")
    require((root / "raw_data/README.md").is_file(), "required file is missing: raw_data/README.md")
    for legacy in (
        "code/" + "cinet",
        "code/" + "preprocess",
        "code/" + "modeling",
        "result/" + "preprocess",
        "result/" + "modeling",
        "configs",
        "scripts",
        "tools",
        "weights",
        "outputs",
        "results",
        "data",
    ):
        require(not (root / legacy).exists(), f"legacy release path still exists: {legacy}")

    expected_run_files = {
        "data_preparing": {"config.yaml", "run.sh"},
        "labeling": {"config.yaml", "run.sh"},
        "final_preprocess": {"config.yaml", "run.sh"},
        "training": {"train.yaml", "train.sh", "inference.yaml", "inference.sh"},
        "validation": {"targets.yaml", "targets.sh", "validation.yaml", "validation.sh"},
    }
    for stage, expected in expected_run_files.items():
        actual = {path.name for path in (root / "code" / stage / "run").iterdir() if path.is_file()}
        require(actual == expected, f"{stage}/run files differ: expected {sorted(expected)}, got {sorted(actual)}")


def verify_configs_and_clis(root: Path) -> None:
    required_keys = {
        "code/data_preparing/run/config.yaml": {"Calibration_table_path", "save_dir", "start_date", "end_date", "channels"},
        "code/labeling/run/config.yaml": {"input_root", "output_dir", "stages"},
        "code/final_preprocess/run/config.yaml": {"time_ranges", "output_root", "sources", "inputs", "labels"},
        "code/training/run/train.yaml": {"dataset", "train", "valid", "model", "loss", "optimizer", "scheduler"},
        "code/training/run/inference.yaml": {"dataset", "output_root", "catalog_path", "model", "inference", "radar_mask"},
        "code/validation/run/targets.yaml": {"tracking", "availability_filter"},
        "code/validation/run/validation.yaml": {"dates", "paths", "validation", "matching", "clusterer", "providers"},
    }
    for relative, keys in required_keys.items():
        path = root / relative
        with path.open("r", encoding="utf-8") as stream:
            config = yaml.safe_load(stream)
        require(isinstance(config, dict), f"invalid config mapping: {path}")
        missing = keys - set(config)
        require(not missing, f"config schema missing {sorted(missing)}: {relative}")
    label_config = root / "code/labeling/run/config.yaml"
    label_stages = yaml.safe_load(label_config.read_text(encoding="utf-8"))["stages"]
    require(set(label_stages) == {"step1", "step2", "step3"}, "demo label config must define three stages")
    scripts = (
        "code/data_preparing/run/run.sh",
        "code/labeling/run/run.sh",
        "code/final_preprocess/run/run.sh",
        "code/training/run/train.sh",
        "code/training/run/inference.sh",
        "code/validation/run/targets.sh",
        "code/validation/run/validation.sh",
    )
    for relative in scripts:
        result = subprocess.run(
            [str(root / relative), "--help"],
            cwd=root,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
            text=True,
        )
        require(result.returncode == 0, f"CLI help failed for {relative}: {result.stderr}")


def verify_privacy(root: Path) -> None:
    failures: list[str] = []
    for path in sorted(root.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in TEXT_SUFFIXES:
            continue
        if any(part in {"outputs", ".git", "__pycache__"} for part in path.relative_to(root).parts):
            continue
        try:
            text = path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            continue
        for label, pattern in FORBIDDEN.items():
            if pattern.search(text):
                failures.append(f"{path.relative_to(root)}: {label}")
    require(not failures, "privacy scan failed:\n" + "\n".join(failures[:50]))


def file_hash(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while block := stream.read(64 * 1024 * 1024):
            digest.update(block)
    return digest.hexdigest()


def verify_manifest(root: Path) -> None:
    manifest = root / "MANIFEST.sha256"
    require(manifest.is_file(), "MANIFEST.sha256 is missing")
    for line_number, line in enumerate(manifest.read_text(encoding="utf-8").splitlines(), start=1):
        expected, relative = line.split("  ", 1)
        path = root / relative
        require(path.is_file(), f"manifest line {line_number}: missing {relative}")
        require(file_hash(path) == expected, f"manifest line {line_number}: checksum mismatch for {relative}")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", required=True, type=Path)
    parser.add_argument("--skip-manifest", action="store_true", help="Skip the expensive full checksum pass")
    args = parser.parse_args()
    root = args.root.resolve()

    verify_layout(root)
    data_root = root / "result/final_preprocess/2025"
    timestamps = {
        source: verify_source(data_root, source, expected_rows)
        for source, expected_rows in EXPECTED_ROWS.items()
    }
    verify_catalog(data_root, timestamps)
    verify_weights(root)
    verify_auxiliary_checks(root)
    verify_reference(root)
    verify_inference_outputs(root)
    verify_demo(root)
    verify_labeling_results(root)
    verify_repository_file_limits(root)
    verify_configs_and_clis(root)
    verify_privacy(root)
    if not args.skip_manifest:
        verify_manifest(root)
    print("CI-Net release verification passed.")


if __name__ == "__main__":
    main()