Source code for hydro.conformance

import logging
from dataclasses import dataclass
from operator import methodcaller

import xarray as xr

# TODO Remove me
from rich.logging import RichHandler

[docs] FORMAT = "%(message)s"
logging.basicConfig( level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[RichHandler()] ) # end Remove
[docs] log = logging.getLogger(__name__)
@dataclass(frozen=True)
[docs] class CheckResult:
[docs] error: str | None = None
[docs] warning: str | None = None
@property
[docs] def ok(self) -> bool: return self.error is None and self.warning is None
[docs] class CCHDOnetCDF10:
[docs] __cchdo_version__ = "1.0"
[docs] def check_cf_version(self, ds): cf_version = "CF-1.8" try: conventions = ds.attrs["Conventions"] except AttributeError: return CheckResult( error=f"No 'attrs' attribute on {ds}, is it an xarray.Dataset?" ) except KeyError: return CheckResult(error="Global attribute 'Conventions' is required") # We need to check that the correct cf conventions version is in the # "Conventions" attribute, since there can be more than one whitespace # seperated convention, we also need to check the split apart version # for exactly the value we expect if cf_version in conventions and cf_version in conventions.split(): return CheckResult() return CheckResult(error="{cf_version} not in {conventions}")
[docs] def iter_errors(self, ds: xr.Dataset): checks = { name: methodcaller(name, ds) for name in dir(self) if name.startswith("check_") } log.debug(f"Check methods: {list(checks.keys())}") for name, check in checks.items(): check_result = check(self) log.debug(f"{name}: {check_result}") if check_result.ok: continue yield check_result
[docs] def validate(self, ds: xr.Dataset): return not any(self.iter_errors(ds))
# Alias the "curren version"
[docs] CCHDOnetCDF = CCHDOnetCDF10