From 7205b14edab15b5a90a2dd92d14f03885125ed1c Mon Sep 17 00:00:00 2001 From: Quintin Date: Mon, 20 Jul 2026 14:36:19 -0700 Subject: [PATCH 1/3] Add scripts and portal file patterns --- portal-prep/README.md | 23 ++ .../portal-file-patterns/bccaqv2_u5.txt | 3 + .../portal-file-patterns/bccaqv2_u6.txt | 22 + .../portal-file-patterns/canada_mosaic.txt | 6 + .../portal-file-patterns/canesm5_m6.txt | 5 + .../portal-file-patterns/canesm5_u6.txt | 5 + .../portal-file-patterns/gridded_daily.txt | 9 + portal-prep/portal-file-patterns/mbcn.txt | 16 + portal-prep/portal-file-patterns/prism.txt | 5 + portal-prep/portal-file-patterns/vicgl.txt | 4 + portal_meta_builder/__init__.py | 4 + portal_meta_builder/builder.py | 152 +++++++ portal_meta_builder/io_utils.py | 29 ++ portal_meta_builder/metadata.py | 243 +++++++++++ portal_meta_builder/minmax.py | 98 +++++ portal_meta_builder/portals.py | 378 ++++++++++++++++++ scripts/calculate-portal-minmax.py | 175 ++++++++ scripts/chunk_gridded.sh | 49 +++ scripts/gen_hardlinks.py | 294 ++++++++++++++ scripts/update-portal-meta.py | 88 ++++ 20 files changed, 1608 insertions(+) create mode 100644 portal-prep/README.md create mode 100644 portal-prep/portal-file-patterns/bccaqv2_u5.txt create mode 100644 portal-prep/portal-file-patterns/bccaqv2_u6.txt create mode 100644 portal-prep/portal-file-patterns/canada_mosaic.txt create mode 100644 portal-prep/portal-file-patterns/canesm5_m6.txt create mode 100644 portal-prep/portal-file-patterns/canesm5_u6.txt create mode 100644 portal-prep/portal-file-patterns/gridded_daily.txt create mode 100644 portal-prep/portal-file-patterns/mbcn.txt create mode 100644 portal-prep/portal-file-patterns/prism.txt create mode 100644 portal-prep/portal-file-patterns/vicgl.txt create mode 100644 portal_meta_builder/__init__.py create mode 100644 portal_meta_builder/builder.py create mode 100644 portal_meta_builder/io_utils.py create mode 100644 portal_meta_builder/metadata.py create mode 100644 portal_meta_builder/minmax.py create mode 100644 portal_meta_builder/portals.py create mode 100644 scripts/calculate-portal-minmax.py create mode 100644 scripts/chunk_gridded.sh create mode 100644 scripts/gen_hardlinks.py create mode 100644 scripts/update-portal-meta.py diff --git a/portal-prep/README.md b/portal-prep/README.md new file mode 100644 index 0000000..2f78722 --- /dev/null +++ b/portal-prep/README.md @@ -0,0 +1,23 @@ +Portal prep files + +`portal-prep/portal-file-patterns/` is the source of truth for which datasets belong to each portal. + +Each portal has a `.txt` file containing one file pattern rule per line: +- Absolute file path, for example `/storage/.../file.nc` +- A glob pattern containing `*`, `?`, or `[` +- An exclude pattern prefixed with `!` + +Rules: +- Blank lines are ignored +- Lines starting with `#` are ignored +- `!(...)` extglob is not supported +- Filenames must remain unique within a portal after expansion because the hardlink mirror uses `/` + +Primary commands: + +```bash +python3 scripts/gen_hardlinks.py +python3 scripts/calculate-portal-minmax.py +python3 scripts/update-portal-meta.py +``` + diff --git a/portal-prep/portal-file-patterns/bccaqv2_u5.txt b/portal-prep/portal-file-patterns/bccaqv2_u5.txt new file mode 100644 index 0000000..d066c34 --- /dev/null +++ b/portal-prep/portal-file-patterns/bccaqv2_u5.txt @@ -0,0 +1,3 @@ +/storage/data/climate/downscale/BCCAQ2/bccaqv2_with_metadata/*_19500101-21001231.nc +!/storage/data/climate/downscale/BCCAQ2/bccaqv2_with_metadata/*subset*_19500101-21001231.nc + diff --git a/portal-prep/portal-file-patterns/bccaqv2_u6.txt b/portal-prep/portal-file-patterns/bccaqv2_u6.txt new file mode 100644 index 0000000..873fd0d --- /dev/null +++ b/portal-prep/portal-file-patterns/bccaqv2_u6.txt @@ -0,0 +1,22 @@ +# BCCAQv2 downscaled climate scenarios (CMIP6_BCCAQv2) +# Excludes Derived by only including files in the top-level model directories. + +/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/*/*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/AgroClimate/* +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CWEC2020_Factors/* +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/Degree_Climatologies/* +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/Ensemble_Averages/* +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r2i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r3i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r4i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r5i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r6i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r7i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r8i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r9i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*_r10i1p2f1_*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/dtr_monmean_r1_ssp585.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/dtr_r1_ssp585.nc + + + diff --git a/portal-prep/portal-file-patterns/canada_mosaic.txt b/portal-prep/portal-file-patterns/canada_mosaic.txt new file mode 100644 index 0000000..2eb72a5 --- /dev/null +++ b/portal-prep/portal-file-patterns/canada_mosaic.txt @@ -0,0 +1,6 @@ +# Canada Mosaic 30arcsec portal membership +/storage/data/climate/observations/gridded/Canada_mosaic_30arcsec/*.nc +!/storage/data/climate/observations/gridded/Canada_mosaic_30arcsec/*seasann*.nc + + + diff --git a/portal-prep/portal-file-patterns/canesm5_m6.txt b/portal-prep/portal-file-patterns/canesm5_m6.txt new file mode 100644 index 0000000..70ed7aa --- /dev/null +++ b/portal-prep/portal-file-patterns/canesm5_m6.txt @@ -0,0 +1,5 @@ +# CanESM5 (Multivariate) — MBCn/CanESM5_10 +# Top-level files only (exclude Derived trees that duplicate basenames) +/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*txtn*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*370*.nc diff --git a/portal-prep/portal-file-patterns/canesm5_u6.txt b/portal-prep/portal-file-patterns/canesm5_u6.txt new file mode 100644 index 0000000..3ebff88 --- /dev/null +++ b/portal-prep/portal-file-patterns/canesm5_u6.txt @@ -0,0 +1,5 @@ +# CanESM5 (Univariate) — BCCAQv2/CanESM5 +# Top-level files only (exclude Derived trees that duplicate basenames) +/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/*.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/dtr_monmean_r1_ssp585.nc +!/storage/data/climate/downscale/BCCAQ2/CMIP6_BCCAQv2/CanESM5/dtr_r1_ssp585.nc diff --git a/portal-prep/portal-file-patterns/gridded_daily.txt b/portal-prep/portal-file-patterns/gridded_daily.txt new file mode 100644 index 0000000..15e12a7 --- /dev/null +++ b/portal-prep/portal-file-patterns/gridded_daily.txt @@ -0,0 +1,9 @@ +# Daily Gridded Meteorological Datasets portal membership +# PCIC Blend diagonal files +/storage/data/climate/observations/gridded/PCIC_Blend/diagonal/*day_PCIC_Blended_Observations_v1_1950-2012.nc + +# VIC gen2 forcing (if present) +/storage/data/projects/dataportal/data/vic-gen2-forcing/anusplin_*.nc +/storage/data/projects/dataportal/data/vic-gen2-forcing/PNWNAmet_*.nc +!/storage/data/projects/dataportal/data/vic-gen2-forcing/PNWNAmet_*invert_lat.nc + diff --git a/portal-prep/portal-file-patterns/mbcn.txt b/portal-prep/portal-file-patterns/mbcn.txt new file mode 100644 index 0000000..c644407 --- /dev/null +++ b/portal-prep/portal-file-patterns/mbcn.txt @@ -0,0 +1,16 @@ +# MBCn downscaled climate scenarios (CMIP6_MBCn) +# Top-level daily files in model folders. + +/storage/data/climate/downscale/MBCn/CMIP6_MBCn/*/*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r2i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r3i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r4i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r5i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r6i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r7i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r8i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r9i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*_r10i1p2f1_*.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/TESTING_REGION/template_PCIC-Blend_PNW.nc +!/storage/data/climate/downscale/MBCn/CMIP6_MBCn/CanESM5_10/*txtn*.nc + diff --git a/portal-prep/portal-file-patterns/prism.txt b/portal-prep/portal-file-patterns/prism.txt new file mode 100644 index 0000000..7b61370 --- /dev/null +++ b/portal-prep/portal-file-patterns/prism.txt @@ -0,0 +1,5 @@ +# PRISM portal membership + +/storage/data/climate/observations/gridded/PRISM/split_climos/*ClimMean_PRISM_historical_*.nc + +/storage/data/climate/observations/gridded/PRISM/bc_*_monthly_CAI_timeseries_19500101_20071231.nc diff --git a/portal-prep/portal-file-patterns/vicgl.txt b/portal-prep/portal-file-patterns/vicgl.txt new file mode 100644 index 0000000..dade5a1 --- /dev/null +++ b/portal-prep/portal-file-patterns/vicgl.txt @@ -0,0 +1,4 @@ +# Gridded Hydrologic Model Output (VICGL) +/storage/data/projects/hydrology/dataportal/CMIP5/VICGL/**/*.nc +!/storage/data/projects/hydrology/dataportal/CMIP5/VICGL/*TRANSP_VEG_pfc.nc + diff --git a/portal_meta_builder/__init__.py b/portal_meta_builder/__init__.py new file mode 100644 index 0000000..0dbdae8 --- /dev/null +++ b/portal_meta_builder/__init__.py @@ -0,0 +1,4 @@ +from .builder import build_portal_payload + +__all__ = ["build_portal_payload"] + diff --git a/portal_meta_builder/builder.py b/portal_meta_builder/builder.py new file mode 100644 index 0000000..1b0ba59 --- /dev/null +++ b/portal_meta_builder/builder.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict, List, Set, Tuple + +from .io_utils import file_fingerprint, load_json, utc_now_iso +from .metadata import ensure_derived_fields, read_netcdf_metadata +from .minmax import select_minmax_record +from .portals import build_menu_tree, derive_menu_fields, get_portal_config + + +def files_from_mirror_root(mirror_root: Path, portal_id: str) -> List[Path]: + portal_root = mirror_root / portal_id + if not portal_root.exists() or not portal_root.is_dir(): + raise SystemExit(f"Mirror portal directory not found: {portal_root}") + + files = sorted( + [path for path in portal_root.iterdir() if path.is_file()], + key=lambda path: path.name.lower(), + ) + if not files: + raise SystemExit(f"No files found in mirror portal directory: {portal_root}") + return files + + +def build_entry( + portal_id: str, + src: Path, + fingerprint: Dict[str, int], + metadata: Dict[str, Any], +) -> Dict[str, Any]: + return { + "portal": portal_id, + "sourcePath": str(src), + "basename": src.name, + "fingerprint": fingerprint, + "thredds": { + "urlPath": f"data/{portal_id}/{src.name}", + "fileServer": f"/thredds/fileServer/data/{portal_id}/{src.name}", + "wms": f"/thredds/wms/data/{portal_id}/{src.name}", + "ncssGrid": f"/thredds/ncss/grid/data/{portal_id}/{src.name}", + "ncml": f"/thredds/ncml/data/{portal_id}/{src.name}", + }, + "metadata": metadata, + "menuFields": derive_menu_fields(portal_id, metadata), + "updatedAt": utc_now_iso(), + } + + +def build_portal_payload( + portal_id: str, + mirror_root: Path, + out_path: Path, + minmax_lookup: Dict[str, List[Dict[str, Any]]], + minmax_source_name: str, + prune: bool = False, +) -> Tuple[Dict[str, Any], Dict[str, int], int]: + files = files_from_mirror_root(mirror_root, portal_id) + payload = load_json(out_path) + entries: Dict[str, Any] = ( + payload.get("files") if isinstance(payload.get("files"), dict) else {} + ) + + updated = 0 + kept = 0 + removed = 0 + current_keys: Set[str] = set() + menu_items: List[Tuple[str, Dict[str, str]]] = [] + + for src in files: + source_key = str(src) + current_keys.add(source_key) + + existing = entries.get(source_key) if isinstance(entries.get(source_key), dict) else {} + fingerprint = file_fingerprint(src) + needs_refresh = ( + not existing + or existing.get("fingerprint") != fingerprint + or not isinstance(existing.get("metadata"), dict) + or not isinstance(existing.get("menuFields"), dict) + ) + + if needs_refresh: + metadata = read_netcdf_metadata(src) + entry = build_entry(portal_id, src, fingerprint, metadata) + lookup = select_minmax_record(minmax_lookup, source_key, src.name, portal_id) + if lookup: + entry["rendering"] = lookup + entries[source_key] = entry + updated += 1 + else: + entry = existing + kept += 1 + + cached_metadata = entry.get("metadata") if isinstance(entry.get("metadata"), dict) else {} + if cached_metadata: + cached_metadata = ensure_derived_fields(cached_metadata) + entry["metadata"] = cached_metadata + recomputed = derive_menu_fields(portal_id, cached_metadata) + if recomputed and recomputed != entry.get("menuFields"): + entry["menuFields"] = recomputed + entry["updatedAt"] = utc_now_iso() + entries[source_key] = entry + updated += 1 + + lookup = select_minmax_record(minmax_lookup, source_key, src.name, portal_id) + existing_render = entry.get("rendering") if isinstance(entry.get("rendering"), dict) else None + if lookup: + if existing_render != lookup: + entry["rendering"] = lookup + entry["updatedAt"] = utc_now_iso() + entries[source_key] = entry + updated += 1 + elif existing_render and existing_render.get("source") == minmax_source_name: + entry.pop("rendering", None) + entry["updatedAt"] = utc_now_iso() + entries[source_key] = entry + updated += 1 + + fields = entry.get("menuFields") if isinstance(entry.get("menuFields"), dict) else {} + menu_items.append((src.name, fields)) + if portal_id == "bccaqv2_u6": + alt = str(fields.get("scenarioPcic12") or "").strip() + if alt: + fields_alt = dict(fields) + fields_alt["scenario"] = alt + menu_items.append((src.name, fields_alt)) + + if prune: + stale = [key for key in list(entries.keys()) if key not in current_keys] + for key in stale: + del entries[key] + removed = len(stale) + + config = get_portal_config(portal_id) + schema = config["menuSchema"] + order = list(schema.get("order") or ["variable"]) + tree = build_menu_tree(menu_items, order) + + new_payload = { + "portal": { + "id": portal_id, + "menuSchema": schema, + }, + "generatedAt": utc_now_iso(), + "count": len(entries), + "files": entries, + "menu": tree, + } + stats = {"updated": updated, "kept": kept, "removed": removed} + return new_payload, stats, len(files) + diff --git a/portal_meta_builder/io_utils.py b/portal_meta_builder/io_utils.py new file mode 100644 index 0000000..f9b9bec --- /dev/null +++ b/portal_meta_builder/io_utils.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict + + +def utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def load_json(path: Path) -> Dict[str, Any]: + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +def save_json(path: Path, payload: Dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, indent=2, sort_keys=False) + "\n", encoding="utf-8" + ) + + +def file_fingerprint(path: Path) -> Dict[str, int]: + st = path.stat() + return {"size": int(st.st_size), "mtime": int(st.st_mtime)} + diff --git a/portal_meta_builder/metadata.py b/portal_meta_builder/metadata.py new file mode 100644 index 0000000..77c2305 --- /dev/null +++ b/portal_meta_builder/metadata.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from netCDF4 import Dataset, num2date + + +SCENARIO_LABELS_FROM_SSP: Dict[str, str] = { + "historical,ssp126": "historical,rcp26", + "historical,ssp245": "historical,rcp45", + "historical,ssp585": "historical,rcp85", + "historical+ssp126": "historical,rcp26", + "historical+ssp245": "historical,rcp45", + "historical+ssp585": "historical,rcp85", +} + +SCENARIO_LABELS_FROM_RCP: Dict[str, str] = { + "historical,rcp26": "historical,rcp26", + "historical,rcp45": "historical,rcp45", + "historical,rcp85": "historical,rcp85", + "historical+rcp26": "historical,rcp26", + "historical+rcp45": "historical,rcp45", + "historical+rcp85": "historical,rcp85", +} + +SCENARIO_LABELS_FROM_SSP_U6: Dict[str, str] = { + "historical,ssp126": "Historical, SSP1-2.6", + "historical,ssp245": "Historical, SSP2-4.5", + "historical,ssp370": "Historical, SSP3-7.0", + "historical,ssp585": "Historical, SSP5-8.5", + "historical+ssp126": "Historical, SSP1-2.6", + "historical+ssp245": "Historical, SSP2-4.5", + "historical+ssp370": "Historical, SSP3-7.0", + "historical+ssp585": "Historical, SSP5-8.5", +} + + +def parse_year(value: str) -> Optional[int]: + if not value: + return None + match = re.search(r"(\d{4})", str(value)) + if not match: + return None + return int(match.group(1)) + + +def choose_primary_variable(ds: Dataset) -> Optional[str]: + coord_names = { + "time", + "lat", + "latitude", + "lon", + "longitude", + "bnds", + "climatology_bnds", + "climatology_bounds", + "height", + } + candidates: List[str] = [] + for name, variable in ds.variables.items(): + lower_name = name.lower() + if lower_name in coord_names: + continue + dims = [dimension.lower() for dimension in getattr(variable, "dimensions", ())] + if len(dims) < 2: + continue + has_lat = any(dimension in ("lat", "latitude") for dimension in dims) + has_lon = any(dimension in ("lon", "longitude") for dimension in dims) + if has_lat and has_lon: + candidates.append(name) + + if candidates: + return candidates[0] + + for name, variable in ds.variables.items(): + if getattr(variable, "ndim", 0) >= 2 and name.lower() not in coord_names: + return name + return None + + +def first_non_empty(values: Sequence[Any]) -> str: + for value in values: + text = str(value or "").strip() + if text: + return text + return "" + + +def normalize_scenario_label(raw: str, scenario_style: str) -> str: + value = str(raw or "").strip() + if not value: + return "Unknown" + if scenario_style == "ssp_u6" and value in SCENARIO_LABELS_FROM_SSP_U6: + return SCENARIO_LABELS_FROM_SSP_U6[value] + if value in SCENARIO_LABELS_FROM_SSP: + return SCENARIO_LABELS_FROM_SSP[value] + if value in SCENARIO_LABELS_FROM_RCP: + return SCENARIO_LABELS_FROM_RCP[value] + return value.replace("+", ",") + + +def normalize_run_label(raw: str, preserve_forcing: bool = False) -> str: + value = str(raw or "").strip() + if not value: + return "Unknown" + if preserve_forcing: + return value + match = re.match(r"^r(?P\d+)i(?P\d+)p(?P

\d+)(?:f\d+)?$", value) + if match: + return f"r{match.group('r')}i{match.group('i')}p{match.group('p')}" + return value + + +def compose_run_from_attrs(global_attrs: Dict[str, Any], prefixes: Sequence[str]) -> str: + for prefix in prefixes: + realization = str(global_attrs.get(f"{prefix}realization_index") or "").strip() + initialization = str( + global_attrs.get(f"{prefix}initialization_index") or "" + ).strip() + physics = str(global_attrs.get(f"{prefix}physics_index") or "").strip() + forcing = str(global_attrs.get(f"{prefix}forcing_index") or "").strip() + if realization and initialization and physics: + run = f"r{realization}i{initialization}p{physics}" + if forcing: + run += f"f{forcing}" + return run + return str(global_attrs.get("run") or "").strip() + + +def derive_common_fields(metadata: Dict[str, Any]) -> Dict[str, Any]: + global_attrs = metadata.get("global", {}) + primary = metadata.get("primary", {}) + time_meta = metadata.get("time", {}) + + scenario_raw = first_non_empty( + [ + global_attrs.get("GCM__experiment_id"), + global_attrs.get("downscaling__GCM__experiment_id"), + global_attrs.get("experiment_id"), + ] + ) + model = first_non_empty( + [ + global_attrs.get("GCM__model_id"), + global_attrs.get("downscaling__GCM__model_id"), + global_attrs.get("model_id"), + ] + ) + run_raw = compose_run_from_attrs( + global_attrs, + prefixes=["GCM__", "downscaling__GCM__", ""], + ) + + climo_start = parse_year(str(global_attrs.get("climo_start_time") or "")) + climo_end = parse_year(str(global_attrs.get("climo_end_time") or "")) + + if climo_start is None: + climo_start = time_meta.get("startYear") + if climo_end is None: + climo_end = time_meta.get("endYear") + + return { + "variableCode": str(primary.get("name") or "").strip(), + "timeCount": int(time_meta.get("count", 0) or 0), + "startYear": time_meta.get("startYear"), + "endYear": time_meta.get("endYear"), + "climoStartYear": climo_start, + "climoEndYear": climo_end, + "frequency": str(global_attrs.get("frequency") or "").strip(), + "scenarioRaw": scenario_raw, + "model": model, + "runRaw": run_raw, + "forcingType": str(global_attrs.get("forcing_type") or "").strip(), + "targetDatasetId": first_non_empty( + [ + global_attrs.get("target__dataset_id"), + global_attrs.get("downscaling__target__dataset_id"), + global_attrs.get("observations__dataset_id"), + global_attrs.get("model_cal__dataset_id"), + ] + ), + "projectId": str(global_attrs.get("project_id") or "").strip(), + "methodId": str(global_attrs.get("method_id") or "").strip(), + } + + +def ensure_derived_fields(metadata: Dict[str, Any]) -> Dict[str, Any]: + derived = metadata.get("derived") + if isinstance(derived, dict): + return metadata + metadata["derived"] = derive_common_fields(metadata) + return metadata + + +def read_netcdf_metadata(path: Path) -> Dict[str, Any]: + metadata: Dict[str, Any] = { + "global": {}, + "time": {}, + "primary": {}, + } + + with Dataset(path) as ds: + for attr in ds.ncattrs(): + try: + value = getattr(ds, attr) + if isinstance(value, (str, int, float, bool)): + metadata["global"][attr] = value + else: + metadata["global"][attr] = str(value) + except Exception: + continue + + primary_name = choose_primary_variable(ds) + if primary_name and primary_name in ds.variables: + variable = ds.variables[primary_name] + metadata["primary"] = { + "name": primary_name, + "units": getattr(variable, "units", None), + "standard_name": getattr(variable, "standard_name", None), + "long_name": getattr(variable, "long_name", None), + "cell_methods": getattr(variable, "cell_methods", None), + } + + if "time" in ds.variables: + time_var = ds.variables["time"] + metadata["time"]["count"] = int(len(time_var)) + units = getattr(time_var, "units", None) + calendar = getattr(time_var, "calendar", "standard") + if units and len(time_var) > 0: + try: + start = num2date(time_var[0], units=units, calendar=calendar) + end = num2date(time_var[-1], units=units, calendar=calendar) + metadata["time"]["start"] = str(start) + metadata["time"]["end"] = str(end) + metadata["time"]["startYear"] = int(start.year) + metadata["time"]["endYear"] = int(end.year) + except Exception: + pass + + return ensure_derived_fields(metadata) + diff --git a/portal_meta_builder/minmax.py b/portal_meta_builder/minmax.py new file mode 100644 index 0000000..eb94ab6 --- /dev/null +++ b/portal_meta_builder/minmax.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import csv +from pathlib import Path +from typing import Any, Dict, List, Optional + + +def parse_float_safe(value: str) -> Optional[float]: + try: + parsed = float(str(value).strip()) + except Exception: + return None + if not (parsed == parsed): + return None + return parsed + + +def is_precip_variable(var_name: str) -> bool: + value = str(var_name or "").strip().lower() + return value in {"pr", "ppt", "prec", "precip", "precipitation", "rainf"} + + +def looks_like_netcdf_ref(value: str) -> bool: + text = str(value or "").strip() + if not text: + return False + lowered = text.lower() + return lowered.endswith(".nc") or lowered.endswith(".nc4") or "/" in text + + +def load_minmax_csv(path: Path) -> Dict[str, List[Dict[str, Any]]]: + if not path.exists(): + return {} + + out: Dict[str, List[Dict[str, Any]]] = {} + with path.open("r", encoding="utf-8", newline="") as handle: + reader = csv.reader(handle) + for row in reader: + if len(row) < 4: + continue + + source_path = "" + var_name = "" + min_value: Optional[float] = None + max_value: Optional[float] = None + + if len(row) >= 5 and looks_like_netcdf_ref(row[1]): + source_path = str(row[1] or "").strip() + var_name = str(row[2] or "").strip() + min_value = parse_float_safe(row[3]) + max_value = parse_float_safe(row[4]) + elif len(row) >= 4 and looks_like_netcdf_ref(row[0]): + source_path = str(row[0] or "").strip() + var_name = str(row[1] or "").strip() + min_value = parse_float_safe(row[2]) + max_value = parse_float_safe(row[3]) + elif len(row) >= 5: + source_path = str(row[1] or "").strip() + var_name = str(row[2] or "").strip() + min_value = parse_float_safe(row[3]) + max_value = parse_float_safe(row[4]) + else: + continue + + if not source_path or min_value is None or max_value is None: + continue + + ensemble = str(row[0] or "").strip() + record = { + "min": min_value, + "max": max_value, + "variable": var_name, + "logScale": is_precip_variable(var_name), + "source": str(path.name), + "ensemble": ensemble, + } + out.setdefault(source_path, []).append(record) + out.setdefault(Path(source_path).name, []).append(record) + return out + + +def select_minmax_record( + lookup: Dict[str, List[Dict[str, Any]]], + source_key: str, + basename: str, + portal_id: str, +) -> Optional[Dict[str, Any]]: + candidates = list(lookup.get(source_key, [])) + list(lookup.get(basename, [])) + if not candidates: + return None + + wanted = str(portal_id or "").strip().lower() + for record in candidates: + ensemble = str(record.get("ensemble") or "").strip().lower() + if ensemble and ensemble == wanted: + return record + return candidates[0] + diff --git a/portal_meta_builder/portals.py b/portal_meta_builder/portals.py new file mode 100644 index 0000000..61c0bd6 --- /dev/null +++ b/portal_meta_builder/portals.py @@ -0,0 +1,378 @@ +from __future__ import annotations + +from typing import Any, Callable, Dict, List, Optional, Set, Tuple + +from .metadata import ensure_derived_fields, normalize_run_label, normalize_scenario_label + + +PRISM_PERIOD_LABELS: Dict[Tuple[int, int], str] = { + (1950, 2007): "Averaged months, 1950-2007", + (1971, 2000): "Climatological averages 1970-2000", + (1981, 2010): "Climatological averages 1981-2010", + (1991, 2020): "Climatological averages 1991-2020", +} + +PRISM_VARIABLE_LABELS: Dict[str, str] = { + "tasmax": "Maximum Temperature", + "tmax": "Maximum Temperature", + "tasmin": "Minimum Temperature", + "tmin": "Minimum Temperature", + "pr": "Total Precipitation", + "ppt": "Total Precipitation", +} + +CANADA_MOSAIC_VARIABLE_LABELS: Dict[str, str] = { + "pr": "Total Precipitation", + "tmax": "Maximum Temperature", + "tmin": "Minimum Temperature", + "tas": "Mean Temperature", + "tasmean": "Mean Temperature", +} + +GRIDDED_DAILY_SOURCE_BY_MODEL: Dict[str, str] = { + "ANUSPLIN_CDA_v2012.1": "NRCANmet 2012", + "TPS_NWNA_v1": "PNWNAmet 2015", + "PCIC_BLEND_v1": "PCIC Blend 2021", +} + +GRIDDED_DAILY_VARIABLE_LABELS: Dict[str, str] = { + "tasmax": "Maximum Temperature", + "tmax": "Maximum Temperature", + "tasmin": "Minimum Temperature", + "tmin": "Minimum Temperature", + "pr": "Precipitation", + "wind": "Wind", +} + +VICGL_VARIABLE_LABELS: Dict[str, str] = { + "BASEFLOW": "Baseflow", + "EVAP": "Evapotranspiration", + "GLAC_AREA": "Glacier Area", + "GLAC_MBAL": "Glacier Mass Balance", + "GLAC_OUTFLOW": "Glacier Outflow", + "PET_NATVEG": "Potential Evapotranspiration", + "PREC": "Precipitation", + "RAINF": "Rainfall", + "SNOW_MELT": "Snow Melt", + "SWE": "Snow Water Equivalent", + "RUNOFF": "Surface Runoff", + "SOIL_MOIST_TOT": "Total Column Soil Moisture", + "TRANSP_VEG": "Transpiration", +} + +PCIC12_MODELS: Set[str] = { + "BCC-CSM2-MR", + "NorESM2-LM", + "MIROC-ES2L", + "MPI-ESM1-2-HR", + "MRI-ESM2-0", + "UKESM1-0-LL", + "EC-Earth3-Veg", + "CMCC-ESM2", + "INM-CM5-0", + "FGOALS-g3", + "TaiESM1", + "IPSL-CM6A-LR", +} + + +MenuBuilder = Callable[[Dict[str, Any], Dict[str, Any]], Dict[str, str]] + + +def prism_period_label(start_year: Optional[int], end_year: Optional[int]) -> str: + if start_year is None or end_year is None: + return "Unknown" + if (start_year, end_year) in PRISM_PERIOD_LABELS: + return PRISM_PERIOD_LABELS[(start_year, end_year)] + return f"{start_year}-{end_year}" + + +def prism_frequency_label(metadata: Dict[str, Any]) -> str: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + frequency = str(derived.get("frequency") or "").lower() + if any(token in frequency for token in ["mclim", "mon", "month"]): + return "Monthly" + if any(token in frequency for token in ["aclim", "ann", "year", "yr"]): + return "Annual" + + time_count = int(derived.get("timeCount", 0) or 0) + if time_count == 12: + return "Monthly" + if time_count == 1: + return "Annual" + if time_count > 12: + return "Monthly" + return "Unknown" + + +def prism_menu_builder(metadata: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, str]: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + variable_code = str(derived.get("variableCode") or "").lower() + variable_label = PRISM_VARIABLE_LABELS.get(variable_code, variable_code or "Unknown") + return { + "period": prism_period_label( + derived.get("climoStartYear"), derived.get("climoEndYear") + ), + "frequency": prism_frequency_label(metadata), + "variable": variable_label, + } + + +def canada_mosaic_menu_builder( + metadata: Dict[str, Any], config: Dict[str, Any] +) -> Dict[str, str]: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + time_count = int(derived.get("timeCount", 0) or 0) + if time_count == 12: + frequency = "Monthly" + elif time_count == 4: + frequency = "Seasonal" + elif time_count == 1: + frequency = "Annual" + else: + frequency = "Unknown" + + variable_code = str(derived.get("variableCode") or "").lower() + variable = CANADA_MOSAIC_VARIABLE_LABELS.get(variable_code, variable_code or "Unknown") + return { + "period": "Climatological averages 1981-2010", + "frequency": frequency, + "variable": variable, + } + + +def gridded_daily_menu_builder( + metadata: Dict[str, Any], config: Dict[str, Any] +) -> Dict[str, str]: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + model_id = str(derived.get("model") or "").strip() + source = GRIDDED_DAILY_SOURCE_BY_MODEL.get(model_id, model_id or "Unknown") + variable_code = str(derived.get("variableCode") or "").lower() + variable = GRIDDED_DAILY_VARIABLE_LABELS.get(variable_code, variable_code or "Unknown") + return {"source": source, "variable": variable} + + +def vicgl_menu_builder(metadata: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, str]: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + global_attrs = metadata.get("global", {}) + + variable_code = str(derived.get("variableCode") or "").upper() + variable = VICGL_VARIABLE_LABELS.get(variable_code, variable_code or "Unknown") + + scenario_raw = str(derived.get("scenarioRaw") or "").lower() + forcing_type = str(derived.get("forcingType") or "").lower() + model_id = str(derived.get("model") or "").strip() + run_raw = str(derived.get("runRaw") or "").strip() + target_dataset = str(derived.get("targetDatasetId") or "").strip() + + if forcing_type == "gridded observations": + scenario = str(global_attrs.get("experiment_id") or "historical") + model = "PNWNAmet base" if target_dataset == "PNWNAmet" else (model_id or "Unknown") + return {"scenario": scenario, "model": model, "variable": variable} + + scenario = normalize_scenario_label(scenario_raw, "legacy") + run = normalize_run_label(run_raw) + model = model_id or "Unknown" + if run != "Unknown": + model = f"{model} {run}" + return {"scenario": scenario or "Unknown", "model": model, "variable": variable} + + +def climate_projection_menu_builder( + metadata: Dict[str, Any], config: Dict[str, Any] +) -> Dict[str, str]: + metadata = ensure_derived_fields(metadata) + derived = metadata.get("derived", {}) + scenario_style = str(config.get("scenarioStyle") or "legacy") + preserve_forcing = bool(config.get("preserveRunForcing")) + + scenario = normalize_scenario_label( + str(derived.get("scenarioRaw") or "").lower(), + scenario_style, + ) + model = str(derived.get("model") or "").strip() or "Unknown" + run = normalize_run_label(str(derived.get("runRaw") or ""), preserve_forcing) + variable = str(derived.get("variableCode") or "").lower() or "unknown" + + fields = { + "scenario": scenario, + "model": model, + "run": run, + "variable": variable, + } + if config.get("addPcic12ScenarioSuffix") and model in PCIC12_MODELS and scenario in { + "Historical, SSP1-2.6", + "Historical, SSP2-4.5", + "Historical, SSP3-7.0", + "Historical, SSP5-8.5", + }: + fields["scenarioPcic12"] = f"{scenario} (PCIC12)" + return fields + + +PORTAL_CONFIGS: Dict[str, Dict[str, Any]] = { + "prism": { + "menuSchema": { + "order": ["period", "frequency", "variable"], + "labels": { + "period": "Period", + "frequency": "Frequency", + "variable": "Variable", + }, + }, + "menuBuilder": prism_menu_builder, + }, + "canada_mosaic": { + "menuSchema": { + "order": ["period", "frequency", "variable"], + "labels": { + "period": "Period", + "frequency": "Frequency", + "variable": "Variable", + }, + }, + "menuBuilder": canada_mosaic_menu_builder, + }, + "gridded_daily": { + "menuSchema": { + "order": ["source", "variable"], + "labels": { + "source": "Dataset", + "variable": "Variable", + }, + }, + "menuBuilder": gridded_daily_menu_builder, + }, + "vicgl": { + "menuSchema": { + "order": ["scenario", "model", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "variable": "Variable", + }, + }, + "menuBuilder": vicgl_menu_builder, + }, + "bccaqv2": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "legacy", + }, + "bccaqv2_u5": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "legacy", + }, + "bccaqv2_u6": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "ssp_u6", + "addPcic12ScenarioSuffix": True, + }, + "canesm5_u6": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "ssp_u6", + "preserveRunForcing": True, + }, + "canesm5_m6": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "ssp_u6", + "preserveRunForcing": True, + }, + "mbcn": { + "menuSchema": { + "order": ["scenario", "model", "run", "variable"], + "labels": { + "scenario": "Scenario", + "model": "Model", + "run": "Run", + "variable": "Variable", + }, + }, + "menuBuilder": climate_projection_menu_builder, + "scenarioStyle": "ssp_u6", + }, +} + + +DEFAULT_PORTAL_CONFIG: Dict[str, Any] = { + "menuSchema": {"order": ["variable"], "labels": {"variable": "Variable"}}, + "menuBuilder": lambda metadata, config: { + "variable": str(metadata.get("derived", {}).get("variableCode") or "Unknown") + }, +} + + +def get_portal_config(portal_id: str) -> Dict[str, Any]: + return PORTAL_CONFIGS.get(portal_id, DEFAULT_PORTAL_CONFIG) + + +def derive_menu_fields(portal_id: str, metadata: Dict[str, Any]) -> Dict[str, str]: + config = get_portal_config(portal_id) + builder = config["menuBuilder"] + return builder(metadata, config) + + +def build_menu_tree( + file_items: List[Tuple[str, Dict[str, str]]], order: List[str] +) -> Dict[str, Any]: + tree: Dict[str, Any] = {} + for basename, fields in file_items: + current: Dict[str, Any] = tree + for key in order[:-1]: + label = str(fields.get(key) or "Unknown") + if label not in current: + current[label] = {} + current = current[label] + leaf = str(fields.get(order[-1]) or "Unknown") + current.setdefault(leaf, []).append(basename) + return tree + diff --git a/scripts/calculate-portal-minmax.py b/scripts/calculate-portal-minmax.py new file mode 100644 index 0000000..74ba1f3 --- /dev/null +++ b/scripts/calculate-portal-minmax.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import csv +import fnmatch +import glob +from pathlib import Path +from typing import Dict, Iterable, List, Optional, Set, Tuple + +import numpy as np +from netCDF4 import Dataset + + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def read_portals(path: Path) -> Tuple[List[str], List[str]]: + includes: List[str] = [] + excludes: List[str] = [] + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + if line.startswith("!"): + excludes.append(line[1:].strip()) + else: + includes.append(line) + return includes, excludes + + +def expand_globs(includes: Iterable[str], excludes: Iterable[str]) -> List[Path]: + files: List[Path] = [] + for selector in includes: + if any(ch in selector for ch in ["*", "?", "["]): + files.extend(Path(p) for p in glob.glob(selector, recursive=True)) + else: + files.append(Path(selector)) + + uniq: List[Path] = [] + seen: Set[str] = set() + for p in files: + key = str(p) + if key in seen: + continue + seen.add(key) + if p.exists() and p.is_file(): + uniq.append(p) + + out: List[Path] = [] + for p in uniq: + source = str(p) + if any(fnmatch.fnmatch(source, ex) for ex in excludes): + continue + out.append(p) + return sorted(out, key=lambda p: p.name.lower()) + + +def choose_primary_variable(ds: Dataset) -> Optional[str]: + coord = {"time", "lat", "latitude", "lon", "longitude", "bnds", "climatology_bnds", "height"} + candidates: List[str] = [] + for name, variable in ds.variables.items(): + if name.lower() in coord: + continue + dims = [d.lower() for d in getattr(variable, "dimensions", ())] + if len(dims) < 2: + continue + has_lat = any(d in ("lat", "latitude") for d in dims) + has_lon = any(d in ("lon", "longitude") for d in dims) + if has_lat and has_lon: + candidates.append(name) + if candidates: + return candidates[0] + for name, variable in ds.variables.items(): + if getattr(variable, "ndim", 0) >= 2 and name.lower() not in coord: + return name + return None + + +def min_max_for_file(path: Path) -> Optional[Tuple[str, float, float]]: + with Dataset(path) as ds: + var_name = choose_primary_variable(ds) + if not var_name or var_name not in ds.variables: + return None + variable = ds.variables[var_name] + data = variable[:] + fill = getattr(variable, "_FillValue", None) + missing = getattr(variable, "missing_value", None) + arr = np.array(data, dtype="float64") + mask = ~np.isfinite(arr) + if fill is not None: + mask |= arr == float(fill) + if missing is not None: + try: + mask |= arr == float(missing) + except Exception: + pass + valid = arr[~mask] + if valid.size == 0: + return None + return var_name, float(np.min(valid)), float(np.max(valid)) + + +def load_existing_rows(path: Path) -> Dict[str, List[str]]: + if not path.exists(): + return {} + out: Dict[str, List[str]] = {} + with path.open("r", encoding="utf-8", newline="") as handle: + reader = csv.reader(handle) + for row in reader: + if len(row) < 5: + continue + source = str(row[1] or "").strip() + if source: + out[source] = row + return out + + +def main() -> int: + parser = argparse.ArgumentParser(description="Compute min/max for portal file-pattern definitions.") + parser.add_argument("--portal", action="append", help="portal id(s), e.g. canada_mosaic") + parser.add_argument("--portals-dir", default=str(REPO_ROOT / "portal-prep" / "portal-file-patterns")) + parser.add_argument("--out-csv", default=str(REPO_ROOT / "portal-prep" / "pdp_min_max.csv")) + parser.add_argument("--all", action="store_true", help="recompute all files, not just missing rows") + args = parser.parse_args() + + portals_dir = Path(args.portals_dir).resolve() + out_csv = Path(args.out_csv).resolve() + portals = [p for p in (args.portal or []) if p] + if not portals: + portals = sorted(p.stem for p in portals_dir.glob("*.txt")) + + existing = load_existing_rows(out_csv) + rows: Dict[str, List[str]] = {} if args.all else dict(existing) + + processed = 0 + added = 0 + failed = 0 + + for portal_id in portals: + selector_file = portals_dir / f"{portal_id}.txt" + if not selector_file.exists(): + print(f"{portal_id}: missing portal pattern file, skipped") + continue + includes, excludes = read_portals(selector_file) + files = expand_globs(includes, excludes) + for src in files: + source_path = str(src) + if (not args.all) and source_path in rows: + continue + processed += 1 + try: + result = min_max_for_file(src) + if not result: + failed += 1 + continue + var_name, min_value, max_value = result + rows[source_path] = [portal_id, source_path, var_name, f"{min_value:.8g}", f"{max_value:.8g}"] + added += 1 + except Exception: + failed += 1 + + out_csv.parent.mkdir(parents=True, exist_ok=True) + with out_csv.open("w", encoding="utf-8", newline="") as handle: + writer = csv.writer(handle) + for source in sorted(rows.keys()): + writer.writerow(rows[source]) + + print(f"portals={len(portals)} processed={processed} added_or_updated={added} failed={failed} total_rows={len(rows)}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/scripts/chunk_gridded.sh b/scripts/chunk_gridded.sh new file mode 100644 index 0000000..582a5e3 --- /dev/null +++ b/scripts/chunk_gridded.sh @@ -0,0 +1,49 @@ +#!/bin/sh +set -eu + +DIR="/storage/..../gridded_daily" +OUT_DIR="/storage/..../gridded_chunked" + +TIME_CHUNK="90" +LAT_CHUNK="64" +LON_CHUNK="64" + +# Set to 1 to overwrite existing outputs +OVERWRITE="${OVERWRITE:-0}" + +if [ ! -d "$DIR" ]; then + echo "ERROR: directory not found: $DIR" >&2 + exit 1 +fi + +mkdir -p "$OUT_DIR" + +find "$DIR" -maxdepth 1 -type f -name "*.nc" -print0 | +while IFS= read -r -d '' in; do + base=$(basename "$in") + + # Skip already chunked inputs + case "$base" in + chunked_*) continue ;; + esac + + out="${OUT_DIR}/chunked_${base}" + + if [ -e "$out" ] && [ "$OVERWRITE" != "1" ]; then + echo "SKIP: output exists: $out" + continue + fi + + echo "CHUNK: $base -> $(basename "$out")" + + nccopy -k 4 -d 0 \ + -c time/$TIME_CHUNK,lat/$LAT_CHUNK,lon/$LON_CHUNK \ + "$in" "$out" + + if [ ! -s "$out" ]; then + echo "ERROR: output missing/empty: $out" >&2 + exit 1 + fi +done + +echo "Done." diff --git a/scripts/gen_hardlinks.py b/scripts/gen_hardlinks.py new file mode 100644 index 0000000..b087d8c --- /dev/null +++ b/scripts/gen_hardlinks.py @@ -0,0 +1,294 @@ +#!/usr/bin/env python3 + +import argparse +import errno +import fnmatch +import glob +import os +from pathlib import Path +from typing import Dict, List, Sequence, Set, Tuple + + +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_MIRROR_ROOT = Path("/storage/data/projects/comp_support/thredds/pdp") +MANIFEST_NAME = ".gen_hardlinks_manifest.txt" + +ALLOWED_SOURCE_PREFIXES = { + Path("/storage/data/climate").resolve(), + Path("/storage/data/projects/hydrology").resolve(), + Path("/storage/data/projects/dataportal").resolve(), + Path("/storage/data/projects/comp_support").resolve(), +} + + +def read_file_list(path: Path) -> Tuple[List[str], List[str]]: + includes: List[str] = [] + excludes: List[str] = [] + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line or line.startswith("#"): + continue + if line.startswith("!"): + excludes.append(line[1:].strip()) + else: + includes.append(line) + return includes, excludes + + +def is_glob_pattern(value: str) -> bool: + return any(ch in value for ch in ["*", "?", "["]) + + +def expand_file_list(includes: Sequence[str], excludes: Sequence[str]) -> List[Path]: + out: List[Path] = [] + for include_pattern in includes: + if is_glob_pattern(include_pattern): + out.extend(Path(p) for p in glob.glob(include_pattern, recursive=True)) + else: + out.append(Path(include_pattern)) + + seen: Set[str] = set() + unique: List[Path] = [] + for path in out: + resolved = path.resolve() + key = str(resolved) + if key in seen: + continue + seen.add(key) + unique.append(resolved) + + files: List[Path] = [] + for path in unique: + if not path.exists() or not path.is_file(): + continue + files.append(path) + + exclude_patterns = [p for p in excludes if p] + if exclude_patterns: + filtered: List[Path] = [] + for path in files: + path_str = str(path) + if any(fnmatch.fnmatch(path_str, pat) for pat in exclude_patterns): + continue + filtered.append(path) + files = filtered + + return sorted(files, key=lambda p: str(p).lower()) + + +def manifest_path(mirror_root: Path) -> Path: + return mirror_root / MANIFEST_NAME + + +def read_manifest(mirror_root: Path) -> Set[Path]: + p = manifest_path(mirror_root) + if not p.exists(): + return set() + out: Set[Path] = set() + for raw in p.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line: + continue + out.add(Path(line)) + return out + + +def write_manifest(mirror_root: Path, managed_paths: Sequence[Path]) -> None: + p = manifest_path(mirror_root) + lines = [str(path) for path in sorted(managed_paths, key=lambda item: str(item).lower())] + p.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8") + + +def prune_from_manifest(mirror_root: Path) -> Tuple[int, int, int]: + if not mirror_root.exists(): + return 0, 0, 0 + + managed = read_manifest(mirror_root) + removed_files = 0 + removed_dirs = 0 + + for rel in sorted(managed, key=lambda p: (len(p.parts), str(p)), reverse=True): + target = mirror_root / rel + if target.is_file() or target.is_symlink(): + target.unlink() + removed_files += 1 + + for path in sorted(mirror_root.rglob("*"), key=lambda p: (len(p.parts), str(p)), reverse=True): + if path.is_dir(): + try: + path.rmdir() + removed_dirs += 1 + except OSError: + pass + + unmanaged = 0 + for path in mirror_root.rglob("*"): + if path.name == MANIFEST_NAME: + continue + if path.is_file() or path.is_symlink(): + unmanaged += 1 + + return removed_files, removed_dirs, unmanaged + + +def ensure_hardlink(dst: Path, src: Path) -> str: + dst.parent.mkdir(parents=True, exist_ok=True) + + if dst.exists(): + if dst.is_dir(): + raise SystemExit(f"Refusing to replace directory with hardlink: {dst}") + if os.path.samefile(dst, src): + return "kept" + dst.unlink() + try: + os.link(src, dst) + except OSError as exc: + if exc.errno == errno.EXDEV: + raise SystemExit( + f"Cross-device hardlink is not possible.\nsource: {src}\ntarget: {dst}\n" + "Put --mirror-root on the same filesystem as source data." + ) from exc + raise + return "updated" + + try: + os.link(src, dst) + except OSError as exc: + if exc.errno == errno.EXDEV: + raise SystemExit( + f"Cross-device hardlink is not possible.\nsource: {src}\ntarget: {dst}\n" + "Put --mirror-root on the same filesystem as source data." + ) from exc + raise + return "created" + + +def enforce_directory_mode(root: Path, mode: int = 0o755) -> None: + root.mkdir(parents=True, exist_ok=True) + os.chmod(root, mode) + for path in root.rglob("*"): + if path.is_dir(): + os.chmod(path, mode) + + +def validate_source_path(src: Path) -> None: + if not any(src.is_relative_to(prefix) for prefix in ALLOWED_SOURCE_PREFIXES): + allowed = ", ".join(sorted(str(p) for p in ALLOWED_SOURCE_PREFIXES)) + raise SystemExit( + f"Source path is outside allowed prefixes: {src}\nAllowed: {allowed}" + ) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Rebuild a hardlink mirror as /pdp//.nc from portal file-pattern definitions." + ) + parser.add_argument( + "--patterns-dir", + default=str(REPO_ROOT / "portal-prep" / "portal-file-patterns"), + help="Directory containing per-portal file-pattern files (.txt).", + ) + parser.add_argument( + "--mirror-root", + default=str(DEFAULT_MIRROR_ROOT), + help="Output root for hardlinks; links are written to //.", + ) + parser.add_argument( + "--portal", + action="append", + help="Portal id(s). If omitted, process all pattern files in --patterns-dir.", + ) + parser.add_argument( + "--no-clean", + action="store_true", + help="Do not clean mirror root before creating hardlinks.", + ) + args = parser.parse_args() + + patterns_dir = Path(args.patterns_dir).resolve() + mirror_root = Path(args.mirror_root).resolve() + + portal_ids = [p for p in (args.portal or []) if p] + if not portal_ids: + portal_ids = sorted(path.stem for path in patterns_dir.glob("*.txt")) + + file_lists = [patterns_dir / f"{portal_id}.txt" for portal_id in portal_ids] + missing_lists = [path for path in file_lists if not path.exists()] + if missing_lists: + lines = "\n".join(f" - {path}" for path in missing_lists) + raise SystemExit(f"Missing portal pattern files:\n{lines}") + if not file_lists: + raise SystemExit(f"No portal file-pattern files found under {patterns_dir}") + + wanted: Dict[Path, Path] = {} + all_matches = 0 + per_portal_counts: Dict[str, int] = {} + for file_list in file_lists: + portal_id = file_list.stem + includes, excludes = read_file_list(file_list) + files = expand_file_list(includes, excludes) + all_matches += len(files) + per_portal_counts[portal_id] = len(files) + + for src in files: + validate_source_path(src) + if src.is_relative_to(mirror_root): + raise SystemExit(f"Source path is inside mirror root, refusing: {src}") + + dst = mirror_root / portal_id / src.name + existing = wanted.get(dst) + if existing is not None and existing != src: + raise SystemExit( + f"Conflicting sources map to same mirror path:\n" + f" dst: {dst}\n" + f" first: {existing}\n" + f" second: {src}\n" + f"Filenames must be unique per portal." + ) + wanted[dst] = src + + removed_files = 0 + removed_dirs = 0 + unmanaged_files = 0 + if not args.no_clean: + removed_files, removed_dirs, unmanaged_files = prune_from_manifest(mirror_root) + + mirror_root.mkdir(parents=True, exist_ok=True) + + created = 0 + updated = 0 + kept = 0 + for dst, src in sorted(wanted.items(), key=lambda item: str(item[0]).lower()): + status = ensure_hardlink(dst, src) + if status == "created": + created += 1 + elif status == "updated": + updated += 1 + else: + kept += 1 + + managed_rel = [path.relative_to(mirror_root) for path in wanted.keys()] + write_manifest(mirror_root, managed_rel) + enforce_directory_mode(mirror_root, mode=0o755) + + print(f"portal pattern files: {len(file_lists)}") + for portal_id in sorted(per_portal_counts): + print(f" {portal_id}: {per_portal_counts[portal_id]} files") + print(f"matches: {all_matches} input files") + print(f"wanted: {len(wanted)}") + if not args.no_clean: + print(f"cleaned files: {removed_files}") + print(f"cleaned dirs: {removed_dirs}") + if unmanaged_files: + print(f"unmanaged files kept: {unmanaged_files}") + print(f"created: {created}") + print(f"updated: {updated}") + print(f"kept: {kept}") + print(f"root: {mirror_root}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + diff --git a/scripts/update-portal-meta.py b/scripts/update-portal-meta.py new file mode 100644 index 0000000..ade0c6b --- /dev/null +++ b/scripts/update-portal-meta.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parent.parent +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from portal_meta_builder.builder import build_portal_payload +from portal_meta_builder.io_utils import save_json +from portal_meta_builder.minmax import load_minmax_csv + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Build per-portal metadata + menu trees from /pdp-style mirror datasets." + ) + parser.add_argument( + "--portal", + action="append", + help="portal id(s). If omitted, processes all portals.", + ) + parser.add_argument( + "--mirror-root", + default="/storage/data/projects/comp_support/thredds/pdp", + help="mirror root containing //.nc hardlinks", + ) + parser.add_argument( + "--out-dir", + default=str(REPO_ROOT / "portal-meta"), + help="output directory for .json", + ) + parser.add_argument( + "--minmax-csv", + default=str(REPO_ROOT / "portal-prep" / "pdp_min_max.csv"), + help="optional CSV with precomputed min/max rows (portal, path, variable, min, max)", + ) + parser.add_argument( + "--prune", + action="store_true", + help="drop stale entries not in current inventory", + ) + args = parser.parse_args() + + mirror_root = Path(args.mirror_root).resolve() + out_dir = Path(args.out_dir).resolve() + minmax_path = Path(args.minmax_csv).resolve() + minmax_lookup = load_minmax_csv(minmax_path) + + portals = [p for p in (args.portal or []) if p] + if not portals: + portals = ( + sorted(path.name for path in mirror_root.iterdir() if path.is_dir()) + if mirror_root.exists() + else [] + ) + + if not portals: + raise SystemExit("No portals found to process.") + + for portal_id in portals: + out_path = out_dir / f"{portal_id}.json" + new_payload, stats, inventory_count = build_portal_payload( + portal_id=portal_id, + mirror_root=mirror_root, + out_path=out_path, + minmax_lookup=minmax_lookup, + minmax_source_name=minmax_path.name, + prune=args.prune, + ) + save_json(out_path, new_payload) + + print( + f"{portal_id}: {inventory_count} inventory -> {new_payload['count']} cached " + f"({stats['updated']} updated, {stats['kept']} unchanged, {stats['removed']} removed), " + f"menu top-level={len(new_payload['menu'])}" + ) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) + From e8f678ed4be4eb1262ba304fa807c60bfa0ed060 Mon Sep 17 00:00:00 2001 From: Quintin Date: Fri, 24 Jul 2026 13:16:30 -0700 Subject: [PATCH 2/3] Update minmax script and readme, add sql --- README.md | 53 ++- portal-prep/README.md | 152 ++++++- portal-prep/min_max_query.sql | 27 ++ scripts/calculate-portal-minmax.py | 672 +++++++++++++++++++++++++---- 4 files changed, 793 insertions(+), 111 deletions(-) create mode 100644 portal-prep/min_max_query.sql diff --git a/README.md b/README.md index e5b0626..0f84453 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # PDP-Next -THREDDS-backed, database-free replacement for the Pacific Climate Data Portal (PDP) with projection support, faster subsetting, additional palettes, and portal-based interactive map viewing. +THREDDS-backed, database-free replacement for the Pacific Climate Data Portal (PDP) with projection support, additional palettes, and portal-based interactive map viewing. ## Layout @@ -11,15 +11,39 @@ THREDDS-backed, database-free replacement for the Pacific Climate Data Portal (P - `viewer/`: current frontend/viewer assets - `thredds/`: local THREDDS config and support files +## Production Files + +The production metadata files are deployed under: + +```text +/.../swarm_files/ +``` + +The active generated metadata is in `portal-meta/*.json`. The production +min/max CSV and portal membership definitions are in `portal-prep/`: + +```text +portal-meta/.json +portal-prep/pdp_min_max.csv +portal-prep/portal-file-patterns/.txt +``` + +`portal-meta/portal-meta-bu/` contains backup JSON files and is not an active +metadata output directory. + ## Current Metadata Flow 1. Build or refresh the `/pdp//.nc` hardlink mirror from portal file patterns. -2. Optionally compute min/max rendering metadata for selected files. -3. Build `portal-meta/.json` from NetCDF metadata and normalized derived fields. +2. Export existing min/max ranges from the `modelmeta` database. +3. Build the ordered min/max CSV, scanning only files missing from the export. +4. Build `portal-meta/.json` from NetCDF metadata and normalized derived fields. The metadata builder reads NetCDF metadata, normalizes common fields into `metadata.derived`, and applies per-portal menu rules from `portal_meta_builder/portals.py`. +See [`portal-prep/README.md`](portal-prep/README.md) for the complete, +copy-pasteable metadata preparation workflow. + ## Scripts ### Build hardlink mirror @@ -34,16 +58,26 @@ Uses file-pattern definitions from `portal-prep/portal-file-patterns/`. ### Compute rendering min/max CSV ```bash -python3 scripts/calculate-portal-minmax.py -python3 scripts/calculate-portal-minmax.py --portal prism +export PDP_MODELMETA_URL='postgresql://USERNAME@HOSTNAME:5432/modelmeta' + +psql --no-psqlrc --quiet --csv --dbname "$PDP_MODELMETA_URL" \ + --file portal-prep/min_max_query.sql \ + > portal-prep/db-export.csv + +python3 scripts/calculate-portal-minmax.py \ + --db-csv portal-prep/db-export.csv \ + --out-csv portal-prep/pdp_min_max.csv \ + --all ``` -Writes `portal-prep/pdp_min_max.csv`. +Uses `portal-prep/db-export.csv` as its first source and scans remaining files +in bounded chunks. The reduced export query is +`portal-prep/min_max_query.sql`. Writes `portal-prep/pdp_min_max.csv`. ### Build portal metadata JSON ```bash -python3 scripts/update-portal-meta.py +python3 scripts/update-portal-meta.py --prune python3 scripts/update-portal-meta.py --portal prism ``` @@ -54,7 +88,10 @@ Writes `portal-meta/.json`. 1. Add a portal pattern file at `portal-prep/portal-file-patterns/.txt`. 2. Add a portal definition in `portal_meta_builder/portals.py`. 3. Run `scripts/gen_hardlinks.py` for the portal. -4. Run `scripts/update-portal-meta.py --portal `. +4. If ModelMeta contains the portal's ensemble, add its database ensemble name + to `portal-prep/min_max_query.sql` and regenerate `portal-prep/db-export.csv`. +5. Run `scripts/calculate-portal-minmax.py --portal `. +6. Run `scripts/update-portal-meta.py --portal `. If the portal needs custom menu labels or grouping, add a focused menu builder in `portal_meta_builder/portals.py`. If it needs additional normalized metadata, add that in diff --git a/portal-prep/README.md b/portal-prep/README.md index 2f78722..f89222d 100644 --- a/portal-prep/README.md +++ b/portal-prep/README.md @@ -1,23 +1,147 @@ -Portal prep files +# Portal metadata preparation -`portal-prep/portal-file-patterns/` is the source of truth for which datasets belong to each portal. +Run these commands from the repository root. The complete flow is: -Each portal has a `.txt` file containing one file pattern rule per line: -- Absolute file path, for example `/storage/.../file.nc` -- A glob pattern containing `*`, `?`, or `[` -- An exclude pattern prefixed with `!` +1. Refresh the hardlink mirror from the portal file patterns. +2. Export existing ranges from the `modelmeta` database. +3. Build the ordered portal min/max CSV, calculating only database misses. +4. Rebuild the portal metadata JSON consumed by the viewer. -Rules: -- Blank lines are ignored -- Lines starting with `#` are ignored -- `!(...)` extglob is not supported -- Filenames must remain unique within a portal after expansion because the hardlink mirror uses `/` +## 1. Refresh the hardlink mirror -Primary commands: +The hardlink mirror under `/storage/data/projects/comp_support/thredds/pdp` +must be current before rebuilding the portal metadata. Refresh it when portal +membership or source files have changed: ```bash python3 scripts/gen_hardlinks.py -python3 scripts/calculate-portal-minmax.py -python3 scripts/update-portal-meta.py ``` +## Production locations + +The production copies used by PDP are under: + +```text +/.../swarm_files/pdp-next +``` + +The active files produced or consumed by this workflow are: + +```text +portal-meta/.json +portal-prep/pdp_min_max.csv +portal-prep/portal-file-patterns/.txt +``` + +## 2. Export ranges from modelmeta + +[`min_max_query.sql`](min_max_query.sql) selects only the fields needed by +the range builder and restricts the result to the relevant database +ensembles. Run it against the `modelmeta` database and save the CSV here: + +```bash +export PDP_MODELMETA_URL='postgresql://USERNAME@HOSTNAME:5432/modelmeta' + +psql \ + --no-psqlrc \ + --quiet \ + --csv \ + --dbname "$PDP_MODELMETA_URL" \ + --file portal-prep/min_max_query.sql \ + > portal-prep/db-export.csv +``` + +`PDP_MODELMETA_URL` is the connection string for the `modelmeta` database on +the DM host. Replace `USERNAME` and `HOSTNAME` with its connection values. + +The export must begin with this exact header: + +```text +filename,netcdf_variable_name,range_min,range_max,ensemble_name +``` + +Check it before continuing: + +```bash +head -1 portal-prep/db-export.csv +wc -l portal-prep/db-export.csv +``` + +`db-export.csv` is an intermediate snapshot. Regenerate it whenever ranges or +ensemble membership change in `modelmeta`. + +## 3. Build the ordered min/max CSV + +Pass the database export to the range builder and request a complete refresh: + +```bash +python3 scripts/calculate-portal-minmax.py \ + --db-csv portal-prep/db-export.csv \ + --out-csv portal-prep/pdp_min_max.csv \ + --all +``` + +For every file selected by `portal-file-patterns/*.txt`, the builder: + +1. Looks for an unambiguous range in `db-export.csv`, preferring an exact + source path and then an unambiguous basename match. +2. Reads only the NetCDF header when it must choose among multiple variables. +3. Scans database misses in bounded chunks, avoiding whole-file memory loads. +4. Writes `pdp_min_max.csv` atomically in deterministic portal/path order. +5. Saves progress after every completed NetCDF scan, so a later failure does + not discard completed work. + +The output has no header and uses this layout: + +```text +portal_id,source_path,variable,min,max,source +``` + +The default is one scan worker with a 256 MB chunk budget. Keep one worker on +memory-constrained hosts; every additional worker has its own chunk budget. +Files already represented by the database do not need to be scanned. + +Useful checks: + +```bash +wc -l portal-prep/pdp_min_max.csv +tail -5 portal-prep/pdp_min_max.csv +``` + +## 4. Rebuild portal metadata + +Build the JSON documents from the hardlink mirror and attach the ranges from +the ordered CSV: + +```bash +python3 scripts/update-portal-meta.py \ + --minmax-csv portal-prep/pdp_min_max.csv \ + --prune +``` + +`--prune` removes cached entries that are no longer present in the mirror. The +result is one `portal-meta/.json` file per portal. Each matched file +receives a `rendering` object containing `min`, `max`, `variable`, `logScale`, +and the min/max source. + +For a focused metadata rebuild, repeat `--portal` as needed: + +```bash +python3 scripts/update-portal-meta.py \ + --portal vicgl \ + --minmax-csv portal-prep/pdp_min_max.csv \ + --prune +``` + +## Portal file patterns + +`portal-file-patterns/` is the source of truth for portal membership. Each +`.txt` accepts: + +- An absolute file path +- A glob containing `*`, `?`, or `[` patterns +- An exclusion prefixed with `!` + +Blank lines and lines beginning with `#` are ignored. Bash extglob syntax such +as `!(...)` is not supported. Basenames must be unique within a portal because +the hardlink mirror stores files as `/`. diff --git a/portal-prep/min_max_query.sql b/portal-prep/min_max_query.sql new file mode 100644 index 0000000..b2bcc9b --- /dev/null +++ b/portal-prep/min_max_query.sql @@ -0,0 +1,27 @@ +SELECT + df.filename, + dfv.netcdf_variable_name, + dfv.range_min, + dfv.range_max, + e.ensemble_name +FROM pcic_meta.data_file_variables AS dfv +JOIN pcic_meta.data_files AS df + ON df.data_file_id = dfv.data_file_id +JOIN pcic_meta.ensemble_data_file_variables AS edfv + ON edfv.data_file_variable_id = dfv.data_file_variable_id +JOIN pcic_meta.ensembles AS e + ON e.ensemble_id = edfv.ensemble_id +WHERE dfv.range_min IS NOT NULL + AND dfv.range_max IS NOT NULL + AND e.ensemble_name IN ( + 'bccaq_version_2', + 'bccaq2_canesm5', + 'bccaq2_cmip6', + 'bc_prism_monthly_and_climos', + 'gridded-obs-met-data', + 'hydro_test', + 'mbcn_canesm5', + 'mbcn_cmip6', + 'vicgl_cmip5' + ) +ORDER BY e.ensemble_name, df.filename, dfv.data_file_variable_id; diff --git a/scripts/calculate-portal-minmax.py b/scripts/calculate-portal-minmax.py index 74ba1f3..91d8842 100644 --- a/scripts/calculate-portal-minmax.py +++ b/scripts/calculate-portal-minmax.py @@ -1,175 +1,669 @@ #!/usr/bin/env python3 +""" +Compute portal NetCDF min/max values, DB-first, netCDF4 fallback. + +For each candidate file (selected via portal pattern files), we first look +the filename up in a CSV export of the data_file_variables/data_files/ +variable_aliases/ensembles join (see min_max_query.sql). If the DB already +has usable range_min/range_max for that filename, we reuse it -- no file +read needed. Only files with no usable DB row get scanned, and that scan +is done with netCDF4 + numpy directly (no CDO subprocess), reading in +memory-bounded chunks and reducing with numpy's C-level min/max. + +Usage: + python3 scripts/calculate-portal-minmax.py [--db-csv db_export.csv] +""" + from __future__ import annotations import argparse import csv import fnmatch import glob +from collections import defaultdict +from concurrent.futures import ProcessPoolExecutor, as_completed from pathlib import Path from typing import Dict, Iterable, List, Optional, Set, Tuple import numpy as np from netCDF4 import Dataset - REPO_ROOT = Path(__file__).resolve().parent.parent +RowKey = Tuple[str, str] +MinMaxResult = Tuple[str, float, float] +DbRecord = Tuple[str, float, float, bool, str] +DbLookup = Tuple[Dict[str, List[DbRecord]], Dict[str, List[DbRecord]]] + + +# --------------------------------------------------------------------------- +# DB export lookup +# --------------------------------------------------------------------------- + + +def load_db_lookup(path: Path) -> DbLookup: + """Build exact-path and basename indexes for usable database ranges. + + from the reduced data_file_variables/data_files export produced by + min_max_query.sql. Rows with a missing/NULL range_min or range_max are + skipped -- those still need computing. + """ + exact_lookup: Dict[str, List[DbRecord]] = defaultdict(list) + basename_lookup: Dict[str, List[DbRecord]] = defaultdict(list) + + def is_missing(value: Optional[str]) -> bool: + return value is None or value.strip() == "" or value.strip().upper() == "NULL" + + with path.open("r", encoding="utf-8", newline="") as handle: + reader = csv.DictReader(handle) + + required = { + "filename", + "netcdf_variable_name", + "range_min", + "range_max", + "ensemble_name", + } + missing_cols = required - set(reader.fieldnames or []) + if missing_cols: + raise ValueError( + f"--db-csv is missing expected column(s): {sorted(missing_cols)}" + ) + + for row in reader: + filename = (row.get("filename") or "").strip() + variable_name = (row.get("netcdf_variable_name") or "").strip() + ensemble_name = (row.get("ensemble_name") or "").strip().lower() + + if not filename or not variable_name or not ensemble_name: + continue + + if is_missing(row.get("range_min")) or is_missing(row.get("range_max")): + continue + + try: + range_min = float(row["range_min"]) + range_max = float(row["range_max"]) + except ValueError: + continue + + disabled_raw = (row.get("disabled") or "").strip().lower() + disabled = disabled_raw in ("true", "t", "1") + + record = ( + variable_name, + range_min, + range_max, + disabled, + ensemble_name, + ) + if record not in exact_lookup[filename]: + exact_lookup[filename].append(record) + basename = Path(filename).name + if record not in basename_lookup[basename]: + basename_lookup[basename].append(record) + + return dict(exact_lookup), dict(basename_lookup) + + +def select_db_record( + source: Path, + lookup: DbLookup, + include_disabled: bool, +) -> Optional[DbRecord]: + """Return one unambiguous DB range for a portal source file. + + Exact source paths take precedence. Basename matching supports portal + mirrors and moved source trees. A file containing multiple data variables + requires a cheap NetCDF header read so the selected DB variable agrees + with the portal metadata builder's primary-variable selection. + """ + exact_lookup, basename_lookup = lookup + entries = exact_lookup.get(str(source)) or basename_lookup.get(source.name) or [] + usable = [entry for entry in entries if include_disabled or not entry[3]] + if not usable: + return None + + variable_names = {entry[0].lower() for entry in usable} + if len(variable_names) > 1: + try: + with Dataset(source, mode="r") as dataset: + primary = choose_primary_variable(dataset) + except Exception: + return None + if not primary: + return None + usable = [entry for entry in usable if entry[0].lower() == primary.lower()] + + # Basename collisions or inconsistent duplicate rows are unsafe to guess. + distinct = {(entry[0].lower(), entry[1], entry[2]) for entry in usable} + if len(distinct) != 1: + return None + return usable[0] + + +# --------------------------------------------------------------------------- +# Portal pattern handling +# --------------------------------------------------------------------------- def read_portals(path: Path) -> Tuple[List[str], List[str]]: includes: List[str] = [] excludes: List[str] = [] + for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() + if not line or line.startswith("#"): continue + if line.startswith("!"): excludes.append(line[1:].strip()) else: includes.append(line) + return includes, excludes -def expand_globs(includes: Iterable[str], excludes: Iterable[str]) -> List[Path]: +def expand_globs( + includes: Iterable[str], + excludes: Iterable[str], +) -> List[Path]: files: List[Path] = [] + for selector in includes: - if any(ch in selector for ch in ["*", "?", "["]): + if any(ch in selector for ch in ("*", "?", "[")): files.extend(Path(p) for p in glob.glob(selector, recursive=True)) else: files.append(Path(selector)) - uniq: List[Path] = [] + unique_files: List[Path] = [] seen: Set[str] = set() - for p in files: - key = str(p) - if key in seen: + + for path in files: + source = str(path) + + if source in seen: continue - seen.add(key) - if p.exists() and p.is_file(): - uniq.append(p) - - out: List[Path] = [] - for p in uniq: - source = str(p) - if any(fnmatch.fnmatch(source, ex) for ex in excludes): + + seen.add(source) + + if path.exists() and path.is_file(): + unique_files.append(path) + + output: List[Path] = [] + + for path in unique_files: + source = str(path) + + if any(fnmatch.fnmatch(source, pattern) for pattern in excludes): continue - out.append(p) - return sorted(out, key=lambda p: p.name.lower()) + output.append(path) + + return sorted(output, key=lambda path: str(path).lower()) + + +def choose_primary_variable(dataset: Dataset) -> Optional[str]: + coordinate_names = { + "time", + "lat", + "latitude", + "lon", + "longitude", + "bnds", + "climatology_bnds", + "height", + } -def choose_primary_variable(ds: Dataset) -> Optional[str]: - coord = {"time", "lat", "latitude", "lon", "longitude", "bnds", "climatology_bnds", "height"} candidates: List[str] = [] - for name, variable in ds.variables.items(): - if name.lower() in coord: + + for name, variable in dataset.variables.items(): + if name.lower() in coordinate_names: continue - dims = [d.lower() for d in getattr(variable, "dimensions", ())] - if len(dims) < 2: + + dimensions = [ + dimension.lower() for dimension in getattr(variable, "dimensions", ()) + ] + + if len(dimensions) < 2: continue - has_lat = any(d in ("lat", "latitude") for d in dims) - has_lon = any(d in ("lon", "longitude") for d in dims) - if has_lat and has_lon: + + has_latitude = any(dimension in ("lat", "latitude") for dimension in dimensions) + has_longitude = any( + dimension in ("lon", "longitude") for dimension in dimensions + ) + + if has_latitude and has_longitude: candidates.append(name) + if candidates: return candidates[0] - for name, variable in ds.variables.items(): - if getattr(variable, "ndim", 0) >= 2 and name.lower() not in coord: + + for name, variable in dataset.variables.items(): + if getattr(variable, "ndim", 0) >= 2 and name.lower() not in coordinate_names: return name + return None -def min_max_for_file(path: Path) -> Optional[Tuple[str, float, float]]: - with Dataset(path) as ds: - var_name = choose_primary_variable(ds) - if not var_name or var_name not in ds.variables: - return None - variable = ds.variables[var_name] - data = variable[:] - fill = getattr(variable, "_FillValue", None) - missing = getattr(variable, "missing_value", None) - arr = np.array(data, dtype="float64") - mask = ~np.isfinite(arr) - if fill is not None: - mask |= arr == float(fill) - if missing is not None: - try: - mask |= arr == float(missing) - except Exception: - pass - valid = arr[~mask] - if valid.size == 0: - return None - return var_name, float(np.min(valid)), float(np.max(valid)) +# --------------------------------------------------------------------------- +# netCDF4 min/max scan +# --------------------------------------------------------------------------- + + +def _chunked_min_max(variable, max_chunk_bytes: int) -> Tuple[float, float]: + """Stream the variable's leading dimension in memory-bounded chunks and + reduce with numpy's C-level min/max. Respects netCDF4's auto-mask, so + _FillValue/missing_value cells are excluded from the result.""" + + shape = variable.shape + + if len(shape) == 0: + data = variable[...] + if np.ma.isMaskedArray(data): + data = data.compressed() + if data.size == 0: + raise ValueError("variable has no unmasked data") + return float(data.min()), float(data.max()) + + outer_dim = shape[0] + per_slice_elems = 1 + for size in shape[1:]: + per_slice_elems *= size + + # Packed variables may be expanded by netCDF4 during auto-scaling. Budget + # for float64 even when their on-disk dtype is smaller. + itemsize = max( + 8, + variable.dtype.itemsize if hasattr(variable.dtype, "itemsize") else 8, + ) + per_slice_bytes = max(1, per_slice_elems * itemsize) + chunk_len = max(1, max_chunk_bytes // per_slice_bytes) + + current_min: Optional[float] = None + current_max: Optional[float] = None + for start in range(0, outer_dim, chunk_len): + end = min(start + chunk_len, outer_dim) + block = variable[start:end, ...] -def load_existing_rows(path: Path) -> Dict[str, List[str]]: + if np.ma.isMaskedArray(block): + if block.mask is np.ma.nomask: + block_data = block.data + if block_data.size == 0: + continue + block_min = block_data.min() + block_max = block_data.max() + else: + compressed = block.compressed() + if compressed.size == 0: + continue + block_min = compressed.min() + block_max = compressed.max() + else: + if block.size == 0: + continue + block_min = block.min() + block_max = block.max() + + block_min = float(block_min) + block_max = float(block_max) + + if not np.isfinite(block_min) or not np.isfinite(block_max): + # fall back to nan-aware reduction only for this chunk + finite = ( + block[np.isfinite(block)] + if not np.ma.isMaskedArray(block) + else block.compressed() + ) + finite = np.asarray(finite) + finite = finite[np.isfinite(finite)] + if finite.size == 0: + continue + block_min = float(finite.min()) + block_max = float(finite.max()) + + if current_min is None or block_min < current_min: + current_min = block_min + if current_max is None or block_max > current_max: + current_max = block_max + + if current_min is None or current_max is None: + raise ValueError("no valid (unmasked, finite) data found") + + return current_min, current_max + + +def netcdf4_min_max_for_file( + path: Path, + max_chunk_bytes: int, + trust_metadata_range: bool, +) -> MinMaxResult: + with Dataset(path, mode="r") as dataset: + variable_name = choose_primary_variable(dataset) + + if not variable_name: + raise ValueError("no suitable primary variable found") + + variable = dataset.variables[variable_name] + + if trust_metadata_range: + for attr_name in ("actual_range", "valid_range"): + if attr_name in variable.ncattrs(): + values = np.atleast_1d(variable.getncattr(attr_name)) + if values.size == 2: + return variable_name, float(values[0]), float(values[1]) + + has_min = "valid_min" in variable.ncattrs() + has_max = "valid_max" in variable.ncattrs() + if has_min and has_max: + return ( + variable_name, + float(variable.getncattr("valid_min")), + float(variable.getncattr("valid_max")), + ) + + minimum, maximum = _chunked_min_max(variable, max_chunk_bytes) + + return variable_name, minimum, maximum + + +# --------------------------------------------------------------------------- +# Output CSV handling (extended with a `source` column) +# --------------------------------------------------------------------------- + + +def load_existing_rows(path: Path) -> Dict[RowKey, List[str]]: if not path.exists(): return {} - out: Dict[str, List[str]] = {} + + rows: Dict[RowKey, List[str]] = {} + with path.open("r", encoding="utf-8", newline="") as handle: reader = csv.reader(handle) + for row in reader: if len(row) < 5: continue - source = str(row[1] or "").strip() - if source: - out[source] = row - return out + + portal_id = str(row[0] or "").strip() + source_path = str(row[1] or "").strip() + + if portal_id and source_path: + rows[(portal_id, source_path)] = row + + return rows + + +def write_rows(path: Path, rows: Dict[RowKey, List[str]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = path.with_suffix(path.suffix + ".tmp") + + with temporary_path.open( + "w", + encoding="utf-8", + newline="", + ) as handle: + writer = csv.writer(handle) + + for key in sorted(rows): + writer.writerow(rows[key]) + + temporary_path.replace(path) + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- def main() -> int: - parser = argparse.ArgumentParser(description="Compute min/max for portal file-pattern definitions.") - parser.add_argument("--portal", action="append", help="portal id(s), e.g. canada_mosaic") - parser.add_argument("--portals-dir", default=str(REPO_ROOT / "portal-prep" / "portal-file-patterns")) - parser.add_argument("--out-csv", default=str(REPO_ROOT / "portal-prep" / "pdp_min_max.csv")) - parser.add_argument("--all", action="store_true", help="recompute all files, not just missing rows") + parser = argparse.ArgumentParser( + description=( + "Compute portal NetCDF min/max values: reuse the DB export " + "wherever a filename already has values there, and scan with " + "netCDF4 + numpy (no CDO) only for files the DB doesn't know " + "about." + ) + ) + parser.add_argument( + "--db-csv", + help=( + "CSV export produced by portal-prep/min_max_query.sql. " + "Used to skip scanning " + "filenames that already have range_min/range_max." + ), + default=str(REPO_ROOT / "portal-prep" / "db-export.csv"), + ) + parser.add_argument( + "--portal", + action="append", + help="portal id(s), e.g. canada_mosaic", + ) + parser.add_argument( + "--portals-dir", + default=str(REPO_ROOT / "portal-prep" / "portal-file-patterns"), + ) + parser.add_argument( + "--out-csv", + default=str(REPO_ROOT / "portal-prep" / "pdp_min_max.csv"), + ) + parser.add_argument( + "--all", + action="store_true", + help="refresh all rows from the DB or a scan instead of skipping existing rows", + ) + parser.add_argument( + "--workers", + type=int, + default=1, + help="number of NetCDF files scanned concurrently (default: 1)", + ) + parser.add_argument( + "--include-disabled", + action="store_true", + help=( + "By default, DB rows with disabled=True are ignored (treated " + "as if the DB doesn't have them, so they get rescanned). Pass " + "this flag to trust disabled DB rows too." + ), + ) + parser.add_argument( + "--max-chunk-mb", + type=int, + default=256, + help=( + "Memory budget per read chunk when scanning a file, in MB " + "(default: 256). Larger = fewer, bigger reads (usually " + "faster up to a point); smaller = lower peak memory." + ), + ) + parser.add_argument( + "--trust-metadata-range", + action="store_true", + help=( + "If the variable carries actual_range/valid_range/valid_min+" + "valid_max attributes, use them directly instead of scanning " + "the data. Fast, but only as accurate as those attributes -- " + "off by default because valid_range is sometimes a QC bound " + "rather than the true data range." + ), + ) args = parser.parse_args() + if args.workers <= 0: + parser.error("--workers must be greater than zero") + if args.max_chunk_mb <= 0: + parser.error("--max-chunk-mb must be greater than zero") + + max_chunk_bytes = args.max_chunk_mb * 1024 * 1024 + + db_csv_path = Path(args.db_csv).resolve() + db_lookup = load_db_lookup(db_csv_path) + print( + f"Loaded DB export: {len(db_lookup[0])} unique paths, " + f"{len(db_lookup[1])} unique basenames", + flush=True, + ) + portals_dir = Path(args.portals_dir).resolve() out_csv = Path(args.out_csv).resolve() - portals = [p for p in (args.portal or []) if p] + + portals = [portal for portal in (args.portal or []) if portal] + if not portals: - portals = sorted(p.stem for p in portals_dir.glob("*.txt")) + portals = sorted(path.stem for path in portals_dir.glob("*.txt")) existing = load_existing_rows(out_csv) - rows: Dict[str, List[str]] = {} if args.all else dict(existing) + rows: Dict[RowKey, List[str]] = {} if args.all else dict(existing) - processed = 0 - added = 0 - failed = 0 + # Group portal/file pairs by source file, so a file referenced by + # multiple portals is only scanned once. + pending_by_source: Dict[str, List[str]] = {} + candidate_pairs = 0 + skipped_pairs = 0 + db_hits = 0 for portal_id in portals: selector_file = portals_dir / f"{portal_id}.txt" + if not selector_file.exists(): - print(f"{portal_id}: missing portal pattern file, skipped") + print( + f"{portal_id}: missing portal pattern file, skipped", + flush=True, + ) continue + includes, excludes = read_portals(selector_file) files = expand_globs(includes, excludes) - for src in files: - source_path = str(src) - if (not args.all) and source_path in rows: + + for source in files: + candidate_pairs += 1 + source_path = str(source) + key = (portal_id, source_path) + + if not args.all and key in rows: + skipped_pairs += 1 + continue + + # --- DB-first check ------------------------------------------- + db_record = select_db_record( + source, + db_lookup, + args.include_disabled, + ) + + if db_record: + variable_name, minimum, maximum, _disabled, ensemble_name = db_record + rows[key] = [ + portal_id, + source_path, + variable_name, + f"{minimum:.8g}", + f"{maximum:.8g}", + f"db:{ensemble_name}", + ] + db_hits += 1 continue - processed += 1 + + # Not in the DB (or only disabled rows) -- queue for a scan. + pending_by_source.setdefault(source_path, []).append(portal_id) + + total_unique_files = len(pending_by_source) + + print( + f"Starting: portals={len(portals)} " + f"candidate_pairs={candidate_pairs} " + f"skipped_existing={skipped_pairs} " + f"filled_from_db={db_hits} " + f"unique_files_to_scan={total_unique_files} " + f"workers={args.workers} " + f"existing_rows={len(rows)}", + flush=True, + ) + + if total_unique_files == 0: + write_rows(out_csv, rows) + print( + f"Finished: scanned=0 failed=0 total_rows={len(rows)}", + flush=True, + ) + return 0 + + completed_count = 0 + added_rows = 0 + failed_files = 0 + + # ProcessPoolExecutor, not threads: the numpy/HDF5 decompression work + # here is CPU-bound, so separate processes give real parallelism + # instead of fighting over the GIL. + with ProcessPoolExecutor(max_workers=args.workers) as executor: + futures = {} + + for source_path in pending_by_source: + print(f"QUEUE: {source_path}", flush=True) + future = executor.submit( + netcdf4_min_max_for_file, + Path(source_path), + max_chunk_bytes, + args.trust_metadata_range, + ) + futures[future] = source_path + + for future in as_completed(futures): + source_path = futures[future] + completed_count += 1 + try: - result = min_max_for_file(src) - if not result: - failed += 1 - continue - var_name, min_value, max_value = result - rows[source_path] = [portal_id, source_path, var_name, f"{min_value:.8g}", f"{max_value:.8g}"] - added += 1 - except Exception: - failed += 1 - - out_csv.parent.mkdir(parents=True, exist_ok=True) - with out_csv.open("w", encoding="utf-8", newline="") as handle: - writer = csv.writer(handle) - for source in sorted(rows.keys()): - writer.writerow(rows[source]) + variable_name, minimum, maximum = future.result() + + for portal_id in pending_by_source[source_path]: + key = (portal_id, source_path) + rows[key] = [ + portal_id, + source_path, + variable_name, + f"{minimum:.8g}", + f"{maximum:.8g}", + "computed", + ] + added_rows += 1 + + write_rows(out_csv, rows) + + print( + f"[{completed_count}/{total_unique_files}] " + f"DONE {Path(source_path).name}: " + f"{variable_name} " + f"min={minimum:.8g} " + f"max={maximum:.8g} " + f"rows_added={len(pending_by_source[source_path])} " + f"saved_rows={len(rows)}", + flush=True, + ) + + except Exception as exc: + failed_files += 1 + print( + f"[{completed_count}/{total_unique_files}] " + f"FAILED {source_path}: " + f"{type(exc).__name__}: {exc}", + flush=True, + ) + + write_rows(out_csv, rows) + + print( + f"Finished: scanned={total_unique_files} " + f"rows_added={added_rows} " + f"filled_from_db={db_hits} " + f"failed_files={failed_files} " + f"total_rows={len(rows)}", + flush=True, + ) - print(f"portals={len(portals)} processed={processed} added_or_updated={added} failed={failed} total_rows={len(rows)}") return 0 if __name__ == "__main__": raise SystemExit(main()) - From e0e5249b7bdef98902c9a2a2402a6c632aa73254 Mon Sep 17 00:00:00 2001 From: Quintin Date: Mon, 27 Jul 2026 14:04:54 -0700 Subject: [PATCH 3/3] Handle cmip5 run attrs --- portal_meta_builder/builder.py | 5 ++--- portal_meta_builder/metadata.py | 24 ++++++++++++++++++------ 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/portal_meta_builder/builder.py b/portal_meta_builder/builder.py index 1b0ba59..cb1e7a7 100644 --- a/portal_meta_builder/builder.py +++ b/portal_meta_builder/builder.py @@ -4,7 +4,7 @@ from typing import Any, Dict, List, Set, Tuple from .io_utils import file_fingerprint, load_json, utc_now_iso -from .metadata import ensure_derived_fields, read_netcdf_metadata +from .metadata import derive_common_fields, read_netcdf_metadata from .minmax import select_minmax_record from .portals import build_menu_tree, derive_menu_fields, get_portal_config @@ -94,7 +94,7 @@ def build_portal_payload( cached_metadata = entry.get("metadata") if isinstance(entry.get("metadata"), dict) else {} if cached_metadata: - cached_metadata = ensure_derived_fields(cached_metadata) + cached_metadata["derived"] = derive_common_fields(cached_metadata) entry["metadata"] = cached_metadata recomputed = derive_menu_fields(portal_id, cached_metadata) if recomputed and recomputed != entry.get("menuFields"): @@ -149,4 +149,3 @@ def build_portal_payload( } stats = {"updated": updated, "kept": kept, "removed": removed} return new_payload, stats, len(files) - diff --git a/portal_meta_builder/metadata.py b/portal_meta_builder/metadata.py index 77c2305..eb01d7f 100644 --- a/portal_meta_builder/metadata.py +++ b/portal_meta_builder/metadata.py @@ -115,11 +115,24 @@ def normalize_run_label(raw: str, preserve_forcing: bool = False) -> str: def compose_run_from_attrs(global_attrs: Dict[str, Any], prefixes: Sequence[str]) -> str: for prefix in prefixes: - realization = str(global_attrs.get(f"{prefix}realization_index") or "").strip() - initialization = str( - global_attrs.get(f"{prefix}initialization_index") or "" - ).strip() - physics = str(global_attrs.get(f"{prefix}physics_index") or "").strip() + realization = first_non_empty( + [ + global_attrs.get(f"{prefix}realization_index"), + global_attrs.get(f"{prefix}realization"), + ] + ) + initialization = first_non_empty( + [ + global_attrs.get(f"{prefix}initialization_index"), + global_attrs.get(f"{prefix}initialization_method"), + ] + ) + physics = first_non_empty( + [ + global_attrs.get(f"{prefix}physics_index"), + global_attrs.get(f"{prefix}physics_version"), + ] + ) forcing = str(global_attrs.get(f"{prefix}forcing_index") or "").strip() if realization and initialization and physics: run = f"r{realization}i{initialization}p{physics}" @@ -240,4 +253,3 @@ def read_netcdf_metadata(path: Path) -> Dict[str, Any]: pass return ensure_derived_fields(metadata) -