From 7f04da32d9c95c2b61b17ad8847b42117f32c0ad Mon Sep 17 00:00:00 2001 From: wpbonelli Date: Fri, 2 May 2025 15:33:34 -0400 Subject: [PATCH] refactor(LocalRegistry): back to the old indexing pattern Sometimes you end up right where you started, a bit wiser for your travails --- modflow_devtools/models.py | 101 +++++++++++++++---------------------- 1 file changed, 42 insertions(+), 59 deletions(-) diff --git a/modflow_devtools/models.py b/modflow_devtools/models.py index 3beedbb6..58dd26cb 100644 --- a/modflow_devtools/models.py +++ b/modflow_devtools/models.py @@ -5,7 +5,7 @@ import hashlib import importlib.resources as pkg_resources from abc import ABC, abstractmethod -from collections.abc import Callable, Iterable +from collections.abc import Callable from functools import partial from os import PathLike from pathlib import Path @@ -96,73 +96,56 @@ class LocalRegistry(ModelRegistry): exclude: ClassVar = [".DS_Store", "compare"] - def __init__( + def __init__(self) -> None: + self._paths: set[Path] = set() + self._files: dict[str, Path] = {} + self._models: dict[str, list[Path]] = {} + self._examples: dict[str, list[str]] = {} + + def index( self, - path: str | PathLike | Iterable[str | PathLike], - namefile_pattern: str = "mfsim.nam", + path: str | PathLike, + prefix: str = "", + namefile: str = "mfsim.nam", ): """ - Create a registry from models under the given - directory path. + Add models found under the given path to the registry. + + Call this once or more to prepare a registry. If called on the same + `path` again, the models will be reloaded — thus this method + is idempotent and may be used to reload the registry e.g. if model + files have changed since the registry was created. The `path` may consist of model subdirectories at arbitrary depth. Model input subdirectories are identified by the presence of a namefile matching `namefile_pattern`. """ - # check if path is iterable of str\pathlike - if isinstance(path, Iterable) and not isinstance(path, str): - path = [Path(p).expanduser().resolve().absolute() for p in path] # type: ignore - missing = [p for p in path if not p.is_dir()] # type: ignore - if any(missing): - missing = [str(p) for p in missing] # type: ignore - raise NotADirectoryError( - f"Directory paths not found: {', '.join(missing)}" # type: ignore - ) - self._path = path - else: - path = Path(path).expanduser().resolve().absolute() - if not path.is_dir(): - raise NotADirectoryError(f"Directory path not found: {path}") - self._path = [path] - self._files: dict[str, dict[str, str | None]] = {} - self._models: dict[str, list[str]] = {} - self._examples: dict[str, list[str]] = {} - self.namefile_pattern = namefile_pattern - self.index() - def index(self): - """ - Build the registry from models found under the configured path. This - method can be called to reload the registry e.g. if model files have - changed since the registry was created. - """ + path = Path(path).expanduser().resolve().absolute() + if not path.is_dir(): + raise NotADirectoryError(f"Directory path not found: {path}") + self._paths.add(path) - self._files = {} - self._models = {} - self._examples = {} - for path in self._path: - model_paths = get_model_paths(path, namefile=self.namefile_pattern) - for model_path in model_paths: - model_path = model_path.expanduser().resolve().absolute() - rel_path = model_path.relative_to(path) - model_name = "/".join(rel_path.parts) - self._models[model_name] = [] - if len(rel_path.parts) > 1: - name = rel_path.parts[0] - if name not in self._examples: - self._examples[name] = [] - self._examples[name].append(model_name) - for p in model_path.rglob("*"): - if not p.is_file() or any( - e in p.name for e in LocalRegistry.exclude - ): - continue - relpath = p.expanduser().absolute().relative_to(path) - name = "/".join(relpath.parts) - hash = _sha256(p) - self._files[name] = {"hash": hash, "path": p, "relpath": relpath} - self._models[model_name].append(p) + model_paths = get_model_paths(path, namefile=namefile) + for model_path in model_paths: + model_path = model_path.expanduser().resolve().absolute() + rel_path = model_path.relative_to(path) + parts = [prefix, *list(rel_path.parts)] if prefix else list(rel_path.parts) + model_name = "/".join(parts) + self._models[model_name] = [] + if len(rel_path.parts) > 1: + name = rel_path.parts[0] + if name not in self._examples: + self._examples[name] = [] + self._examples[name].append(model_name) + for p in model_path.rglob("*"): + if not p.is_file() or any(e in p.name for e in LocalRegistry.exclude): + continue + relpath = p.expanduser().absolute().relative_to(path) + name = "/".join(relpath.parts) + self._files[name] = p + self._models[model_name].append(p) def copy_to( self, workspace: str | PathLike, model_name: str, verbose: bool = False @@ -192,8 +175,8 @@ def copy_to( return workspace @property - def path(self) -> list[Path]: - return self._path # type: ignore + def paths(self) -> set[Path]: + return self._paths @property def files(self) -> dict: