#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ check_fill_value.py — verify CF-ish fill semantics and dtype hygiene Scans NetCDFs under the given roots (defaults: ../PAR001_UQAM_PK and ../PAR_UNBC_TERRACE) and reports, per corpus and per variable: - files presence count - dtype set, shape set - _FillValue and missing_value attribute sets - total sample count, #NaN, #equals(_FillValue), #zeros Flags: - missing _FillValue on float data - multiple distinct _FillValue values across files - float64 (f8) where float32 (f4) likely suffices - high zero count with no fill (possible zero-fill of missing) Dependencies: pathlib, argparse, numpy, netCDF4. No pandas/xarray. """ from __future__ import annotations import argparse from pathlib import Path from collections import defaultdict, Counter import numpy as np from netCDF4 import Dataset # ---------- IO helpers ---------- def iter_nc_files(root: Path) -> list[Path]: if not root.exists(): return [] return sorted(p for p in root.rglob("*.nc") if p.is_file()) def open_ds(p: Path): try: return Dataset(str(p), "r") except Exception as e: print(f"[skip] cannot open {p}: {e}") return None def is_float_dtype(arr_like) -> bool: try: dt = np.asarray(arr_like).dtype return np.issubdtype(dt, np.floating) except Exception: return False def count_nan_fill_zero(var) -> tuple[int,int,int,int]: """ Returns (total, n_nan, n_fill, n_zero). Mask-safe, and only counts 'fill' if var has a finite _FillValue and dtype is float. """ try: data = var[:] except Exception: return (0, 0, 0, 0) total = int(data.size) if total == 0: return (0, 0, 0, 0) # Normalize to ndarray with NaNs for masked points if np.ma.isMaskedArray(data): data = data.filled(np.nan) else: data = np.asarray(data) n_nan = int(np.count_nonzero(np.isnan(data))) if is_float_dtype(data) else 0 fv = getattr(var, "_FillValue", None) n_fill = 0 if is_float_dtype(data) and fv is not None and np.isfinite(fv): try: n_fill = int(np.count_nonzero(data == fv)) except Exception: n_fill = 0 # Zero hits (exact 0.0 for floats; 0 for ints); useful to catch zero-filled missing try: n_zero = int(np.count_nonzero(data == 0)) except Exception: n_zero = 0 return (total, n_nan, n_fill, n_zero) # ---------- Aggregation ---------- def analyze_corpus(root: Path, sample_limit: int | None = None) -> dict: files = iter_nc_files(root) if sample_limit: files = files[:sample_limit] per_var = defaultdict(lambda: { "files": 0, "total": 0, "nan": 0, "fill": 0, "zero": 0, "dtypes": Counter(), "shapes": Counter(), "_FillValue": Counter(), "missing_value": Counter(), }) opened = 0 for p in files: ds = open_ds(p) if ds is None: continue opened += 1 try: for name, v in ds.variables.items(): pv = per_var[name] pv["files"] += 1 # attribute signatures fv = getattr(v, "_FillValue", None) mv = getattr(v, "missing_value", None) pv["_FillValue"][repr(fv)] += 1 pv["missing_value"][repr(mv)] += 1 pv["dtypes"][str(np.asarray(v[:1]).dtype if v.size else np.asarray(v[:]).dtype)] += 1 pv["shapes"][str(tuple(int(x) for x in v.shape))] += 1 # tallies total, n_nan, n_fill, n_zero = count_nan_fill_zero(v) pv["total"] += total pv["nan"] += n_nan pv["fill"] += n_fill pv["zero"] += n_zero finally: ds.close() return {"root": root, "n_files": opened, "per_var": per_var} # ---------- Reporting ---------- def pct(part: int, whole: int) -> float: return 100.0 * part / whole if whole > 0 else 0.0 def most_common(counter: Counter) -> str: return counter.most_common(1)[0][0] if counter else "-" def flag_var(name: str, rec: dict) -> list[str]: """Return a list of human-readable flags for this variable record.""" flags = [] dtype_set = set(rec["dtypes"].keys()) # Float without a fill value in any file # If any dtype is float and the only _FillValue observed is 'None' if any(dt.startswith("float") for dt in dtype_set): fv_keys = set(rec["_FillValue"].keys()) if fv_keys == {"None"}: flags.append("missing _FillValue on float") elif len([k for k in fv_keys if k != "None"]) > 1: flags.append("inconsistent _FillValue across files") # zero-heavy without fill if rec["fill"] == 0 and pct(rec["zero"], rec["total"]) >= 1.0 and pct(rec["nan"], rec["total"]) < 0.1: flags.append("zeros present but no fill hits (possible zero-fill of missing)") # f8 detection if any(dt.startswith("float64") for dt in dtype_set): flags.append("float64 detected (consider float32)") # shape inconsistency if len(rec["shapes"]) > 1: flags.append("multiple shapes across files") return flags def print_corpus_report(title: str, corpus: dict, show_all: bool = False): root = corpus["root"] per_var = corpus["per_var"] n_files = corpus["n_files"] print(f"\n=== {title} ===") print(f"root: {root} | opened files: {n_files}") print("name files dtype* shape* FillValue* miss_val* totals NaN% fill% zero% flags") print(" *most-common shown per corpus; percentages over all samples of that var") names = sorted(per_var.keys()) for name in names: rec = per_var[name] files = rec["files"] if files == 0: continue dtype_star = most_common(rec["dtypes"]) shape_star = most_common(rec["shapes"]) fv_star = most_common(rec["_FillValue"]) mv_star = most_common(rec["missing_value"]) nanp = pct(rec["nan"], rec["total"]) fillp = pct(rec["fill"], rec["total"]) zerop = pct(rec["zero"], rec["total"]) flags = flag_var(name, rec) if not show_all and not flags: # hide clean rows unless user wants all continue print(f"{name:28s} {files:5d} {dtype_star:20s} {shape_star:15s} {fv_star:15s} {mv_star:15s} {rec['total']:9d} {nanp:6.2f}% {fillp:6.2f}% {zerop:6.2f}% {'; '.join(flags) if flags else ''}") # ---------- CLI ---------- def main(): here = Path(__file__).resolve().parent default_uqam = (here.parent / "PAR001_UQAM_PK").resolve() default_terr = (here.parent / "PAR_UNBC_TERRACE").resolve() ap = argparse.ArgumentParser(description="Check _FillValue/missing semantics for NetCDF disdrometer outputs.") ap.add_argument("--root", action="append", type=Path, help="Root folder to scan (can be given multiple times). Defaults to ../PAR001_UQAM_PK and ../PAR_UNBC_TERRACE") ap.add_argument("--sample", type=int, default=None, help="Limit number of files per root (for speed).") ap.add_argument("--show-all", action="store_true", help="Show all variables, not only flagged ones.") args = ap.parse_args() roots = args.root if args.root else [default_uqam, default_terr] reports = [] for r in roots: reports.append(analyze_corpus(r, sample_limit=args.sample)) # Print flagged-first reports for r in reports: title = r["root"].name or str(r["root"]) print_corpus_report(title, r, show_all=args.show_all) # Quick summary line print("\nSummary:") for r in reports: print(f" - {r['root']}: scanned {r['n_files']} files, {len(r['per_var'])} variables observed") if __name__ == "__main__": raise SystemExit(main())