File size: 3,955 Bytes
7da2ecb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | #!/usr/bin/env python3
"""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} # type: ignore[index]
summary["total_files"] = sum(int(value["files"]) for value in summary["groups"].values()) # type: ignore[union-attr]
summary["total_bytes"] = sum(int(value["bytes"]) for value in summary["groups"].values()) # type: ignore[union-attr]
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()
|