| |
| """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() |
|
|
