Coordinate validation against the CMOR tables

checks/coordinate_checks/__init__.py

@@ -0,0 +1,8 @@
+"""Coordinate checks: validate a file's coordinates against the CMOR-table
+standard (via the ``plugins.coordinate_standard`` engine)."""
+
+from checks.coordinate_checks.check_coordinates import check_coordinates
+
+__all__ = [
+    "check_coordinates",
+]

checks/coordinate_checks/build_standard.py

@@ -0,0 +1,67 @@
+"""Inspect the STANDARD built from the CMOR tables.
+
+Shim over ``plugins/coordinate_standard.py`` (the engine, built on
+``compliance_checker.cf.util``) -- no logic lives here. The CLI summarises the
+built standard: how many coordinate objects, and the role/representation
+coverage.
+
+Run
+---
+    python build_standard.py                 # default: CMIP7 tables in ./tables
+    python build_standard.py --prefix CMIP6
+
+Needs the repo checkout (bootstrapped below) and compliance-checker.
+"""
+from __future__ import annotations
+
+import sys
+from collections import Counter
+from pathlib import Path
+
+# Bootstrap the repo root only when run as a script; as a module import the
+# root is already on sys.path, and imports must not mutate it.
+if __name__ == "__main__":
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from plugins.coordinate_standard import (  # noqa: E402, F401  (re-exports)
+    Role, Representation, VariableKind, Coordinate, Standard,
+    StandardProvider, CmorTableProvider, EsgvocProvider,
+    classify_role, classify_representation,
+    VERTICAL_STANDARD_NAMES, PARAMETRIC_STANDARD_NAMES, ROLE_STANDARD_NAMES,
+)
+
+TABLES_DIR = Path(__file__).parent / "tables"
+
+
+def summarise(std: Standard) -> None:
+    """Print the standard's provenance and classification coverage."""
+    print(f"Standard: {std.source}"
+          + (f"  (version {std.version})" if std.version else ""))
+    print(f"  {len(std.coordinates)} coordinate objects "
+          f"| skipped: {len(std.formula_terms)} formula terms, "
+          f"{len(std.bounds_vars)} bounds, {len(std.grid_mappings)} grid mapping(s)\n")
+    for title, key in (("Roles", lambda c: c.role.value),
+                       ("Representations", lambda c: c.representation.value)):
+        print(f"  {title}")
+        for value, n in Counter(map(key, std.coordinates)).most_common():
+            print(f"    {value:24} {n}")
+    unknown = [c.name for c in std.coordinates if c.role is Role.UNKNOWN]
+    print(f"  Unknown roles: {len(unknown)}"
+          + (f" -> {', '.join(unknown)}" if unknown else ""))
+
+
+def main() -> None:
+    import argparse
+
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--tables", default=str(TABLES_DIR),
+                    help="directory of CMOR table JSON")
+    ap.add_argument("--prefix", default="CMIP7",
+                    help="table prefix: CMIP6 | CMIP7 | CORDEX-CMIP6")
+    args = ap.parse_args()
+
+    summarise(CmorTableProvider(args.tables, args.prefix).build())
+
+
+if __name__ == "__main__":
+    main()

checks/coordinate_checks/check_coordinates.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python
+"""
+Validate a file's coordinates against the CMOR-table standard.
+
+This is the compliance-checker glue. It builds the standard from the already
+loaded CMOR tables (passed in -- no re-reading), classifies the file with the
+``plugins.coordinate_standard`` engine, matches each coordinate to its standard
+entry, and emits one Result per coordinate. Soft findings (matched with
+warnings, or unresolvable through no fault of the file) pass the main check
+and are surfaced as LOW-severity advisories.
+
+Facets with an existing atomic check are delegated instead of re-implemented:
+
+* direction: once a coordinate is pinned to its table entry, the entry's
+  ``stored_direction`` is fed to [VAR005] ``check_coordinate_monotonicity``.
+  Coordinates whose direction is already handled by a plugin's TOML rules can
+  be excluded via ``skip_direction_for`` (the TOML rule wins; no double check).
+* missing coordinate variables: resolved by the engine's
+  ``missing_coordinates`` (a data variable uses a dimension matching a table
+  ``out_name``, but no such variable exists), then reported through [VAR001]
+  ``check_variable_existence``.
+* bounds correctness ([VAR004]/[VAR012]) is NOT invoked here: the plugins
+  already run both for every coordinate that has a ``bounds`` attribute. The
+  only gap -- bounds required by the table but absent -- is caught by the
+  engine's ``diff`` ("missing bounds").
+
+Wiring (in a plugin's check method)::
+
+    from checks.coordinate_checks.check_coordinates import check_coordinates
+
+    def check_coordinates_vs_standard(self, ds):
+        return check_coordinates(
+            ds, self.CTcoords, self.CTgrids, self.CTformulas,
+            severity=BaseCheck.HIGH,
+            skip_direction_for=set(self.coords_cfg_monotonicity_names),
+        )
+
+``ds`` is the netCDF4 dataset compliance-checker passes to NC checks.
+"""
+from compliance_checker.base import BaseCheck, TestCtx
+
+from checks.variable_checks.check_coordinate_monotonicity import (
+    check_coordinate_monotonicity,
+)
+from checks.variable_checks.check_variable_existence import (
+    check_variable_existence,
+)
+from plugins.coordinate_standard import (
+    GOOD_OUTCOMES,
+    WARN_OUTCOMES,
+    CmorTableProvider,
+    classify_dataset,
+    match,
+    missing_coordinates,
+    read_1d_values,
+)
+
+CHECK_ID = "COORD001"
+
+# Outcomes whose detail is the resolved standard-entry name; only then can
+# table-driven parameters (stored_direction) be handed to other checks.
+_RESOLVED = ("MATCH", "PINNED", "MATCH_WITH_WARNINGS", "MISMATCH")
+
+
+def check_coordinates(ds, ctcoords, ctgrids, ctformulas,
+                      severity=BaseCheck.HIGH, project="CMOR",
+                      skip_direction_for=None, ct_dimensions=None):
+    """Check every coordinate in ``ds`` against the standard.
+
+    Parameters
+    ----------
+    ds : netCDF4.Dataset
+        The dataset being checked.
+    ctcoords, ctgrids, ctformulas : dict
+        The already-loaded CMOR coordinate / grids / formula_terms tables
+        (e.g. ``self.CTcoords`` etc.).
+    severity : str
+        Check severity (default: ``BaseCheck.HIGH``).
+    project : str
+        Project label for provenance, e.g. "CMIP6" / "CORDEX-CMIP6".
+    skip_direction_for : iterable of str, optional
+        Coordinate names whose monotonicity direction is already checked
+        elsewhere (e.g. via a plugin's TOML ``monotonicity`` rule); [VAR005]
+        is not invoked for these.
+    ct_dimensions : iterable of str, optional
+        Token list from the data variable's CMOR *variable table* entry
+        ``dimensions`` string (e.g. ``self.CT[table_id]["variable_entry"]
+        [var]["dimensions"].split()``). Enables detection of missing scalar
+        and auxiliary coordinates (``height2m``, 2-D lat/lon) that are never
+        netCDF dimensions.
+
+    Returns
+    -------
+    list[Result]
+        One Result per coordinate variable found in the file, [VAR005]
+        Results for table-directed coordinates, [VAR001] Results for
+        expected-but-missing coordinate variables, and a LOW-severity
+        advisory Result per soft finding.
+    """
+    skip_direction = set(skip_direction_for or ())
+
+    std = CmorTableProvider(prefix=project, tables={
+        "coordinate": ctcoords,
+        "grids": ctgrids,
+        "formula_terms": ctformulas,
+    }).build()
+
+    kinds, candidates = classify_dataset(ds)
+    if not candidates:
+        ctx = TestCtx(severity, f"[{CHECK_ID}] Coordinates")
+        ctx.add_pass()                       # nothing to check in this file
+        return [ctx.to_result()]
+
+    results = []
+    for name, candidate in candidates.items():
+        ctx = TestCtx(severity, f"[{CHECK_ID}] Coordinate '{name}'")
+        try:
+            values = read_1d_values(ds.variables[name])
+            outcome, detail = match(candidate, values, std)
+        except Exception as e:   # a malformed entry must not kill the suite
+            ctx.add_failure(
+                f"internal error while checking coordinate '{name}': {e}")
+            results.append(ctx.to_result())
+            continue
+        kind = outcome.split(":", 1)[0]      # strip the ":<why>" off MISMATCH
+
+        if kind in GOOD_OUTCOMES:
+            ctx.add_pass()
+        elif kind in WARN_OUTCOMES:
+            # Matched with soft issues, or cannot resolve which entry (not
+            # the file's fault, e.g. time/lat/lon): pass the main check and
+            # surface the finding as a LOW-severity advisory.
+            ctx.add_pass()
+            warn_ctx = TestCtx(BaseCheck.LOW,
+                               f"[{CHECK_ID}] Coordinate '{name}' (advisory)")
+            warn_ctx.add_failure(
+                f"coordinate '{name}': {outcome}"
+                + (f" [candidates: {detail}]" if detail else ""))
+            results.append(warn_ctx.to_result())
+        else:
+            ctx.add_failure(
+                f"coordinate '{name}' ({candidate.role.value}, "
+                f"{candidate.representation.value}) does not match the standard: "
+                f"{outcome}" + (f" [candidates: {detail}]" if detail else "")
+            )
+        results.append(ctx.to_result())
+
+        # Direction: delegate to [VAR005] with the pinned entry's
+        # stored_direction (TOML-managed coordinates are skipped -- the
+        # plugin's own VAR005 call wins).
+        if kind in _RESOLVED and isinstance(detail, str) and name not in skip_direction:
+            entry = next(
+                (c for c in std.coordinates
+                 if c.name == detail
+                 and c.representation == candidate.representation), None)
+            if (entry is not None and entry.stored_direction
+                    and name in getattr(ds, "dimensions", {})):
+                results.extend(check_coordinate_monotonicity(
+                    ds, coord_name=name,
+                    direction=entry.stored_direction, severity=severity))
+
+    # Expected coordinate variables that don't exist (a data variable uses a
+    # dimension that matches a table out_name, but no such variable exists):
+    # resolved by the engine, reported through [VAR001].
+    try:
+        missing = missing_coordinates(ds, kinds, std,
+                                      ct_dimensions=ct_dimensions)
+    except Exception as e:
+        ctx = TestCtx(severity, f"[{CHECK_ID}] Missing coordinates")
+        ctx.add_failure(f"internal error while checking for missing "
+                        f"coordinates: {e}")
+        results.append(ctx.to_result())
+        missing = {}
+    for name in missing:
+        results.extend(check_variable_existence(ds, name, severity))
+
+    return results

checks/coordinate_checks/classify_file.py

@@ -0,0 +1,70 @@
+"""Classify a netCDF file's coordinates and check them against the STANDARD.
+
+Shim over ``plugins/coordinate_standard.py`` (the engine, built on
+``compliance_checker.cf.util``) -- no logic lives here. ``classify_file`` is
+kept as an alias of the engine's ``classify_dataset`` for old callers.
+
+Run
+---
+    python classify_file.py FILE.nc                  # default: CMIP7 standard
+    python classify_file.py FILE.nc --prefix CMIP6
+
+Exit code = number of coordinates with FAIL findings (0 = all clean), so it is
+usable in a folder sweep / CI loop.
+
+Needs the repo checkout (bootstrapped below), netCDF4 and compliance-checker.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Bootstrap the repo root only when run as a script; as a module import the
+# root is already on sys.path, and imports must not mutate it.
+if __name__ == "__main__":
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from plugins.coordinate_standard import (  # noqa: E402, F401  (re-exports)
+    Role, Representation, VariableKind, Coordinate, Standard,
+    CmorTableProvider, classify_dataset, match, diff, read_1d_values, report,
+    GOOD_OUTCOMES, WARN_OUTCOMES,
+)
+
+classify_file = classify_dataset            # old name for the previous API
+
+TABLES_DIR = Path(__file__).parent / "tables"
+
+
+def main() -> None:
+    import argparse
+
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("netcdf", help="path to the netCDF file")
+    ap.add_argument("--tables", default=str(TABLES_DIR),
+                    help="directory of CMOR table JSON")
+    ap.add_argument("--prefix", default="CMIP7",
+                    help="table prefix: CMIP6 | CMIP7 | CORDEX-CMIP6")
+    args = ap.parse_args()
+
+    try:
+        from netCDF4 import Dataset
+    except ImportError:
+        sys.exit("netCDF4 not found. Activate an environment with netCDF4 "
+                 "installed.")
+
+    std = CmorTableProvider(args.tables, args.prefix).build()
+    print(f"Standard: {std.source}"
+          + (f" (version {std.version})" if std.version else "")
+          + f", {len(std.coordinates)} coords")
+    print(f"File:     {args.netcdf}")
+
+    ds = Dataset(args.netcdf)
+    try:
+        fails = report(ds, std)
+    finally:
+        ds.close()
+    sys.exit(min(fails, 255))
+
+
+if __name__ == "__main__":
+    main()