"""Centralized gin-config loader for multi-config pipelines. Reference implementation of the recommended pattern: - One function `load_all_configs()` parses ALL .gin files at once - `gin.clear_config()` resets global state before loading - Individual `get_*_cfg()` kept ONLY for isolated testing This file serves as a TEMPLATE — copy and adapt for each project. """ from __future__ import annotations import logging from pathlib import Path from typing import Any import gin logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Import config classes (adapt to your project) # --------------------------------------------------------------------------- # from conf.pipeline_conf import PipelineConfig # from conf.hardware_conf import HardwareConfig # from conf.models_conf import ModelsConfig # from conf.input_conf import InputConfig # from conf.seg_conf import SegConfig # --------------------------------------------------------------------------- # Central loader (PRODUCTION USE) # --------------------------------------------------------------------------- def load_all_configs(path2cfg: str) -> dict[str, Any]: """Parse ALL .gin files at once and instantiate all config objects. This is the **only** function that should call gin.parse in production. It clears gin global state first, then loads all .gin files in sorted order, then creates config instances from the fully-populated state. Args: path2cfg: Path to config directory (WITH trailing slash). Example: ``"/home/servml/project/in/config_files/"`` Returns: Dictionary with config objects keyed by short name:: { "pipeline": PipelineConfig(), "hardware": HardwareConfig(), "models": ModelsConfig(), "input": InputConfig(), "seg": SegConfig(), } Raises: FileNotFoundError: If path2cfg does not exist or contains no .gin files. Example:: configs = load_all_configs("/home/servml/project/in/config_files/") run_pipeline(configs["pipeline"], configs["hardware"], ...) """ cfg_dir = Path(path2cfg) if not cfg_dir.is_dir(): raise FileNotFoundError(f"Config directory not found: {cfg_dir}") gin_files = sorted(cfg_dir.glob("*.gin")) if not gin_files: raise FileNotFoundError(f"No .gin files in {cfg_dir}") # CRITICAL: clear global gin state before loading. # Without this, parameters from previous calls accumulate silently. gin.clear_config() # Parse ALL .gin files in one call — order is alphabetical (sorted). gin.parse_config_files_and_bindings( config_files=[str(f) for f in gin_files], bindings=[], # No CLI overrides; add if needed. ) logger.info("Loaded %d gin files from %s", len(gin_files), cfg_dir) # Instantiate all configs from the fully-populated gin state. # Each class picks its parameters from gin via @gin.configurable. configs = { "pipeline": PipelineConfig(), "hardware": HardwareConfig(), "models": ModelsConfig(), "input": InputConfig(), "seg": SegConfig(), } logger.info("Created %d config objects: %s", len(configs), list(configs.keys())) return configs # --------------------------------------------------------------------------- # Individual loaders (TESTING / DEBUGGING ONLY) # --------------------------------------------------------------------------- # Use these when you need a single config in isolation (unit tests, notebooks). # ALWAYS call gin.clear_config() first to avoid state leaks. def get_pipeline_cfg(path2cfg: str) -> Any: # -> PipelineConfig """Load ONLY pipeline config (for isolated testing).""" gin.clear_config() gin.parse_config_file(f"{path2cfg}pipeline.gin") return PipelineConfig() def get_hardware_cfg(path2cfg: str) -> Any: # -> HardwareConfig """Load ONLY hardware config (for isolated testing).""" gin.clear_config() gin.parse_config_file(f"{path2cfg}hardware.gin") return HardwareConfig() def get_models_cfg(path2cfg: str) -> Any: # -> ModelsConfig """Load ONLY models config (for isolated testing).""" gin.clear_config() gin.parse_config_file(f"{path2cfg}models.gin") return ModelsConfig() def get_input_cfg(path2cfg: str) -> Any: # -> InputConfig """Load ONLY input config (for isolated testing).""" gin.clear_config() gin.parse_config_file(f"{path2cfg}input.gin") return InputConfig() def get_seg_cfg(path2cfg: str) -> Any: # -> SegConfig """Load ONLY segmentation config (for isolated testing).""" gin.clear_config() gin.parse_config_file(f"{path2cfg}segmentation.gin") return SegConfig() # --------------------------------------------------------------------------- # Project root discovery # --------------------------------------------------------------------------- # Markers that indicate the project root directory. # Checked in order: first match wins. _ROOT_MARKERS = ("pyproject.toml", ".git", "in") def get_proj_dir() -> str: """Return project root directory with trailing slash. Walks up from this file's location until a directory containing one of the root markers is found. More robust than the legacy ``split('src')[0]`` approach which breaks when ``src`` appears elsewhere in the path (e.g. ``/home/user/resources/project/src/``). Root markers (checked in order): ``pyproject.toml``, ``.git``, ``in/``. Returns: Absolute path to project root with trailing ``/``. Raises: RuntimeError: If no marker is found within 10 parent levels. """ current = Path(__file__).resolve().parent for _ in range(10): if any((current / marker).exists() for marker in _ROOT_MARKERS): return str(current) + "/" current = current.parent raise RuntimeError( f"Cannot find project root: none of {_ROOT_MARKERS} found " f"in ancestors of {Path(__file__).resolve()}" ) def main() -> None: """Template entry point: load all configs -> run pipeline.""" proj_dir = get_proj_dir() path2cfg = f"{proj_dir}in/config_files/" configs = load_all_configs(path2cfg) # Unpack and pass explicitly to pipeline: # run_pipeline( # configs["pipeline"], # configs["hardware"], # configs["models"], # configs["input"], # configs["seg"], # ) # Print loaded configs for verification: for name, cfg in configs.items(): print(f"\n[{name}]") for k, v in vars(cfg).items(): print(f" {k} = {v}") if __name__ == "__main__": main()