Utilities API

Small helpers in ecgen.utils.

Config

resolve_config

resolve_config(config_cls: Type[T], config: Optional[T | dict[str, Any]], flat_kwargs: dict[str, Any]) -> T

Build a config dataclass for a LightningModule constructor.

Wrappers save their config as flat hyperparameters, so load_from_checkpoint calls __init__(**hparams) with one keyword per config field instead of config=.... This accepts either form.

Parameters:

Name	Type	Description	Default
config_cls	Type[T]	Config dataclass type.	required
config	Optional[T | dict[str, Any]]	A config instance, a dict of its fields, or None.	required
flat_kwargs	dict[str, Any]	Extra keyword arguments passed to __init__.	required

Returns:

Type	Description
T	A config_cls instance. An explicit config takes precedence over
T	flat_kwargs (Lightning passes both when config= is given to
T	load_from_checkpoint).

Source code in src/ecgen/utils/config.py

def resolve_config(
    config_cls: Type[T],
    config: Optional[T | dict[str, Any]],
    flat_kwargs: dict[str, Any],
) -> T:
    """
    Build a config dataclass for a LightningModule constructor.

    Wrappers save their config as flat hyperparameters, so
    ``load_from_checkpoint`` calls ``__init__(**hparams)`` with one keyword per
    config field instead of ``config=...``. This accepts either form.

    Args:
        config_cls: Config dataclass type.
        config: A config instance, a dict of its fields, or None.
        flat_kwargs: Extra keyword arguments passed to ``__init__``.

    Returns:
        A ``config_cls`` instance. An explicit ``config`` takes precedence over
        ``flat_kwargs`` (Lightning passes both when ``config=`` is given to
        ``load_from_checkpoint``).
    """
    valid = {f.name for f in fields(config_cls)}
    unknown = set(flat_kwargs) - valid
    if unknown:
        raise TypeError(f"Unexpected {config_cls.__name__} fields: {sorted(unknown)}")

    if config is None:
        return config_cls(**flat_kwargs)
    if isinstance(config, dict):
        return config_cls(**config)
    return config

Seed

set_global_seed

set_global_seed(seed: int, deterministic: Optional[bool] = None) -> None

Set seeds for Python, NumPy, and PyTorch (if available).

Source code in src/ecgen/utils/seed.py

def set_global_seed(seed: int, deterministic: Optional[bool] = None) -> None:
    """
    Set seeds for Python, NumPy, and PyTorch (if available).
    """

    random.seed(seed)
    os.environ["PYTHONHASHSEED"] = str(seed)
    np.random.seed(seed)

    try:
        import torch

        torch.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
        if deterministic is True:
            torch.backends.cudnn.deterministic = True
            torch.backends.cudnn.benchmark = False
    except ImportError:
        # PyTorch not installed; skip torch-specific seeding.
        return

I/O

read_yaml

read_yaml(path: str | Path) -> Any

Load a YAML file and return the parsed object.

Source code in src/ecgen/utils/io.py

def read_yaml(path: str | Path) -> Any:
    """
    Load a YAML file and return the parsed object.
    """

    with Path(path).open("r") as f:
        return yaml.safe_load(f)

write_yaml

write_yaml(obj: Any, path: str | Path) -> None

Serialize an object to YAML.

Source code in src/ecgen/utils/io.py

def write_yaml(obj: Any, path: str | Path) -> None:
    """
    Serialize an object to YAML.
    """

    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with p.open("w") as f:
        yaml.safe_dump(obj, f)

write_json

write_json(obj: Any, path: str | Path) -> None

Serialize an object to JSON.

Source code in src/ecgen/utils/io.py

def write_json(obj: Any, path: str | Path) -> None:
    """
    Serialize an object to JSON.
    """

    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    with p.open("w") as f:
        json.dump(obj, f, indent=2, sort_keys=True)

Logging

get_logger

get_logger(name: Optional[str] = None) -> logging.Logger

Get a logger with a simple, consistent format.

Source code in src/ecgen/utils/logging.py

def get_logger(name: Optional[str] = None) -> logging.Logger:
    """
    Get a logger with a simple, consistent format.
    """

    logger = logging.getLogger(name)
    if not logging.getLogger().handlers:
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s | %(name)s | %(levelname)s | %(message)s",
        )
    return logger

Run metadata

collect_run_metadata

collect_run_metadata(*, config_path: Path | None, cfg: Dict[str, Any], run_dir: Path, argv: list[str] | None = None) -> Dict[str, Any]

Collect lightweight metadata about the current run.

Source code in src/ecgen/utils/metadata.py

def collect_run_metadata(
    *,
    config_path: Path | None,
    cfg: Dict[str, Any],
    run_dir: Path,
    argv: list[str] | None = None,
) -> Dict[str, Any]:
    """
    Collect lightweight metadata about the current run.
    """

    py_version = sys.version.replace("\n", " ")

    metadata: Dict[str, Any] = {
        "command": " ".join(argv or sys.argv),
        "config_path": str(config_path.resolve()) if config_path else None,
        "run_dir": str(run_dir.resolve()),
        "python": {
            "version": py_version,
            "executable": sys.executable,
            "platform": platform.platform(),
        },
        "packages": {
            "torch": _safe_import_version("torch"),
            "pytorch_lightning": _safe_import_version("pytorch-lightning"),
            "ecgen": _safe_import_version("ecgen"),
        },
        "env": {
            "CUDA_VISIBLE_DEVICES": os.environ.get("CUDA_VISIBLE_DEVICES"),
        },
        "git": _git_info(config_path.resolve() if config_path else Path.cwd()),
    }

    # Store a shallow copy of the experiment section for quick reference
    exp_cfg = cfg.get("experiment", {})
    metadata["experiment"] = {
        "name": exp_cfg.get("name"),
        "seed": exp_cfg.get("seed"),
    }

    return metadata