diff --git a/docs/api.rst b/docs/api.rst index b32ca02..f476d9a 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -10,6 +10,9 @@ API .. automodule:: gents.timeseries :members: +.. automodule:: gents.tsinventory + :members: + .. automodule:: gents.mhfdataset :members: diff --git a/docs/llm/architecture.md b/docs/llm/architecture.md index 210e185..90dc701 100644 --- a/docs/llm/architecture.md +++ b/docs/llm/architecture.md @@ -25,7 +25,8 @@ TSCollection — list of order dicts gents/timeseries.py │ modifiers: include/exclude, add_args, apply_compression, │ apply_chunk_target_bytes, apply_path_swap, apply_overwrite, │ append_timestep_dirs, add_attrs, skip_existing (resume a - │ continued run -- narrows/drops orders against existing output) + │ continued run -- narrows/drops orders against existing output, + │ found via a TSInventory of the output directory) │ .execute() — batches orders sharing source files (optimize=True), │ pool (or in-process) → generate_time_series() per batch ▼ @@ -46,14 +47,17 @@ bundled YAML config (`gents/configs/*.yaml`) merged with command-line flags. | Module | Owns | Depends on | |---|---|---| | `datastore.py` | `GenTSDataStore` — the *only* place `netCDF4.Dataset` is constructed in the pipeline; context manager + delegation | netCDF4 | -| `meta.py` | `netCDFMeta` (cached header metadata, with opt-out load flags), `is_var_secondary` (primary/secondary rules), `get_meta_from_path` (picklable factory), `get_attributes`, `get_time_variables_names` | datastore | +| `meta.py` | `netCDFMeta` (cached header metadata, with opt-out load flags), `is_var_secondary` (primary/secondary rules), `get_meta_from_path` (picklable factory), `read_meta_map` (pool-or-serial bulk read, shared by both collections), `split_valid` (drop unloaded/invalid entries), `get_attributes`, `get_time_variables_names` | datastore, utils | | `hfcollection.py` | `HFCollection` + module-level helpers for discovery, grouping, date math (`get_time_boundary_num`, `get_year_boundary_num`), fragmentation merging | meta, utils | | `mhfdataset.py` | `MHFDataset` — presents a file group as one virtual dataset; time→file mapping; memory-bounded variable cache; tile reassembly | datastore, meta | -| `timeseries.py` | `TSCollection` (including `skip_existing`, the resume modifier), order construction/execution, `write_timeseries_file`, `compute_chunksizes`, `CHUNK_TARGET_BYTES`, `check_timeseries_integrity`, `check_timeseries_conform`, `get_timestamp_format`, `get_timestep_label` | mhfdataset, datastore, meta, utils | +| `timeseries.py` | `TSCollection` (including `skip_existing`, the resume modifier), order construction/execution, `write_timeseries_file`, `compute_chunksizes`, `CHUNK_TARGET_BYTES`, `check_timeseries_integrity`, `check_timeseries_conform`, `get_timestamp_format` | mhfdataset, tsinventory, datastore, meta, utils | +| `tsinventory.py` | `TSInventory` — index of *existing* time series files (the output-side mirror of `HFCollection`): lazy metadata, `include`/`exclude`, `get_variables()` grouping by `.`, `get_latest_time()`. Keeps only files carrying `gents_version`. An empty or missing directory is an empty inventory, not an error. Also `get_ts_meta_from_path` (its picklable factory) | hfcollection (`find_files`, `get_group_timestep_delta`), meta, datastore, utils | | `cli.py` | argparse, YAML config loading/merging, `check_config`, `main()` | hfcollection, timeseries, utils | -| `utils.py` | `enable_logging`, `ProgressBar`, `get_version`, `get_time_stamp`, `log_hfcollection_info`, `log_tscollection_info`, `LOG_LEVEL_IO_WARNING = 5` | — | +| `utils.py` | `enable_logging`, `ProgressBar`, `filter_by_globs` (path-glob filter behind both collections' `include`/`exclude`), `get_timestep_label`, `get_version`, `get_time_stamp`, `log_hfcollection_info`, `log_tscollection_info`, `LOG_LEVEL_IO_WARNING = 5` | — | -Dependency direction is one-way: `cli → (hfcollection, timeseries) → (meta, mhfdataset) → datastore`. +Dependency direction is one-way: `cli → (hfcollection, timeseries) → (meta, mhfdataset) → datastore`, +and `timeseries → tsinventory → hfcollection`. `hfcollection` must never import `tsinventory` +(nor `timeseries`), which is what keeps `timeseries ↔ tsinventory` from becoming a cycle. `utils` is a leaf used by the upper layers. Keep it that way. ## Conformity subsystem (`gents/conformity/`) @@ -184,9 +188,10 @@ number of simultaneously open handles is therefore 1, independent of group size. sub-minute `%Y%m%d%H%M%S`, hourly `%Y%m%d%H`, daily `%Y%m%d`, monthly `%Y%m`, yearly `%Y`. `ts_string` is `f"{start.strftime(fmt)}-{end.strftime(fmt)}"` using the chosen time-alignment method; `update_ts_orders` finds the range on raw float times -(alignment applied vectorized) and decodes only each file's two endpoint candidates. `get_timestep_label` maps the same delta to a frequency +(alignment applied vectorized) and decodes only each file's two endpoint candidates. `utils.get_timestep_label` maps the same delta to a frequency label (`hour_6`, `day_1`, `month_1`, `year_1`, or `unsorted`), used by -`append_timestep_dirs` and by `case_builder.log_case_summary`. +`append_timestep_dirs`, `TSInventory.get_variables`, and `case_builder.log_case_summary` +(still importable as `gents.timeseries.get_timestep_label`). ## Parallelism model @@ -203,7 +208,9 @@ makes `py-spy` see real frames (see the profiling section of [workflows.md](workflows.md)). Everything submitted to a pool must be picklable: module-level functions -(`get_meta_from_path`, `generate_time_series`) with plain-data arguments. Failures in +(`get_meta_from_path`, `get_ts_meta_from_path`, `generate_time_series`) with plain-data arguments. +The metadata pull itself (pool/serial split, progress bar, per-file error handling) lives once in +`meta.read_meta_map`, which `HFCollection` and `TSInventory` both call. Failures in workers are logged (`raise_errors=False` default) rather than raised, so a bad file doesn't kill a long batch job. Dask was removed; `dask_client` kwargs remain only as deprecation shims. diff --git a/docs/llm/concepts.md b/docs/llm/concepts.md index e288bcc..8ec798b 100644 --- a/docs/llm/concepts.md +++ b/docs/llm/concepts.md @@ -65,10 +65,11 @@ - **Resume / incremental generation (`TSCollection.skip_existing`):** Narrows each already-built order to the HF source files not yet covered by existing output, for a CESM/E3SM run that was continued from a restart after time series were already generated - for the earlier portion. Scans the output directory (its own directory only, not - recursively) for files matching `{ts_path_template}.{primary_var}.*.nc`, and takes the - latest raw time any *complete* (`check_timeseries_integrity`-passing) match covers as the - cutoff. Drops each `hf_paths` entry *whole* if its own last raw time step does not exceed + for the earlier portion. Builds a `TSInventory` of the order's output directory (one per + directory, walked once) and narrows it to files matching + `{ts_path_template}.{primary_var}.*.nc`; the latest raw time any *complete* match covers + (`TSInventory.get_latest_time`) is the cutoff. Complete means the file loads and carries + `gents_version`; anything else is dropped with a warning and counts as no coverage. Drops each `hf_paths` entry *whole* if its own last raw time step does not exceed that cutoff, keeps it whole otherwise — **file granularity, not step granularity**: a continued run always starts a fresh HF file at the restart boundary (a stream's step count can't change mid-run without a namelist edit and a fresh run), so no single file can diff --git a/docs/llm/conventions.md b/docs/llm/conventions.md index 68e56c9..2072fb6 100644 --- a/docs/llm/conventions.md +++ b/docs/llm/conventions.md @@ -16,13 +16,23 @@ Path-glob filters must stay usable *before* the pull (that's the performance model: filter first, read headers second). Methods needing metadata call `check_pulled()`, which auto-pulls. +- **`TSInventory` follows the same model.** It is the output-side mirror of `HFCollection`: + lazy `None` metadata, immutable `include`/`exclude` via `.copy()`, and `pull_metadata`/ + `check_validity` mutating in place (the same internal exception). It may import from + `hfcollection`; `hfcollection` must never import `tsinventory`. +- **Collection plumbing is shared, not duplicated.** `include`/`exclude` go through + `utils.filter_by_globs`, the metadata pull through `meta.read_meta_map`, and validity + filtering through `meta.split_valid`, for both `HFCollection` and `TSInventory`. Fix a bug + there once. Tests that mock the pool or progress bar must patch `gents.meta.ProcessPoolExecutor` + / `gents.meta.ProgressBar`, not the names in `gents.hfcollection`. - **Copies inherit metadata.** Filtering a pulled collection yields pulled sub-collections; never re-read headers that a parent already read. (They also inherit the parent's *groups*, which is a bug — see "`include()`/`exclude()` after a pull" below.) - **Worker picklability.** Anything submitted to `ProcessPoolExecutor` must be a - module-level function with picklable args. `get_meta_from_path` exists precisely as - the picklable factory for `netCDFMeta`; keep that pattern. + module-level function with picklable args. `get_meta_from_path` (history files) and + `tsinventory.get_ts_meta_from_path` (time series files) exist precisely as the picklable + factories for `netCDFMeta`; keep that pattern. - **`num_processes <= 1` means no pool at all.** Both `pull_metadata` and `execute` branch to a plain serial loop rather than a one-worker pool. Profiling depends on it (see [workflows.md](workflows.md)); keep both branches in step, including their error diff --git a/docs/llm/index.md b/docs/llm/index.md index b02b368..39e798b 100644 --- a/docs/llm/index.md +++ b/docs/llm/index.md @@ -41,14 +41,15 @@ verified against the source; when docs and code disagree, trust the code and fix ## Repository map ``` -gents/ Package source (7 pipeline modules ~3,700 lines, + conformity/) +gents/ Package source (8 pipeline modules ~4,000 lines, + conformity/) hfcollection.py HFCollection: discover/filter/group/slice history files timeseries.py TSCollection: build + execute time-series "orders"; file writing meta.py netCDFMeta: cached per-file metadata; primary/secondary classification + tsinventory.py TSInventory: index of existing time series files (used by skip_existing) mhfdataset.py MHFDataset: virtual aggregated dataset over a file group, with cache datastore.py GenTSDataStore: thin context-manager wrapper around netCDF4.Dataset cli.py argparse CLI + YAML-config-driven main() - utils.py logging setup, ProgressBar, version, collection info loggers + utils.py logging setup, ProgressBar, path-glob filter, timestep label, version, collection info loggers configs/ Bundled YAML model configs (gents_example.yaml, gents_cesm3.yaml) conformity/ End-to-end verification against real model cases. Outside the pipeline; separate from unit tests. See its own README.md. @@ -56,7 +57,7 @@ gents/ Package source (7 pipeline modules ~3,700 lines, + confo check.py gents_conform — run a model spec against generated output report.py pass/fail/skip collector; text + JSON rendering models/cesm3.py what correct CESM3 output looks like (the file that matters) - tests/ pytest suite (284 tests); test_cases.py generates synthetic fixtures + tests/ pytest suite (335 tests); test_cases.py generates synthetic fixtures docs/ Sphinx docs (index/install/user/dev/api .rst) benchmarks/ ASV performance benchmarks + fixtures.build_bench_case() pipeline_bench.py Standalone py-spy profiling driver (repo root, not an ASV suite) @@ -82,7 +83,7 @@ group once into a memory-bounded cache) and writing one netCDF file per variable ```bash pip install -e ".[dev]" # editable install with test/docs/bench extras -pytest gents/tests/ # full test suite (284 tests, ~35 s) +pytest gents/tests/ # full test suite (335 tests, ~40 s) docker build --target test -t gents-tests . && docker run --rm gents-tests # CI-equivalent run_gents --model CESM3 --dryrun # CLI dry run (read-only validation) ``` diff --git a/docs/llm/workflows.md b/docs/llm/workflows.md index 0d1a409..e3bb577 100644 --- a/docs/llm/workflows.md +++ b/docs/llm/workflows.md @@ -15,7 +15,7 @@ pip install -r requirements.txt && pip install -e . ## Testing ```bash -pytest gents/tests/ # full suite (284 tests, ~35 s) +pytest gents/tests/ # full suite (335 tests, ~40 s) pytest gents/tests/test_workflow.py # one file ``` diff --git a/docs/user.rst b/docs/user.rst index a44825a..25798a4 100644 --- a/docs/user.rst +++ b/docs/user.rst @@ -228,6 +228,24 @@ The above code will print the list of time series dictionaries. By default, ``TS ts_collection.execute() +Inspecting Existing Output with ``TSInventory`` +----------------------------------------------- + +The ``gents.tsinventory.TSInventory`` class is the counterpart of ``HFCollection`` for time series files that already exist on disk. It finds the files under a directory, reads their headers on request, and groups the files that GenTS wrote by variable and frequency. A missing or empty directory gives an empty inventory instead of an error. ``TSCollection.skip_existing`` uses it to find what has already been generated. + +.. code-block:: python + + from gents.tsinventory import TSInventory + + inventory = TSInventory("my/file/system/scratch/GCM_run/timeseries/", num_processes=8) + inventory = inventory.include("*/atm/*") + inventory.pull_metadata() + + for variable, paths in inventory.get_variables().items(): + print(variable, len(paths)) + +Like ``HFCollection``, ``include`` and ``exclude`` return copies and are applied to the path string before any header is read. ``pull_metadata`` drops files that were not produced by GenTS or were not completely written (those missing the ``gents_version`` attribute). Each entry of ``get_variables`` is keyed as ``.`` (e.g. ``TS.month_1``) and its paths are ordered by start time. + Custom Dask Workflows with ``TSCollection`` ------------------------------------------- diff --git a/gents/hfcollection.py b/gents/hfcollection.py index e5ae0de..3a6afa0 100644 --- a/gents/hfcollection.py +++ b/gents/hfcollection.py @@ -5,11 +5,10 @@ Developer: Cameron Cummins Contact: cameron.cummins@utexas.edu """ -from gents.meta import get_meta_from_path -from gents.utils import ProgressBar, LOG_LEVEL_IO_WARNING +from gents.meta import get_meta_from_path, read_meta_map, split_valid +from gents.utils import filter_by_globs, LOG_LEVEL_IO_WARNING from cftime import num2date from pathlib import Path -from concurrent.futures import ProcessPoolExecutor, as_completed import numpy as np import os import fnmatch @@ -650,8 +649,9 @@ def pull_metadata(self, check_valid=True, raise_errors=False, show_progress=True """ Reads the header of every history file in the collection. - Runs :func:`~gents.meta.get_meta_from_path` over a process pool when - ``num_processes > 1`` and serially otherwise, then computes each group's + Reads headers with :func:`~gents.meta.read_meta_map` (running + :func:`~gents.meta.get_meta_from_path` over a process pool when + ``num_processes > 1`` and serially otherwise), then computes each group's timestep delta. :param check_valid: Run :meth:`check_validity` afterwards, dropping files @@ -672,31 +672,9 @@ def pull_metadata(self, check_valid=True, raise_errors=False, show_progress=True logger.info(f"Pulling metadata...") paths = list(self.__hf_to_meta_map.keys()) - prog_bar = ProgressBar(total=len(paths), label="Pulling Metadata", quiet=not show_progress) - if self.__num_processes > 1: - with ProcessPoolExecutor(max_workers=self.__num_processes) as executor: - futures = {executor.submit(get_meta_from_path, path): path for path in paths} - for future in as_completed(futures): - path = futures[future] - try: - result = future.result() - self.__hf_to_meta_map[path] = result - except Exception as exc: - logger.warning(f"Failed to load metadata for {path}: {exc}", exc_info=True) - if raise_errors: - raise - finally: - prog_bar.step() - else: - for path in paths: - try: - self.__hf_to_meta_map[path] = get_meta_from_path(path) - except Exception as exc: - logger.warning(f"Failed to load metadata for {path}: {exc}", exc_info=True) - if raise_errors: - raise - finally: - prog_bar.step() + self.__hf_to_meta_map.update(read_meta_map( + paths, get_meta_from_path, self.__num_processes, raise_errors, show_progress + )) if check_valid: self.check_validity() @@ -725,14 +703,7 @@ def check_validity(self): :rtype: dict """ logger.debug(f"Validating metadata...") - new_map = {} - removed = {} - for path in self.__hf_to_meta_map: - if self.__hf_to_meta_map[path] is not None and self.__hf_to_meta_map[path].is_valid(): - new_map[path] = self.__hf_to_meta_map[path] - else: - removed[path] = self.__hf_to_meta_map[path] - logger.warning(f"Could not pull valid/complete metadata for '{path}'.") + new_map, removed = split_valid(self.__hf_to_meta_map, lambda meta: meta.is_valid()) self.__hf_to_meta_map = new_map logger.debug(f"{len(new_map)} files valdiated ({len(removed)} removed).") return removed @@ -765,15 +736,7 @@ def include(self, glob_patterns): :type glob_patterns: list[str] or str :rtype: HFCollection """ - if type(glob_patterns) is str: - glob_patterns = [glob_patterns] - - filtered_path_map = {} - for path in self.__hf_to_meta_map: - for pattern in glob_patterns: - if fnmatch.fnmatch(str(path), pattern): - filtered_path_map[path] = self.__hf_to_meta_map[path] - break + filtered_path_map = filter_by_globs(self.__hf_to_meta_map, glob_patterns, keep_matches=True) logger.debug(f"Inclusive filter(s) applied: '{glob_patterns}'") return self.copy(meta_map=filtered_path_map) @@ -788,18 +751,7 @@ def exclude(self, glob_patterns): :type glob_patterns: list[str] or str :rtype: HFCollection """ - if type(glob_patterns) is str: - glob_patterns = [glob_patterns] - - filtered_path_map = {} - for path in self.__hf_to_meta_map: - matches = False - for pattern in glob_patterns: - if fnmatch.fnmatch(str(path), pattern): - matches = True - - if not matches: - filtered_path_map[path] = self.__hf_to_meta_map[path] + filtered_path_map = filter_by_globs(self.__hf_to_meta_map, glob_patterns, keep_matches=False) logger.debug(f"Exclusive filter(s) applied: '{glob_patterns}'") return self.copy(meta_map=filtered_path_map) diff --git a/gents/meta.py b/gents/meta.py index 2288dfc..0ea19f9 100644 --- a/gents/meta.py +++ b/gents/meta.py @@ -1,13 +1,18 @@ #!/usr/bin/env python """ -Per-file netCDF metadata: variable classification and cached header contents. +Per-file netCDF metadata: variable classification, cached header contents, and bulk header reads. Developer: Cameron Cummins Contact: cameron.cummins@utexas.edu """ +import logging import numpy as np +from concurrent.futures import ProcessPoolExecutor, as_completed from cftime import num2date from gents.datastore import GenTSDataStore +from gents.utils import ProgressBar + +logger = logging.getLogger(__name__) def is_var_secondary(variable, @@ -514,4 +519,80 @@ def get_meta_from_path(path: str): except Exception as e: raise type(e)(f"{e} Path: {path}") from e - return ds_meta \ No newline at end of file + return ds_meta + + +def read_meta_map(paths, factory, num_processes=1, raise_errors=False, show_progress=True): + """ + Reads the metadata of many files, in parallel when ``num_processes > 1``. + + Shared by the collections' ``pull_metadata``. A file that fails to load is + logged and left out of the result, so the caller's entry for it stays + unloaded. + + :param paths: Files to read. + :type paths: list[pathlib.Path] + :param factory: Picklable callable mapping a path to its metadata, such as + :func:`get_meta_from_path`. + :type factory: callable + :param num_processes: Worker processes; ``1`` reads serially in-process. + :type num_processes: int + :param raise_errors: Raise the first per-file failure instead of logging it. + :type raise_errors: bool + :param show_progress: If ``False``, suppress the stdout progress bar. + :type show_progress: bool + :returns: ``{path: metadata}`` for every file that loaded. + :rtype: dict + """ + loaded = {} + prog_bar = ProgressBar(total=len(paths), label="Pulling Metadata", quiet=not show_progress) + if num_processes > 1: + with ProcessPoolExecutor(max_workers=num_processes) as executor: + futures = {executor.submit(factory, path): path for path in paths} + for future in as_completed(futures): + path = futures[future] + try: + loaded[path] = future.result() + except Exception as exc: + logger.warning(f"Failed to load metadata for {path}: {exc}", exc_info=True) + if raise_errors: + raise + finally: + prog_bar.step() + else: + for path in paths: + try: + loaded[path] = factory(path) + except Exception as exc: + logger.warning(f"Failed to load metadata for {path}: {exc}", exc_info=True) + if raise_errors: + raise + finally: + prog_bar.step() + return loaded + + +def split_valid(meta_map, is_valid_fn): + """ + Splits a ``{path: metadata}`` map into usable and unusable entries. + + An entry is unusable if its metadata is ``None`` (never loaded, or the read + failed) or ``is_valid_fn`` rejects it. Warns about each unusable entry. + + :param meta_map: Metadata by path; not modified. + :type meta_map: dict + :param is_valid_fn: Predicate taking one metadata object, such as + :meth:`netCDFMeta.is_valid`. + :type is_valid_fn: callable + :returns: ``(kept, removed)`` dicts of ``{path: metadata}``. + :rtype: tuple[dict, dict] + """ + kept = {} + removed = {} + for path, meta in meta_map.items(): + if meta is not None and is_valid_fn(meta): + kept[path] = meta + else: + removed[path] = meta + logger.warning(f"Could not pull valid/complete metadata for '{path}'.") + return kept, removed diff --git a/gents/tests/test_meta.py b/gents/tests/test_meta.py index 8d47cd6..bd787dd 100644 --- a/gents/tests/test_meta.py +++ b/gents/tests/test_meta.py @@ -1,5 +1,5 @@ from gents.tests.test_cases import * -from gents.meta import netCDFMeta, is_var_secondary, get_attributes, get_time_variables_names, get_meta_from_path +from gents.meta import netCDFMeta, is_var_secondary, get_attributes, get_time_variables_names, get_meta_from_path, read_meta_map, split_valid from gents.datastore import GenTSDataStore import numpy as np import pytest @@ -415,3 +415,27 @@ def test_netcdfmeta_is_valid_does_not_decode(tmp_path): with patch("gents.meta.num2date") as mock_num2date: assert meta.is_valid() is True mock_num2date.assert_not_called() + + +@pytest.mark.parametrize("num_processes", [1, 2]) +def test_read_meta_map(tmp_path, num_processes): + """read_meta_map() returns metadata for loadable files and omits (not raises on) failures, serially and in a pool.""" + good = tmp_path / "good.nc" + generate_history_file(str(good), [15.0], [[0.0, 30.0]]) + missing = tmp_path / "missing.nc" + + loaded = read_meta_map([good, missing], get_meta_from_path, num_processes=num_processes, show_progress=False) + assert list(loaded) == [good] + assert isinstance(loaded[good], netCDFMeta) + + with pytest.raises(Exception): + read_meta_map([missing], get_meta_from_path, num_processes=num_processes, raise_errors=True, show_progress=False) + + +def test_split_valid(): + """split_valid() keeps entries the predicate accepts and removes both rejected and None entries.""" + meta_map = {"a": 1, "b": 2, "c": None} + kept, removed = split_valid(meta_map, lambda meta: meta == 1) + assert kept == {"a": 1} + assert removed == {"b": 2, "c": None} + assert meta_map == {"a": 1, "b": 2, "c": None} diff --git a/gents/tests/test_parallel.py b/gents/tests/test_parallel.py index 0bb0ba9..b53ba50 100644 --- a/gents/tests/test_parallel.py +++ b/gents/tests/test_parallel.py @@ -75,7 +75,7 @@ def test_dataset_opens(simple_case): input_head_dir, output_head_dir = simple_case hf_paths = [f"{input_head_dir}/{filename}" for filename in listdir(input_head_dir) if ".nc" in filename] - with patch("gents.hfcollection.ProcessPoolExecutor", ThreadPoolExecutor): + with patch("gents.meta.ProcessPoolExecutor", ThreadPoolExecutor): with patch("gents.meta.GenTSDataStore", wraps=GenTSDataStore) as mock_ds: assert mock_ds.call_count == 0 hf_collection = HFCollection(input_head_dir, num_processes=1) diff --git a/gents/tests/test_tscollection.py b/gents/tests/test_tscollection.py index fec691e..be48c1a 100644 --- a/gents/tests/test_tscollection.py +++ b/gents/tests/test_tscollection.py @@ -436,11 +436,6 @@ def test_tscollection_add_attrs(simple_case): assert ts_ds.getncattr("gents_test_key2") == "gents_test_val2" -def test_get_timestep_label_unknown_delta(): - """An unknown timestep duration is labelled 'unsorted' rather than guessed at.""" - assert get_timestep_label(None) == "unsorted" - - def test_update_ts_orders_no_full_time_decoding(multistep_large_case): """TSCollection construction (sorting + order building) decodes at most the per-file endpoint candidates, never a full time array.""" @@ -932,3 +927,15 @@ def test_batched_execute_keeps_per_variable_output_paths(tmp_path): assert isfile(f"{output_dir}/hist/testing.hf.VAR0.185001-185003.nc") assert isfile(f"{output_dir}/swapped/testing.hf.VAR1.185001-185003.nc") + + +def test_skip_existing_output_dir_with_glob_characters(continued_case): + """Glob characters in the output path are matched literally, so existing output there is still found.""" + input_head_dir, output_head_dir, extend = continued_case + special_dir = Path(output_head_dir) / "run[1]*" + TSCollection(HFCollection(input_head_dir), str(special_dir)).execute() + assert len(listdir(special_dir)) == CONTINUED_NUM_VARS + + resumed = TSCollection(HFCollection(input_head_dir), str(special_dir)).skip_existing() + + assert len(resumed) == 0 diff --git a/gents/tests/test_tsinventory.py b/gents/tests/test_tsinventory.py new file mode 100644 index 0000000..ccd738e --- /dev/null +++ b/gents/tests/test_tsinventory.py @@ -0,0 +1,99 @@ +from pathlib import Path +from gents.tests.test_cases import * +from gents.hfcollection import HFCollection +from gents.timeseries import TSCollection +from gents.tsinventory import TSInventory, get_ts_meta_from_path, get_time_sort_key, is_gents_output +import pytest + + +@pytest.fixture +def two_batch_output(continued_case): + """Time series output written in two runs, so each variable has two files covering consecutive periods.""" + input_head_dir, output_head_dir, extend = continued_case + TSCollection(HFCollection(input_head_dir), str(output_head_dir)).execute() + extend() + TSCollection(HFCollection(input_head_dir), str(output_head_dir)).skip_existing().execute() + return output_head_dir + + +def test_missing_and_empty_dir(tmp_path): + """A missing or empty directory gives an empty inventory instead of raising.""" + for head_dir in [tmp_path / "does_not_exist", tmp_path]: + inventory = TSInventory(str(head_dir)) + assert len(inventory) == 0 + assert inventory.is_pulled() + assert inventory.get_variables() == {} + + +def test_discover_and_pull(two_batch_output): + """Files are found unpulled, then pulled and grouped into one sorted entry per variable.""" + inventory = TSInventory(str(two_batch_output)) + assert len(inventory) == CONTINUED_NUM_VARS * 2 + assert not inventory.is_pulled() + assert all(meta is None for meta in inventory.values()) + + assert inventory.pull_metadata(show_progress=False) is inventory + assert inventory.is_pulled() + assert all(is_gents_output(meta) for meta in inventory.values()) + + variables = inventory.get_variables() + assert len(variables) == CONTINUED_NUM_VARS + for key, paths in variables.items(): + assert key.endswith(".month_1") + assert len(paths) == 2 + keys = [get_time_sort_key(inventory[path]) for path in paths] + assert keys == sorted(keys) + assert keys[0] != keys[1] + + +def test_parallel_pull_matches_serial(two_batch_output): + """Pulling with a process pool yields the same variable grouping as a serial pull.""" + serial = TSInventory(str(two_batch_output)) + parallel = TSInventory(str(two_batch_output), num_processes=2) + serial.pull_metadata(show_progress=False) + parallel.pull_metadata(show_progress=False) + assert serial.get_variables() == parallel.get_variables() + + +def test_unfinished_and_foreign_files_are_dropped(two_batch_output): + """Files without the gents_version stamp (unfinished writes, history files) are removed on pull.""" + foreign = f"{two_batch_output}/foreign.nc" + generate_history_file(foreign, [15.0], [[0.0, 30.0]]) + + inventory = TSInventory(str(two_batch_output)) + assert len(inventory) == CONTINUED_NUM_VARS * 2 + 1 + inventory.pull_metadata(show_progress=False) + + assert len(inventory) == CONTINUED_NUM_VARS * 2 + assert Path(foreign) not in inventory + + unvalidated = TSInventory(str(two_batch_output)).pull_metadata(check_valid=False, show_progress=False) + assert len(unvalidated) == CONTINUED_NUM_VARS * 2 + 1 + + +def test_filters_are_immutable_and_prune_variables(two_batch_output): + """include()/exclude() return new inventories and a cached variable grouping is pruned, not rebuilt.""" + inventory = TSInventory(str(two_batch_output)).pull_metadata(show_progress=False) + variables = inventory.get_variables() + var_key = sorted(variables)[0] + var_name = var_key.split(".")[0] + + only_var = inventory.include(f"*.{var_name}.*") + without_var = inventory.exclude(f"*.{var_name}.*") + assert only_var is not inventory + assert len(inventory) == CONTINUED_NUM_VARS * 2 + assert len(only_var) == 2 + assert len(without_var) == (CONTINUED_NUM_VARS - 1) * 2 + assert list(only_var.get_variables()) == [var_key] + assert only_var.get_variables()[var_key] == variables[var_key] + assert var_key not in without_var.get_variables() + assert len(inventory.include([])) == 0 + + +def test_get_ts_meta_from_path_keeps_variable_attrs(two_batch_output): + """Time series metadata keeps per-variable attributes, which history-file metadata drops.""" + inventory = TSInventory(str(two_batch_output)) + path = next(iter(inventory)) + meta = get_ts_meta_from_path(path) + var_name = meta.get_primary_variables()[0] + assert isinstance(meta.get_variable_attrs(var_name), dict) diff --git a/gents/tests/test_utils.py b/gents/tests/test_utils.py index e6aa2e0..bede5df 100644 --- a/gents/tests/test_utils.py +++ b/gents/tests/test_utils.py @@ -3,6 +3,7 @@ from gents.hfcollection import HFCollection from gents.timeseries import TSCollection import pytest +from datetime import timedelta LOGGER_HEADER_NUM_LINES = 2 @@ -157,7 +158,41 @@ def test_log_hfcollection_info_show_progress_reaches_metadata_pull(simple_case): hf_collection = HFCollection(input_head_dir) assert not hf_collection.is_pulled() - with patch("gents.hfcollection.ProgressBar") as mock_bar: + with patch("gents.meta.ProgressBar") as mock_bar: log_hfcollection_info(hf_collection, show_progress=False) assert mock_bar.call_args.kwargs["quiet"] is True + + +def test_filter_by_globs(): + """filter_by_globs() keeps or drops matches, accepts a str or list, and leaves its input untouched.""" + path_map = {"/a/0_dir/x.nc": 1, "/a/1_dir/y.nc": 2, "/a/2_dir/z.nc": 3} + assert list(filter_by_globs(path_map, "*/0_dir/*", True)) == ["/a/0_dir/x.nc"] + assert list(filter_by_globs(path_map, ["*/0_dir/*", "*/1_dir/*"], False)) == ["/a/2_dir/z.nc"] + assert filter_by_globs(path_map, [], True) == {} + assert filter_by_globs(path_map, [], False) == path_map + assert len(path_map) == 3 + + +def test_get_timestep_label_unknown_delta(): + """An unknown timestep duration is labelled 'unsorted' rather than guessed at.""" + assert get_timestep_label(None) == "unsorted" + + +@pytest.mark.parametrize("delta, label", [ + (timedelta(hours=1), "hour_1"), + (timedelta(hours=6), "hour_6"), + (timedelta(hours=23), "hour_23"), + (timedelta(hours=24), "day_1"), + (timedelta(days=27), "day_27"), + (timedelta(days=28), "month_1"), + (timedelta(days=31), "month_1"), + (timedelta(days=90), "month_3"), + (timedelta(days=335), "month_11"), + (timedelta(days=360), "year_1"), + (timedelta(days=365), "year_1"), + (timedelta(days=3650), "year_10"), +]) +def test_get_timestep_label(delta, label): + """get_timestep_label() maps a step duration to its frequency label, including the hour/day/month/year boundaries.""" + assert get_timestep_label(delta) == label diff --git a/gents/timeseries.py b/gents/timeseries.py index 9424580..f9f8bc2 100644 --- a/gents/timeseries.py +++ b/gents/timeseries.py @@ -7,13 +7,15 @@ """ import numpy as np import fnmatch +import glob from os.path import isfile from os import remove, makedirs, listdir from pathlib import Path from gents.meta import get_attributes, get_meta_from_path from gents.mhfdataset import MHFDataset +from gents.tsinventory import TSInventory from gents.datastore import GenTSDataStore -from gents.utils import get_version, LOG_LEVEL_IO_WARNING, ProgressBar +from gents.utils import get_version, get_timestep_label, LOG_LEVEL_IO_WARNING, ProgressBar from concurrent.futures import ProcessPoolExecutor, as_completed import traceback import logging @@ -377,34 +379,6 @@ def get_timestamp_format(dt, subhour_format="%Y%m%d%H%M%S", hourly_format="%Y%m% return time_format -def get_timestep_label(dt): - """ - Returns the frequency label for a time-step duration: ``'hour_N'``, - ``'day_N'``, ``'month_N'``, ``'year_N'``, or ``'unsorted'`` if unknown. - - Used by :meth:`TSCollection.append_timestep_dirs` as a directory name. - - :param dt: Duration of a single model time step, or ``None`` if unknown. - :type dt: datetime.timedelta or None - :rtype: str - """ - if dt is None: - return "unsorted" - - hours = np.rint(dt.total_seconds() / 60.0 / 60.0) - days = np.rint(hours / 24.0) - months = np.rint(days / 30) - years = np.rint(months / 12) - - if hours < 24: - return f"hour_{int(hours)}" - elif days < 28: - return f"day_{int(days)}" - elif months < 12: - return f"month_{int(months)}" - return f"year_{int(years)}" - - def _matches_any(value, globs): """ Returns whether ``value`` matches at least one ``fnmatch`` glob. @@ -822,27 +796,29 @@ def skip_existing(self): Returns a new collection with each order narrowed to timesteps not already covered by existing output, for resuming a continued run. + Existing output is found with a :class:`~gents.tsinventory.TSInventory` + of each order's output directory. Only files that load and carry the + ``gents_version`` stamp count as coverage; incomplete or corrupt files + are ignored with a warning. + :rtype: TSCollection """ + inventories = {} # parent dir -> unpulled inventory, so each directory is walked once + latest_times = {} # (parent dir, file prefix, variable) -> latest covered time, so each file is read once new_orders = [] for order_dict in copy.deepcopy(self.__orders): template = Path(order_dict["ts_path_template"]) var_name = order_dict["primary_var"] - name_glob = f"{template.name}.{var_name}.*.nc" - latest_existing_time = None - - if template.parent.is_dir(): - for name in listdir(template.parent): - if not fnmatch.fnmatch(name, name_glob): - continue - candidate_path = str(template.parent / name) - if not check_timeseries_integrity(candidate_path): - continue - ts_times = get_meta_from_path(candidate_path).get_float_times() - ts_times = np.ma.getdata(np.atleast_1d(ts_times)) - - if latest_existing_time is None or ts_times[-1] > latest_existing_time: - latest_existing_time = ts_times[-1] + key = (template.parent, template.name, var_name) + + if key not in latest_times: + if template.parent not in inventories: + inventories[template.parent] = TSInventory(str(template.parent)) + prefix = glob.escape(str(template.parent / f"{template.name}.{var_name}")) + existing = inventories[template.parent].include(f"{prefix}.*.nc") + existing.pull_metadata(show_progress=False) + latest_times[key] = existing.get_latest_time() + latest_existing_time = latest_times[key] if latest_existing_time is not None: original_hf_paths = order_dict["hf_paths"] diff --git a/gents/tsinventory.py b/gents/tsinventory.py new file mode 100644 index 0000000..c3e0db5 --- /dev/null +++ b/gents/tsinventory.py @@ -0,0 +1,359 @@ +#!/usr/bin/env python +""" +Discovery, filtering, and indexing of existing GenTS time series files. + +Developer: Cameron Cummins +Contact: cameron.cummins@utexas.edu +""" +from gents.datastore import GenTSDataStore +from gents.hfcollection import find_files, get_group_timestep_delta +from gents.meta import netCDFMeta, read_meta_map, split_valid +from gents.utils import filter_by_globs, get_timestep_label +import numpy as np +import logging + +logger = logging.getLogger(__name__) + + +def get_ts_meta_from_path(path): + """ + Reads a time series file's header into a :class:`~gents.meta.netCDFMeta`. + + Unlike :func:`gents.meta.get_meta_from_path`, keeps per-variable attributes + and skips dimension bounds, which only history-file tiling needs. Picklable, + for process pools. + + :param path: Path to the time series file. + :type path: str + :rtype: gents.meta.netCDFMeta + """ + try: + with GenTSDataStore(path, "r") as ds: + return netCDFMeta(ds, path, decode_dates=False, compute_dim_bounds=False) + except Exception as e: + raise type(e)(f"{e} Path: {path}") from e + + +def is_gents_output(ts_meta): + """ + Returns whether a file was produced, and completely written, by GenTS. + + The opposite of :meth:`gents.meta.netCDFMeta.is_valid`, which treats the + ``gents_version`` attribute as disqualifying. GenTS stamps it last, so it + also marks a finished write. + + :param ts_meta: Metadata for the time series file. + :type ts_meta: gents.meta.netCDFMeta + :rtype: bool + """ + return "gents_version" in ts_meta.get_attributes() + + +def get_meta_timestep_label(ts_meta): + """ + Returns a file's frequency label (``'month_1'``, ``'day_1'``, ...). + + Derived from the file's own time coordinate. Fewer than two time steps, or + no usable time coordinate, gives ``'unsorted'``. + + :param ts_meta: Metadata for the time series file. + :type ts_meta: gents.meta.netCDFMeta + :rtype: str + """ + try: + delta = get_group_timestep_delta([ts_meta]) + except (ValueError, TypeError): + return get_timestep_label(None) + return get_timestep_label(delta) + + +def get_time_sort_key(ts_meta): + """ + Returns a sort key ordering time series files by earliest time value. + + Compares raw float times, which are monotonic with their decoded dates. + Files with no usable time coordinate sort last rather than raising. + + :param ts_meta: Metadata for the time series file. + :type ts_meta: gents.meta.netCDFMeta + :rtype: tuple[int, float] + """ + float_times = ts_meta.get_float_times() + if float_times is None: + return (1, 0.0) + + values = np.ma.getdata(np.atleast_1d(float_times)) + if values.shape[0] == 0: + return (1, 0.0) + return (0, float(np.min(values))) + + +class TSInventory: + """ + An index of the GenTS time series files that already exist on disk. + + Mirrors :class:`~gents.hfcollection.HFCollection` for output instead of + input: discover files, filter them by path, pull their metadata lazily, then + group them by variable. Filters return new inventories; only + :meth:`pull_metadata` and :meth:`check_validity` mutate in place. + """ + + def __init__(self, ts_head_dir, num_processes=1, ts_glob_pattern="*.nc*", meta_map=None, ts_var_map=None): + """ + A missing or empty directory is not an error, since a first run has no + output yet; it gives an empty inventory. + + :param ts_head_dir: Root directory to search for time series files. + :type ts_head_dir: str + :param num_processes: Worker processes used for parallel metadata reads. + :type num_processes: int + :param ts_glob_pattern: ``fnmatch`` pattern used to discover files. + :type ts_glob_pattern: str + :param meta_map: Pre-populated ``{path: netCDFMeta}`` mapping, from :meth:`copy`. + Skips discovery. + :type meta_map: dict or None + :param ts_var_map: Pre-built variable grouping, from :meth:`copy`. Stale + entries are pruned by :meth:`get_variables`. + :type ts_var_map: dict or None + """ + self.__num_processes = num_processes + self.__ts_dir = ts_head_dir + self.__ts_glob_pattern = ts_glob_pattern + self.__ts_var_map = ts_var_map + + if meta_map is None: + self.__ts_to_meta_map = {path: None for path in find_files(ts_head_dir, ts_glob_pattern)} + logger.info(f"Initialized TSInventory at '{ts_head_dir}'") + logger.info(f"{len(self.__ts_to_meta_map)} netCDF files found.") + else: + self.__ts_to_meta_map = meta_map + + def __getitem__(self, key): + return self.__ts_to_meta_map[key] + + def __contains__(self, key): + return key in self.__ts_to_meta_map + + def __iter__(self): + return iter(self.__ts_to_meta_map) + + def __len__(self): + return len(self.__ts_to_meta_map) + + def items(self): + return self.__ts_to_meta_map.items() + + def values(self): + return self.__ts_to_meta_map.values() + + def keys(self): + return self.__ts_to_meta_map.keys() + + def include(self, glob_patterns): + """ + Returns a new inventory holding only files matching at least one pattern. + + Patterns are ``fnmatch`` globs tested against path strings; an empty list + matches nothing and so empties the inventory. + + :param glob_patterns: One or more glob patterns; a single string is also + accepted. + :type glob_patterns: list[str] or str + :rtype: TSInventory + """ + filtered_path_map = filter_by_globs(self.__ts_to_meta_map, glob_patterns, keep_matches=True) + logger.debug(f"Inclusive filter(s) applied: '{glob_patterns}'") + return self.copy(meta_map=filtered_path_map) + + def exclude(self, glob_patterns): + """ + Returns a new inventory with files matching any pattern removed. + + Patterns are ``fnmatch`` globs tested against path strings. + + :param glob_patterns: One or more glob patterns; a single string is also + accepted. + :type glob_patterns: list[str] or str + :rtype: TSInventory + """ + filtered_path_map = filter_by_globs(self.__ts_to_meta_map, glob_patterns, keep_matches=False) + logger.debug(f"Exclusive filter(s) applied: '{glob_patterns}'") + return self.copy(meta_map=filtered_path_map) + + def is_pulled(self): + """ + Returns whether metadata has been loaded for every file. + + :rtype: bool + """ + return all(meta is not None for meta in self.__ts_to_meta_map.values()) + + def get_input_dir(self): + """ + Returns the head directory this inventory was initialised from. + + :rtype: str + """ + return self.__ts_dir + + def check_pulled(self): + """ + Pulls metadata if it has not been pulled already. + """ + if not self.is_pulled(): + self.pull_metadata() + + def copy(self, num_processes=None, meta_map=None, ts_var_map=None): + """ + Returns a new inventory derived from this one, with optional overrides. + + Shares the head directory and glob pattern; arguments left ``None`` are + inherited. Every filter returns through here, keeping the API immutable. + + :param num_processes: Worker process count for the copy. + :type num_processes: int or None + :param meta_map: Metadata map to assign to the copy. + :type meta_map: dict or None + :param ts_var_map: Variable grouping to assign to the copy. Inheriting is + safe when ``meta_map`` narrows, since :meth:`get_variables` prunes + stale entries; a caller widening it must pass a grouping instead. + :type ts_var_map: dict or None + :rtype: TSInventory + """ + if num_processes is None: + num_processes = self.__num_processes + if meta_map is None: + meta_map = self.__ts_to_meta_map + if ts_var_map is None: + ts_var_map = self.__ts_var_map + return TSInventory( + self.__ts_dir, + num_processes=num_processes, + ts_glob_pattern=self.__ts_glob_pattern, + meta_map=meta_map, + ts_var_map=ts_var_map + ) + + def check_validity(self): + """ + Drops files whose metadata is missing or not :func:`is_gents_output`, + warning about each. + + :returns: The removed ``{path: metadata}`` entries. + :rtype: dict + """ + logger.debug(f"Validating metadata...") + new_map, removed = split_valid(self.__ts_to_meta_map, is_gents_output) + self.__ts_to_meta_map = new_map + self.__ts_var_map = None + logger.debug(f"{len(new_map)} files validated ({len(removed)} removed).") + return removed + + def pull_metadata(self, check_valid=True, raise_errors=False, show_progress=True): + """ + Reads the header of every time series file in the inventory. + + Runs :func:`get_ts_meta_from_path` over a process pool when + ``num_processes > 1``, serially otherwise. + + :param check_valid: Run :meth:`check_validity` afterwards, dropping files + GenTS did not produce or did not finish writing. + :type check_valid: bool + :param raise_errors: Raise per-file failures instead of logging them. + :type raise_errors: bool + :param show_progress: If ``False``, suppress the stdout progress bar. + :type show_progress: bool + :returns: This inventory, mutated in place, so the call can be chained. + :rtype: TSInventory + """ + logger.info(f"Pulling metadata...") + self.__ts_var_map = None + paths = list(self.__ts_to_meta_map.keys()) + + self.__ts_to_meta_map.update(read_meta_map( + paths, get_ts_meta_from_path, self.__num_processes, raise_errors, show_progress + )) + + if check_valid: + self.check_validity() + else: + logger.warning(f"Skipping metadata validation may result in errors due to missing attributes or coordinate data.") + logger.info(f"Metadata pulled.") + return self + + def get_variables(self): + """ + Groups the inventory's files by variable and frequency. + + Keys are ``'.'``, e.g. ``'TS.month_1'``. A file + holding several primary variables is listed under each; one holding none + is skipped with a warning. + + Paths are ordered by earliest time value. A grouping inherited from a + filtered copy is pruned rather than rebuilt, which preserves that order. + + :returns: ``{'.': [pathlib.Path]}``. + :rtype: dict[str, list[pathlib.Path]] + """ + if self.__ts_var_map is None: + self.check_pulled() + + grouped = {} + for path, meta in self.__ts_to_meta_map.items(): + primary_vars = meta.get_primary_variables() + + if len(primary_vars) == 0: + logger.warning(f"No primary variable found for '{path}', skipping.") + continue + if len(primary_vars) > 1: + logger.warning( + f"Expected one primary variable per time series file, " + f"found {len(primary_vars)} in '{path}': {primary_vars}." + ) + + timestep_label = get_meta_timestep_label(meta) + for var_name in primary_vars: + grouped.setdefault(f"{var_name}.{timestep_label}", []).append((path, meta)) + + self.__ts_var_map = { + key: [path for path, _ in sorted(entries, key=lambda entry: get_time_sort_key(entry[1]))] + for key, entries in grouped.items() + } + logger.debug(f"{len(self.__ts_var_map)} variable groups built.") + else: + pruned = {} + reassign = False + for key, paths in self.__ts_var_map.items(): + kept = [path for path in paths if path in self.__ts_to_meta_map] + if len(paths) != len(kept): + reassign = True + if len(kept) > 0: + pruned[key] = kept + + if reassign: + self.__ts_var_map = pruned + + return self.__ts_var_map + + def get_latest_time(self): + """ + Returns the latest raw time value covered by any file in the inventory, + pulling metadata first if necessary. + + Raw floats are comparable only between files sharing ``units`` and + ``calendar``, as the output of one order's template and variable does. + + :returns: The latest time value, or ``None`` if no file has a usable time + coordinate (including an empty inventory). + :rtype: float or None + """ + self.check_pulled() + latest = None + for meta in self.__ts_to_meta_map.values(): + float_times = meta.get_float_times() + if float_times is None: + continue + ts_times = np.ma.getdata(np.atleast_1d(float_times)) + if ts_times.shape[0] > 0 and (latest is None or ts_times[-1] > latest): + latest = ts_times[-1] + return latest diff --git a/gents/utils.py b/gents/utils.py index d2b01f7..4c4e546 100644 --- a/gents/utils.py +++ b/gents/utils.py @@ -1,6 +1,6 @@ #!/usr/bin/env python """ -Logging setup, progress reporting, versioning, and collection summaries. +Logging setup, progress reporting, path filtering, time step labels, versioning, and collection summaries. Developer: Cameron Cummins Contact: cameron.cummins@utexas.edu @@ -10,11 +10,66 @@ import logging import sys import datetime +import fnmatch import numpy as np LOG_LEVEL_IO_WARNING = 5 +def filter_by_globs(path_map, glob_patterns, keep_matches): + """ + Filters a ``{path: value}`` map by ``fnmatch`` globs tested against the + path as a string. + + Shared by the collections' ``include`` (``keep_matches=True``) and + ``exclude`` (``keep_matches=False``). An empty pattern list matches nothing. + + :param path_map: Mapping keyed by path; not modified. + :type path_map: dict + :param glob_patterns: One or more glob patterns; a single string is also accepted. + :type glob_patterns: list[str] or str + :param keep_matches: Keep paths matching any pattern if ``True``, drop them if ``False``. + :type keep_matches: bool + :returns: A new dict holding the retained entries. + :rtype: dict + """ + if isinstance(glob_patterns, str): + glob_patterns = [glob_patterns] + return { + path: value for path, value in path_map.items() + if any(fnmatch.fnmatch(str(path), pattern) for pattern in glob_patterns) == keep_matches + } + + +def get_timestep_label(dt): + """ + Returns the frequency label for a time-step duration: ``'hour_N'``, + ``'day_N'``, ``'month_N'``, ``'year_N'``, or ``'unsorted'`` if unknown. + + Used by :meth:`~gents.timeseries.TSCollection.append_timestep_dirs` as a + directory name, and by :class:`~gents.tsinventory.TSInventory` to group files. + + :param dt: Duration of a single model time step, or ``None`` if unknown. + :type dt: datetime.timedelta or None + :rtype: str + """ + if dt is None: + return "unsorted" + + hours = np.rint(dt.total_seconds() / 60.0 / 60.0) + days = np.rint(hours / 24.0) + months = np.rint(days / 30) + years = np.rint(months / 12) + + if hours < 24: + return f"hour_{int(hours)}" + elif days < 28: + return f"day_{int(days)}" + elif months < 12: + return f"month_{int(months)}" + return f"year_{int(years)}" + + def get_time_stamp(): """ Returns the current date and time as a ``'YYYY-MM-DD HH:MM'`` string.