diff --git a/checks/format_checks/check_internal_packing.py b/checks/format_checks/check_internal_packing.py index 4a6cdcab..2ca2e818 100644 --- a/checks/format_checks/check_internal_packing.py +++ b/checks/format_checks/check_internal_packing.py @@ -20,6 +20,8 @@ import numpy as np from compliance_checker.base import BaseCheck, TestCtx +from checks.utils import severity_word + # --------------------------------------------------------------------------- # Optional dependency # --------------------------------------------------------------------------- @@ -27,7 +29,7 @@ from packaging.version import Version import pyfive - _PYFIVE_MIN = Version("1.1.1") + _PYFIVE_MIN = Version("1.1.2") _pyfive_version = Version(__import__("importlib.metadata", fromlist=["version"]).version("pyfive")) if _pyfive_version < _PYFIVE_MIN: raise RuntimeError( @@ -40,7 +42,9 @@ _PYFIVE_OK = False _PYFIVE_ERR = str(e) -_FOUR_MiB = 4 * (2**20) # 4 194 304 bytes +_DEFAULT_MIN_CHUNK_SIZE_BYTES = 4 * (2**20) # 4 194 304 bytes +_INTERNAL_PACKING_FILE_SESSIONS = {} +_INTERNAL_PACKING_FINALIZE_COUNTERS = {} # --------------------------------------------------------------------------- @@ -60,54 +64,173 @@ def _attr_to_str(attr) -> str: return str(np.array(attr).astype("U")) +def _session_key(ds, file_path: str | None) -> str: + return file_path or f"id:{id(ds)}" + + +def _acquire_internal_packing_file(ds, file_path: str): + key = _session_key(ds, file_path) + session = _INTERNAL_PACKING_FILE_SESSIONS.get(key) + if session and session.get("file") is not None: + return session["file"] + + f = pyfive.File(file_path) + _INTERNAL_PACKING_FILE_SESSIONS[key] = {"file": f} + return f + + +def finalize_internal_packing_session(ds, total_packing_checks: int = 3) -> None: + """Called after each internal packing section check method (metadata / time / data). + + Closes the shared FILE004 pyfive file handle after all section check methods + have reported in. + + Parameters + ---------- + ds : dataset + The dataset being checked. + total_packing_checks : int, optional + Total number of internal packing section check methods in the plugin. + Default is 3 (metadata + time + data). Custom plugins that only implement + a subset should pass the actual number so the file is closed promptly. + """ + file_path = _get_file_path(ds) + key = _session_key(ds, file_path) + + entry = _INTERNAL_PACKING_FINALIZE_COUNTERS.get(key) + if entry is None: + # First finalize call: record the threshold and count. + entry = {"count": 1, "total": total_packing_checks} + else: + if entry["total"] != total_packing_checks: + raise ValueError( + f"finalize_internal_packing_session called with total_packing_checks={total_packing_checks} " + f"but was previously initialized with total_packing_checks={entry['total']}. " + "Ensure all section check methods pass the same total_packing_checks value." + ) + entry = {"count": entry["count"] + 1, "total": entry["total"]} + + if entry["count"] >= entry["total"]: + session = _INTERNAL_PACKING_FILE_SESSIONS.pop(key, None) + if session: + try: + session["file"].close() + except Exception: + pass + _INTERNAL_PACKING_FINALIZE_COUNTERS.pop(key, None) + else: + _INTERNAL_PACKING_FINALIZE_COUNTERS[key] = entry + + def _is_single_chunk_or_contiguous(var) -> tuple[bool, str]: """ True if the variable is contiguous (chunks is None) or has exactly one chunk. Mirrors the logic in the official check_cmip7_packing script. """ - chunks = var.chunks # None → contiguous + try: + chunks = var.chunks # None -> contiguous + except Exception as e: + return False, f"unable to read chunk metadata ({e})" if chunks is None: return True, "contiguous" - n = var.id.get_num_chunks() + try: + n = var.id.get_num_chunks() + except Exception as e: + return False, f"unable to read number of chunks ({e})" if n <= 1: return True, f"1 chunk of shape {tuple(chunks)}" return False, f"{n} chunks (expected 1 chunk or contiguous)" -def _check_data_variable(var, variable_id: str) -> tuple[bool, str]: +def _check_data_variable( + var, + min_chunk_size_bytes: int, + frequency: str | None = None, + frequency_min_timesteps: dict[str, int] | None = None, + has_time: bool = False, +) -> tuple[bool, str]: """ Pass conditions (any one sufficient): • contiguous (chunks is None) • exactly 1 chunk - • uncompressed chunk size >= 4 MiB - • adding one element along the leading dimension would reach >= 4 MiB + • uncompressed chunk size >= configurable threshold (4 MiB by default) + • adding one element along the leading dimension would reach threshold (the "lee_way" rule from the official script) + • optional frequency-specific fallback: minimum timesteps per chunk """ - chunks = var.chunks + try: + chunks = var.chunks + except Exception as e: + return False, f"unable to read chunk metadata ({e})" if chunks is None: return True, "contiguous" - n = var.id.get_num_chunks() + try: + n = var.id.get_num_chunks() + except Exception as e: + return False, f"unable to read number of chunks ({e})" if n <= 1: return True, f"1 chunk" - wordsize = var.dtype.itemsize - chunksize = prod(chunks) * wordsize + try: + wordsize = var.dtype.itemsize + chunksize = prod(chunks) * wordsize + except Exception as e: + return False, f"unable to compute chunk byte size ({e})" # Adding one element along leading dim gives this extra size - lee_way = prod(chunks[1:]) * wordsize if len(chunks) > 1 else 0 + try: + lee_way = prod(chunks[1:]) * wordsize if len(chunks) > 1 else 0 + except Exception as e: + return False, f"unable to compute chunk threshold margin ({e})" - if chunksize + lee_way >= _FOUR_MiB: + if chunksize + lee_way >= min_chunk_size_bytes: return True, ( f"chunk size {chunksize} B " - f"(>= {_FOUR_MiB - lee_way} B threshold)" + f"(>= {min_chunk_size_bytes - lee_way} B threshold)" + ) + + if frequency_min_timesteps and has_time: + if not frequency or str(frequency).strip().lower() == "unknown": + known = ", ".join(sorted(str(k) for k in frequency_min_timesteps)) + return False, ( + f"uncompressed chunk size {chunksize} B " + f"(expected at least {min_chunk_size_bytes - lee_way} B, " + f"or 1 chunk, or contiguous). " + f"Warning: frequency-specific exceptions are configured for {known}, " + "but frequency cannot be inferred from the file metadata." + ) + + if frequency in frequency_min_timesteps: + required_steps = int(frequency_min_timesteps[frequency]) + chunk_time_steps = int(chunks[0]) if chunks is not None and len(chunks) > 0 else None + + if chunk_time_steps is not None and chunk_time_steps >= required_steps: + return True, ( + f"chunk size {chunksize} B below threshold, but frequency " + f"exception for '{frequency}' is met " + f"({chunk_time_steps} >= {required_steps} timesteps per time-chunk)" + ) + + return False, ( + f"uncompressed chunk size {chunksize} B " + f"(expected at least {min_chunk_size_bytes - lee_way} B, " + f"or at least {required_steps} timesteps per time-chunk for " + f"frequency '{frequency}', or 1 chunk, or contiguous)" + ) + + known = ", ".join(sorted(str(k) for k in frequency_min_timesteps)) + return False, ( + f"uncompressed chunk size {chunksize} B " + f"(expected at least {min_chunk_size_bytes - lee_way} B, " + f"or 1 chunk, or contiguous)." ) return False, ( f"uncompressed chunk size {chunksize} B " - f"(expected at least {_FOUR_MiB - lee_way} B, " + f"(expected at least {min_chunk_size_bytes - lee_way} B, " f"or 1 chunk, or contiguous)" ) @@ -116,17 +239,46 @@ def _check_data_variable(var, variable_id: str) -> tuple[bool, str]: # Public check function # --------------------------------------------------------------------------- -def check_cmip7_packing(ds, severity=BaseCheck.HIGH) -> list: +def check_internal_packing( + ds, + severity=BaseCheck.HIGH, + severity_metadata=None, + severity_time=None, + severity_data=None, + min_chunk_size_bytes: int = _DEFAULT_MIN_CHUNK_SIZE_BYTES, + frequency: str | None = None, + frequency_min_timesteps: dict[str, int] | None = None, + run_metadata: bool = True, + run_time: bool = True, + run_data: bool = True, +) -> list: """ - [FILE004] CMIP7 internal packing checks. + [FILE004] Internal packing checks for CMIP7 and similar workflows. """ results = [] + try: + min_chunk_size_bytes = int(min_chunk_size_bytes) + if min_chunk_size_bytes <= 0: + raise ValueError + except Exception: + min_chunk_size_bytes = _DEFAULT_MIN_CHUNK_SIZE_BYTES + + default_severity = severity + sev_metadata = severity_metadata or default_severity + sev_time = severity_time or default_severity + sev_data = severity_data or default_severity + + has_time = False + + if not (run_metadata or run_time or run_data): + return results + # -- pyfive available? --------------------------------------------------- if not _PYFIVE_OK: - ctx = TestCtx(BaseCheck.HIGH, "[FILE004] CMIP7 internal packing") + ctx = TestCtx(BaseCheck.HIGH, "[FILE004] Internal packing") ctx.add_failure( - f"Optional dependency 'pyfive >= 1.1.1' is not installed or " + f"Optional dependency 'pyfive >= 1.1.2' is not installed or " f"incompatible — FILE004 skipped. ({_PYFIVE_ERR})" ) return [ctx.to_result()] @@ -134,48 +286,74 @@ def check_cmip7_packing(ds, severity=BaseCheck.HIGH) -> list: # -- get file path ------------------------------------------------------- file_path = _get_file_path(ds) if not file_path: - ctx = TestCtx(severity, "[FILE004] CMIP7 internal packing") - ctx.add_failure("Could not retrieve dataset file path — FILE004 skipped.") + ctx = TestCtx(default_severity, "[FILE004] Internal packing") + qualifier = severity_word(default_severity) + ctx.add_failure( + "Could not retrieve dataset file path — FILE004 skipped. " + f"It is {qualifier} to run internal packing checks on an accessible local file." + ) return [ctx.to_result()] - # -- open with pyfive --------------------------- + # -- open/reuse pyfive file handle --------------------------- try: - f = pyfive.File(file_path) + f = _acquire_internal_packing_file(ds, file_path) except Exception as e: - ctx = TestCtx(severity, "[FILE004] CMIP7 internal packing") + ctx = TestCtx(default_severity, "[FILE004] Internal packing") ctx.add_failure(f"Could not open file with pyfive: {e}") return [ctx.to_result()] - try: - # ---------------------------------------------------------------- - # FILE004a — Consolidated internal metadata - # ---------------------------------------------------------------- - ctx_a = TestCtx(severity, "[FILE004a] CMIP7 internal packing : Consolidated internal metadata") + # ---------------------------------------------------------------- + # FILE004a — Consolidated internal metadata + # ---------------------------------------------------------------- + if run_metadata: + ctx_a = TestCtx(sev_metadata, "[FILE004a] Internal packing : Consolidated internal metadata") try: if f.consolidated_metadata: ctx_a.add_pass() else: + qualifier = severity_word(sev_metadata) ctx_a.add_failure( "File does not have consolidated internal metadata. " - "Run 'cmip7repack' to fix this." + f"It is {qualifier} to consolidate internal metadata " + "using cmip7repack or comparable tools." ) except Exception as e: ctx_a.add_failure(f"Unable to inspect consolidated metadata: {e}") results.append(ctx_a.to_result()) - # ---------------------------------------------------------------- - # FILE004b — Time coordinate: single chunk or contiguous - # ---------------------------------------------------------------- - if "time" in f: + # ---------------------------------------------------------------- + # FILE004b — Time coordinate: single chunk or contiguous + # ---------------------------------------------------------------- + if run_time or run_data: + try: + has_time = "time" in f + except Exception as e: + has_time = False + if run_time: + ctx_b = TestCtx(sev_time, "[FILE004b] Internal packing : Time coordinate chunking") + ctx_b.add_failure(f"Unable to inspect time coordinate presence: {e}") + results.append(ctx_b.to_result()) + + if run_time and has_time: + try: t = f["time"] - ctx_b = TestCtx(severity, "[FILE004b] CMIP7 internal packing : Time coordinate chunking") + except Exception as e: + ctx_b = TestCtx(sev_time, "[FILE004b] Internal packing : Time coordinate chunking") + ctx_b.add_failure(f"Unable to access time coordinate variable: {e}") + results.append(ctx_b.to_result()) + t = None + + if t is not None: + ctx_b = TestCtx(sev_time, "[FILE004b] Internal packing : Time coordinate chunking") ok, detail = _is_single_chunk_or_contiguous(t) if ok: ctx_b.add_pass() else: + qualifier = severity_word(sev_time) ctx_b.add_failure( f"Time coordinate variable 'time' has {detail}. " - f"Run 'cmip7repack' to fix this." + f"It is {qualifier} to repack this variable " + "using cmip7repack or comparable tools." ) results.append(ctx_b.to_result()) @@ -185,52 +363,95 @@ def check_cmip7_packing(ds, severity=BaseCheck.HIGH) -> list: try: if "bounds" in t.attrs: bounds_name = _attr_to_str(t.attrs["bounds"]) - if bounds_name in f: + try: + has_bounds_var = bounds_name in f + except Exception as e: + has_bounds_var = False + ctx_c = TestCtx(sev_time, "[FILE004c] Time bounds chunking") + ctx_c.add_failure(f"Unable to inspect time bounds variable presence: {e}") + results.append(ctx_c.to_result()) + + if has_bounds_var: b = f[bounds_name] ctx_c = TestCtx( - severity, - f"[FILE004c] CMIP7 internal packing : Time bounds chunking ('{bounds_name}')", + sev_time, + f"[FILE004c] Internal packing : Time bounds chunking ('{bounds_name}')", ) ok, detail = _is_single_chunk_or_contiguous(b) if ok: ctx_c.add_pass() else: + qualifier = severity_word(sev_time) ctx_c.add_failure( f"Time bounds variable '{bounds_name}' has {detail}. " - f"Run 'cmip7repack' to fix this." + f"It is {qualifier} to repack this variable " + "using cmip7repack or comparable tools." ) results.append(ctx_c.to_result()) except Exception as e: - ctx_c = TestCtx(severity, "[FILE004c] Time bounds chunking") + ctx_c = TestCtx(sev_time, "[FILE004c] Time bounds chunking") ctx_c.add_failure(f"Unable to inspect time bounds chunking: {e}") results.append(ctx_c.to_result()) - # ---------------------------------------------------------------- - # FILE004d — Data variable chunk size - # ---------------------------------------------------------------- - if "variable_id" in f.attrs: + # ---------------------------------------------------------------- + # FILE004d — Data variable chunk size + # ---------------------------------------------------------------- + if run_data: + try: + has_variable_id_attr = "variable_id" in f.attrs + except Exception as e: + has_variable_id_attr = False + ctx_d = TestCtx(sev_data, "[FILE004d] Internal packing : Data variable chunking") + ctx_d.add_failure(f"Unable to inspect 'variable_id' attribute presence: {e}") + results.append(ctx_d.to_result()) + + if has_variable_id_attr: try: variable_id = _attr_to_str(f.attrs["variable_id"]) except Exception: variable_id = None - if variable_id and variable_id in f: - d = f[variable_id] + try: + has_variable = bool(variable_id) and (variable_id in f) + except Exception as e: + has_variable = False + ctx_d = TestCtx(sev_data, "[FILE004d] Internal packing : Data variable chunking") + ctx_d.add_failure(f"Unable to inspect data variable presence: {e}") + results.append(ctx_d.to_result()) + + if has_variable: + try: + d = f[variable_id] + except Exception as e: + d = None + ctx_d = TestCtx( + sev_data, + f"[FILE004d] Internal packing : Data variable chunking ('{variable_id}')", + ) + ctx_d.add_failure(f"Unable to access data variable '{variable_id}': {e}") + results.append(ctx_d.to_result()) + + if has_variable and d is not None: ctx_d = TestCtx( - severity, - f"[FILE004d] CMIP7 internal packing : Data variable chunking ('{variable_id}')", + sev_data, + f"[FILE004d] Internal packing : Data variable chunking ('{variable_id}')", + ) + ok, detail = _check_data_variable( + d, + min_chunk_size_bytes=min_chunk_size_bytes, + frequency=frequency, + frequency_min_timesteps=frequency_min_timesteps, + has_time=has_time, ) - ok, detail = _check_data_variable(d, variable_id) if ok: ctx_d.add_pass() else: + qualifier = severity_word(sev_data) ctx_d.add_failure( f"Data variable '{variable_id}': {detail}. " - f"Run 'cmip7repack' to fix this." + f"It is {qualifier} to repack chunking " + "using cmip7repack or comparable tools." ) results.append(ctx_d.to_result()) - finally: - f.close() - return results diff --git a/plugins/cmip7/cmip7.py b/plugins/cmip7/cmip7.py index 731f16bd..891a24a9 100644 --- a/plugins/cmip7/cmip7.py +++ b/plugins/cmip7/cmip7.py @@ -20,7 +20,10 @@ from checks.format_checks.check_format import check_format from checks.format_checks.check_compression import check_compression -from checks.format_checks.check_internal_packing import check_cmip7_packing +from checks.format_checks.check_internal_packing import ( + check_internal_packing, + finalize_internal_packing_session, +) from checks.consistency_checks.check_drs_filename_cv import ( check_drs_filename, check_drs_directory, @@ -431,14 +434,65 @@ def check_File_Compression(self, ds): except TypeError: return check_compression(ds, sev) - def check_File_Internal_Packing(self, ds): - if not self.config or not self.config.file or not self.config.file.internal_packing: - return [] + def check_File_Internal_Packing_Metadata(self, ds): + try: + if not self.config or not self.config.file or not self.config.file.internal_packing: + return [] + + r = self.config.file.internal_packing + if not r.metadata: + return [] + + return check_internal_packing( + ds, + severity=self.get_severity(r.metadata.severity), + run_metadata=True, + run_time=False, + run_data=False, + ) + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) - r = self.config.file.internal_packing - sev = self.get_severity(r.severity) + def check_File_Internal_Packing_Time(self, ds): + try: + if not self.config or not self.config.file or not self.config.file.internal_packing: + return [] + + r = self.config.file.internal_packing + if not r.time: + return [] + + return check_internal_packing( + ds, + severity=self.get_severity(r.time.severity), + run_metadata=False, + run_time=True, + run_data=False, + ) + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) - return check_cmip7_packing(ds, severity=sev) + def check_File_Internal_Packing_Data(self, ds): + try: + if not self.config or not self.config.file or not self.config.file.internal_packing: + return [] + + r = self.config.file.internal_packing + if not r.data: + return [] + + return check_internal_packing( + ds, + severity=self.get_severity(r.data.severity), + min_chunk_size_bytes=(r.data.min_chunk_size_bytes or 4 * (2**20)), + frequency=self.frequency, + frequency_min_timesteps=r.data.frequency_min_timesteps, + run_metadata=False, + run_time=False, + run_data=True, + ) + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) # ------------------------------------------------------------------------- # 2) Global attributes diff --git a/plugins/cmip7/config/wcrp/file.toml b/plugins/cmip7/config/wcrp/file.toml index 66aac6b1..de908b79 100644 --- a/plugins/cmip7/config/wcrp/file.toml +++ b/plugins/cmip7/config/wcrp/file.toml @@ -15,6 +15,20 @@ allowed_data_models = ["NETCDF4_CLASSIC", "NETCDF4"] #expected_complevel = 1 #expected_shuffle = true -[file.internal_packing] -# Check FILE004 -severity = "H" \ No newline at end of file +[file.internal_packing.metadata] +# Check FILE004a - consolidated metadata +severity = "H" + +[file.internal_packing.time] +# Check FILE004b-c - time and time bounds chunking +severity = "H" + +[file.internal_packing.data] +# Check FILE004d - data variable chunking +severity = "H" + +# Optional minimum uncompressed chunk size threshold (bytes). Default: 4 MiB. +#min_chunk_size_bytes = 4194304 + +# Optional fallback exceptions by frequency (applied only if file has time dependency). +#frequency_min_timesteps = { "dec" = 1, "yr" = 1, "mon" = 6, "day" = 1, "6hr" = 4, "3hr" = 1, "1hr" = 6 } \ No newline at end of file diff --git a/plugins/cordex_cmip6/cordex_cmip6.py b/plugins/cordex_cmip6/cordex_cmip6.py index a487edba..70a3d75b 100644 --- a/plugins/cordex_cmip6/cordex_cmip6.py +++ b/plugins/cordex_cmip6/cordex_cmip6.py @@ -45,6 +45,10 @@ ) from checks.format_checks.check_compression import check_compression from checks.format_checks.check_format import check_format +from checks.format_checks.check_internal_packing import ( + check_internal_packing, + finalize_internal_packing_session, +) from checks.time_checks.check_time_cordex_cmip6 import ( check_calendar, check_time_chunking, @@ -242,6 +246,90 @@ def check_compression(self, ds): return results + def check_file_internal_packing_metadata(self, ds): + """ + [FILE004a] Internal packing consolidated metadata check. + """ + try: + results = [] + if "format_checks" not in self.config or "internal_packing" not in self.config["format_checks"]: + return results + + check_config = self.config["format_checks"]["internal_packing"] + metadata_cfg = check_config.get("metadata") + if not metadata_cfg: + return results + + results.extend( + check_internal_packing( + ds, + severity=self.get_severity(metadata_cfg.get("severity")), + run_metadata=True, + run_time=False, + run_data=False, + ) + ) + return results + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) + + def check_file_internal_packing_time(self, ds): + """ + [FILE004b-c] Internal packing time and time-bounds checks. + """ + try: + results = [] + if "format_checks" not in self.config or "internal_packing" not in self.config["format_checks"]: + return results + + check_config = self.config["format_checks"]["internal_packing"] + time_cfg = check_config.get("time") + if not time_cfg: + return results + + results.extend( + check_internal_packing( + ds, + severity=self.get_severity(time_cfg.get("severity")), + run_metadata=False, + run_time=True, + run_data=False, + ) + ) + return results + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) + + def check_file_internal_packing_data(self, ds): + """ + [FILE004d] Internal packing data-variable chunking check. + """ + try: + results = [] + if "format_checks" not in self.config or "internal_packing" not in self.config["format_checks"]: + return results + + check_config = self.config["format_checks"]["internal_packing"] + data_cfg = check_config.get("data") + if not data_cfg: + return results + + results.extend( + check_internal_packing( + ds, + severity=self.get_severity(data_cfg.get("severity")), + min_chunk_size_bytes=data_cfg.get("min_chunk_size_bytes", 4 * (2**20)), + frequency=self.frequency, + frequency_min_timesteps=data_cfg.get("frequency_min_timesteps"), + run_metadata=False, + run_time=False, + run_data=True, + ) + ) + return results + finally: + finalize_internal_packing_session(ds, total_packing_checks=3) + def check_data_types(self, ds): """ [VAR011] Checks if the coordinate and variable data types are as expected according to the CORDEX-CMIP6 Archive Specifications. @@ -699,4 +787,4 @@ def check_consistency_filename_from_config(self, ds): ) ) - return results + return results \ No newline at end of file diff --git a/plugins/cordex_cmip6/resources/wcrp_config.toml b/plugins/cordex_cmip6/resources/wcrp_config.toml index 82026be6..eec8c9db 100644 --- a/plugins/cordex_cmip6/resources/wcrp_config.toml +++ b/plugins/cordex_cmip6/resources/wcrp_config.toml @@ -18,6 +18,22 @@ severity = "M" expected_complevel = 1 expected_shuffle = true +[format_checks.internal_packing.metadata] +# Check FILE004a - consolidated metadata +severity = "M" + +[format_checks.internal_packing.time] +# Check FILE004b-c - time and time bounds chunking +severity = "M" + +[format_checks.internal_packing.data] +# Check FILE004d - data variable chunking +severity = "M" +# Optional minimum uncompressed chunk size threshold (bytes). Default: 4 MiB. +#min_chunk_size_bytes = 4194304 +# Optional fallback exceptions by frequency (applied only if file has time dependency). +frequency_min_timesteps = { "mon" = 6, "day" = 1, "6hr" = 4, "3hr" = 1, "1hr" = 6 } + #------------------------------------------------------------------------------ # Coordinate / Variable Checks # ----------------------------------------------------------------------------- diff --git a/plugins/wcrp_schema.py b/plugins/wcrp_schema.py index b8f9a92e..4291cbf6 100644 --- a/plugins/wcrp_schema.py +++ b/plugins/wcrp_schema.py @@ -126,11 +126,47 @@ class FileSection(BaseModel): compression: Optional[FileCompressionRule] = None internal_packing: Optional[FileInternalPackingRule] = None -class FileInternalPackingRule(BaseModel): + +class FileInternalPackingMetadataRule(BaseModel): + model_config = ConfigDict(extra="forbid") + severity: Optional[str] = None + + +class FileInternalPackingTimeRule(BaseModel): model_config = ConfigDict(extra="forbid") severity: Optional[str] = None +class FileInternalPackingDataRule(BaseModel): + model_config = ConfigDict(extra="forbid") + severity: Optional[str] = None + min_chunk_size_bytes: Optional[int] = Field(default=None, ge=1) + frequency_min_timesteps: Optional[Dict[str, int]] = None + + @model_validator(mode="after") + def _validate_frequency_min_timesteps(self): + if self.frequency_min_timesteps is None: + return self + + for freq, steps in self.frequency_min_timesteps.items(): + try: + if int(steps) <= 0: + raise ValueError + except Exception as e: + raise ValueError( + f"frequency_min_timesteps['{freq}'] must be a positive integer" + ) from e + + return self + + +class FileInternalPackingRule(BaseModel): + model_config = ConfigDict(extra="forbid") + metadata: Optional[FileInternalPackingMetadataRule] = None + time: Optional[FileInternalPackingTimeRule] = None + data: Optional[FileInternalPackingDataRule] = None + + # ============================================================================= # drs.toml # ============================================================================= diff --git a/pyproject.toml b/pyproject.toml index 08f2c48e..dbd3fb2b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ dependencies = [ "cf_xarray", "esgvoc", "pooch", - "pyfive >= 1.1.1", + "pyfive >= 1.1.2", "pydantic>=2.12.5", ] version = "2.3.2"