#!/usr/bin/env python3 # check_consistency_month.py — compare master TXT vs daily NC (per-day NaN%) # Usage: # ./run_check.sh check_consistency_month.py \ # --in-dir /instruments/UQAM_PK/PK_Rooftop/parsivel \ # --out-dir /station/instruments/disdrodb/PAR001_UQAM_PK \ # --month 2025-09 \ # --vars sensor_temperature mor_visibility laser_amplitude rainfall_rate_32bit reflectivity_32bit kinetic_energy import argparse, csv, re, sys from pathlib import Path from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple import numpy as np from netCDF4 import Dataset TIME_CANDIDATES = ("Timestamp","timestamp","time","DateTime","datetime","Date","UTC","Time_UTC") EXTRA_MAP = { "rainfall_rate_32bit": ("rainfall_rate_32bit","Intensity of precipitation","Intensity of precipitation (mm/h)"), "reflectivity_32bit": ("reflectivity_32bit","Radar reflectivity","Radar reflectivity (dBz)"), "mor_visibility": ("mor_visibility","MOR Visibility (m)"), "laser_amplitude": ("laser_amplitude","Signal amplitude of Laserband","Sensor voltage (V)"), "number_particles_validated": ("number_particles_validated","Number of detected particles"), "sensor_temperature": ("sensor_temperature","Temperature in sensor (C)","Sensor_temperture"), "kinetic_energy": ("kinetic_energy","Kinetic Energy"), "error_code": ("error_code","Error code"), } _NUM_KEEP = re.compile(r"[^\d\.\+\-eE]") def info(m: str): print(f"[info {datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')}] {m}") def detect_sep(path: Path) -> str: with open(path, "r", encoding="utf-8", errors="ignore") as f: lines = [f.readline() for _ in range(10)] c = sum(l.count(",") for l in lines if l) s = sum(l.count(";") for l in lines if l) return "," if c >= s else ";" def canon(s: str) -> str: return re.sub(r"[^0-9a-z]+", "", s.lower()) def find_col(header: List[str], *aliases: str) -> Optional[int]: idx = {canon(c): i for i, c in enumerate(header)} for a in aliases: ca = canon(a) if ca in idx: return idx[ca] for a in aliases: ca = canon(a) for i, c in enumerate(header): if ca in canon(c): return i return None def parse_utc_timestamp(s: str): if not s: return None t = s.strip() if t.endswith("Z"): t = t[:-1] + "+00:00" try: dt = datetime.fromisoformat(t) dt = dt.replace(tzinfo=dt.tzinfo or timezone.utc) return dt.astimezone(timezone.utc) except Exception: pass for fmt in ("%Y-%m-%d %H:%M:%S","%Y/%m/%d %H:%M:%S","%Y-%m-%d %H:%M","%Y-%m-%dT%H:%M:%S","%Y-%m-%dT%H:%M"): try: return datetime.strptime(t, fmt).replace(tzinfo=timezone.utc) except Exception: pass return None def to_float(s: str) -> float: if s is None: return np.nan t = s.strip() if t == "" or t.lower() in ("nan","none","null"): return np.nan if re.search(r"\d,\d", t) and not re.search(r"\d\.\d", t): t = t.replace(".", "").replace(",", ".") else: t = _NUM_KEEP.sub("", t) try: return float(t) except Exception: return np.nan def month_paths(in_root: Path, month: str, pattern: str): ym = month.replace("-", "_") base = in_root / ym return sorted(base.rglob(pattern)) def days_in_txt(fp: Path): sep = detect_sep(fp) with open(fp, "r", encoding="utf-8", errors="ignore", newline="") as f: r = csv.reader(f, delimiter=sep) header = next(r) t_idx = find_col(header, *TIME_CANDIDATES) if t_idx is None: raise ValueError(f"time column not found in {fp}") counts = {} for row in r: if not row or t_idx >= len(row): continue dt = parse_utc_timestamp(row[t_idx]) if dt is None: continue d = dt.date().isoformat() counts[d] = counts.get(d, 0) + 1 return counts, header, t_idx, sep def nc_path_for_day(out_root: Path, id_tag: str, day: str) -> Path: y, m, d = day.split("-") return out_root / y / f"{y}{m}" / f"{id_tag}_{y}{m}{d}.nc" def compare_month(in_root: Path, out_root: Path, month: str, variables: List[str], pattern: str): id_tag = out_root.name txts = month_paths(in_root, month, pattern) if not txts: info(f"No month files under {in_root} for {month}") return if not variables: variables = list(EXTRA_MAP.keys()) print("# day var txt_n(%) nc_n(%) diff(%) txt_count nc_time_len note") for fp in txts: info(f"TXT: {fp}") try: counts, header, t_idx, sep = days_in_txt(fp) except Exception as e: info(f" skip month: {e}") continue alias_idx = {v: find_col(header, *(EXTRA_MAP.get(v, (v,)))) for v in variables} day_stats = {d: {v: (0,0) for v in variables} for d in counts} with open(fp, "r", encoding="utf-8", errors="ignore", newline="") as f: r = csv.reader(f, delimiter=sep); next(r, None) for row in r: if not row or t_idx >= len(row): continue dt = parse_utc_timestamp(row[t_idx]) if dt is None: continue d = dt.date().isoformat() if d not in day_stats: continue for v in variables: col = alias_idx[v] nn, tt = day_stats[d][v] if col is None or col >= len(row): day_stats[d][v] = (nn, tt + 1) else: val = to_float(row[col]) day_stats[d][v] = (nn + (0 if np.isnan(val) else 1), tt + 1) for d in sorted(counts): nc_fp = nc_path_for_day(out_root, id_tag, d) if not nc_fp.exists(): print(f"{d:10s} {'-':28s} {'-':7s} {'-':7s} {'-':7s} {counts[d]:9d} {'-':12s} missing_nc") continue with Dataset(str(nc_fp), "r") as ds: nctime = ds.dimensions["time"].size if "time" in ds.dimensions else -1 for v in variables: nn, tt = day_stats[d][v] txt_nan = (100.0 * (tt - nn) / tt) if tt else float("nan") if v in ds.variables: arr = ds.variables[v][:] nc_nan = (100.0 * np.isnan(arr).sum() / arr.size) if arr.size else float("nan") diff = (nc_nan - txt_nan) if not (np.isnan(nc_nan) or np.isnan(txt_nan)) else float("nan") print(f"{d:10s} {v:28s} {txt_nan:7.1f} {nc_nan:7.1f} {diff:7.1f} {tt:9d} {nctime:12d}") else: print(f"{d:10s} {v:28s} {txt_nan:7.1f} {'-':7s} {'-':7s} {tt:9d} {nctime:12d} missing_var_in_nc") def main(): ap = argparse.ArgumentParser() ap.add_argument("--in-dir", type=Path, required=True, help="e.g., /instruments/UQAM_PK/PK_Rooftop/parsivel") ap.add_argument("--out-dir", type=Path, required=True, help="e.g., /station/instruments/disdrodb/PAR_UNBC_TERRACE") ap.add_argument("--month", required=True, help="YYYY-MM (or YYYY_MM)") ap.add_argument("--pattern", default="**/master/*.txt") ap.add_argument("--vars", nargs="*", default=[], help="variables to compare") a = ap.parse_args() compare_month(a.in_dir, a.out_dir, a.month, a.vars, a.pattern) if __name__ == "__main__": sys.exit(main())