deepest_paths.py

"""Resolve the two directories that vary between machines: the Zenodo data tree and the baseline
run output.

Each resolves as environment variable, then <repo>/user.yaml, then a repository-relative default.

Run this file to print what is currently in effect.
"""
import os
from pathlib import Path

import yaml

#: setting name -> (environment variable, repo-relative default)
SETTINGS = {
    "data": ("DEEPEST_DATA", "data"),
    "baseline_out": ("DEEPEST_BASELINE_OUT", "baseline_run"),
}

_CONFIG_NAME = "user.yaml"


def repo_root():
    """The repository root, which is this file's own directory."""
    here = Path(__file__).resolve().parent
    env = os.environ.get("DEEPEST_REPO")
    if not env:
        return here
    root = Path(env).expanduser().resolve()
    if root != here:
        # Otherwise code comes from one checkout and data/ from another.
        raise ValueError(
            f"DEEPEST_REPO is {root}, but this module was loaded from {here}. "
            f"Unset DEEPEST_REPO, or point it at the checkout you are actually running."
        )
    return root


def _load_file():
    """Parse user.yaml if there is one. Absent is fine; malformed is not."""
    explicit = os.environ.get("DEEPEST_USER_CONFIG")
    path = Path(explicit).expanduser() if explicit else repo_root() / _CONFIG_NAME
    if not path.is_file():
        # An absent <repo>/user.yaml is normal. A DEEPEST_USER_CONFIG that points at nothing is a typo.
        if explicit:
            raise FileNotFoundError(
                f"DEEPEST_USER_CONFIG points at {path}, which does not exist. "
                f"Fix the path or unset it to use {repo_root() / _CONFIG_NAME}."
            )
        return {}, path

    with open(path) as fh:
        loaded = yaml.safe_load(fh)
    if loaded is None:
        return {}, path
    if not isinstance(loaded, dict):
        raise ValueError(f"{path} must contain a mapping of setting names to paths, "
                         f"got {type(loaded).__name__}")
    unknown = set(loaded) - set(SETTINGS)
    if unknown:
        raise ValueError(f"{path} has unknown setting(s) {sorted(unknown)}; "
                         f"expected any of {sorted(SETTINGS)}")
    return loaded, path


def get(name):
    """Resolve one setting to an absolute Path, without checking that it exists."""
    if name not in SETTINGS:
        raise KeyError(f"unknown setting {name!r}; expected one of {sorted(SETTINGS)}")
    env_var, default = SETTINGS[name]
    root = repo_root()

    # Set but empty is an unset shell variable, not a request for the default.
    value = os.environ.get(env_var)
    if value is not None and not value.strip():
        raise ValueError(
            f"{env_var} is set but empty. Unset it to use the default, or give it a path."
        )
    if value is None:
        cfg = _load_file()[0]
        value = cfg.get(name)
        if value is not None and (not isinstance(value, str) or not value.strip()):
            raise ValueError(
                f"{name!r} in {_load_file()[1]} must be a non-empty path string, got {value!r}"
            )
    if value is None:
        value = default

    path = Path(value).expanduser()
    return path if path.is_absolute() else (root / path).resolve()


def require(name, needs=()):
    """Resolve a setting, checking it exists and contains every entry in `needs`."""
    path = get(name)
    env_var = SETTINGS[name][0]
    _, cfg = _load_file()
    if not path.is_dir():
        raise FileNotFoundError(
            f"the {name!r} directory does not exist: {path}\n"
            f"Set it in {cfg}, or export {env_var}=<path>, or place it at the default "
            f"<repo>/{SETTINGS[name][1]}. See user.example.yaml."
        )
    missing = [n for n in needs if not (path / n).exists()]
    if missing:
        raise FileNotFoundError(
            f"the {name!r} directory {path} is missing: {', '.join(missing)}\n"
            f"That is not the directory this expects. Set {env_var} or {name} in {cfg} to the "
            f"directory that holds {', '.join(needs)}."
        )
    return path


def data(**kw):
    """The Zenodo data tree."""
    return require("data", **kw)


def baseline_out(**kw):
    """Where the baseline run writes. Created on demand."""
    path = get("baseline_out")
    path.mkdir(parents=True, exist_ok=True)
    return path


def rel(p):
    """Path relative to the repository, so printed output carries no machine-specific prefix."""
    p = Path(p)
    try:
        return p.relative_to(repo_root())
    except ValueError:
        return p


def describe():
    """One line per setting."""
    _, cfg = _load_file()
    lines = [f"repo: {repo_root()}", f"config: {cfg}{'' if cfg.is_file() else ' (absent, using defaults)'}"]
    for name in SETTINGS:
        lines.append(f"{name}: {get(name)}")
    return "\n".join(lines)


if __name__ == "__main__":
    print(describe())