| |
| """Copy exactly one UTC day of raw demo inputs and fixed auxiliary files.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| import re |
| import shutil |
| from datetime import datetime, timedelta, timezone |
| from pathlib import Path |
|
|
|
|
| TIMESTAMP = re.compile(r"(\d{12})") |
|
|
|
|
| def copy_file(source: Path, destination: Path) -> int: |
| destination.parent.mkdir(parents=True, exist_ok=True) |
| if destination.exists() and destination.stat().st_size == source.stat().st_size: |
| return destination.stat().st_size |
| temporary = destination.with_name(f".{destination.name}.partial") |
| shutil.copyfile(source, temporary) |
| os.replace(temporary, destination) |
| return destination.stat().st_size |
|
|
|
|
| def copy_tree_files(source_root: Path, destination_root: Path, files: list[Path]) -> tuple[int, int]: |
| total = 0 |
| for index, source in enumerate(files, start=1): |
| total += copy_file(source, destination_root / source.relative_to(source_root)) |
| if index % 100 == 0: |
| print(f"[copy] {index:,}/{len(files):,} files", flush=True) |
| return len(files), total |
|
|
|
|
| def timestamp_of(path: Path) -> datetime | None: |
| match = TIMESTAMP.search(path.name) |
| if match is None: |
| return None |
| return datetime.strptime(match.group(1), "%Y%m%d%H%M") |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--source-auxdata", required=True, type=Path) |
| parser.add_argument("--output-root", required=True, type=Path) |
| parser.add_argument("--date", default="20210702") |
| args = parser.parse_args() |
|
|
| source = args.source_auxdata.resolve() |
| destination = args.output_root.resolve() / "auxdata" |
| utc_start = datetime.strptime(args.date, "%Y%m%d") |
| utc_end = utc_start + timedelta(days=1) |
| radar_start = utc_start + timedelta(hours=9) |
| radar_end = utc_end + timedelta(hours=9) |
|
|
| selected: dict[str, list[Path]] = {} |
| selected["static"] = [source / "gk2a_AMI_Calibration_Table.xlsx", *sorted((source / "latlon").glob("*"))] |
|
|
| for level in ("L1B", "L2"): |
| level_root = source / "gk2a" / level / "EA" / args.date[:6] / args.date[6:] |
| selected[level.lower()] = sorted( |
| path for path in level_root.rglob("*") |
| if path.is_file() and (stamp := timestamp_of(path)) is not None and utc_start <= stamp < utc_end |
| ) |
| for product in ("cappi", "hsr", "hsp"): |
| selected[f"radar_{product}"] = sorted( |
| path for path in (source / "radar" / product).rglob("*") |
| if path.is_file() and (stamp := timestamp_of(path)) is not None and radar_start <= stamp < radar_end |
| ) |
|
|
| summary: dict[str, object] = { |
| "utc_date": args.date, |
| "radar_filename_time": { |
| "timezone": "Asia/Seoul (UTC+09:00)", |
| "start_inclusive": radar_start.strftime("%Y%m%d%H%M"), |
| "end_exclusive": radar_end.strftime("%Y%m%d%H%M"), |
| }, |
| "groups": {}, |
| } |
| for group, files in selected.items(): |
| missing = [str(path.name) for path in files if not path.exists()] |
| if missing: |
| raise FileNotFoundError(f"missing source files for {group}: {missing}") |
| count, size = copy_tree_files(source, destination, files) |
| summary["groups"][group] = {"files": count, "bytes": size} |
|
|
| summary["total_files"] = sum(int(value["files"]) for value in summary["groups"].values()) |
| summary["total_bytes"] = sum(int(value["bytes"]) for value in summary["groups"].values()) |
| args.output_root.mkdir(parents=True, exist_ok=True) |
| with (args.output_root / "demo_raw_manifest.json").open("w", encoding="utf-8") as stream: |
| json.dump(summary, stream, indent=2, ensure_ascii=False) |
| stream.write("\n") |
| print(json.dumps(summary, indent=2), flush=True) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
