Added TSInventory API and improved performance of TSCollection.skip_existing()

docs/api.rst

@@ -10,6 +10,9 @@ API
 .. automodule:: gents.timeseries
    :members:
 
+.. automodule:: gents.tsinventory
+   :members:
+
 .. automodule:: gents.mhfdataset
    :members:
 

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 `<var>.<frequency>`, `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.

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

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

docs/llm/index.md

@@ -41,22 +41,23 @@ 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.
     case_builder.py       gents_conform_build — mirror a case dir as tiny clones
     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 <hf_dir> --model CESM3 --dryrun   # CLI dry run (read-only validation)
 ```

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
 ```
 

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 ``<variable>.<frequency>`` (e.g. ``TS.month_1``) and its paths are ordered by start time.
+
 Custom Dask Workflows with ``TSCollection``
 -------------------------------------------
 

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)