API reference¶

def app_recipe(
    name: str,
    requirements: Iterable[str],
    imports: Iterable[str],
    *,
    prefix: str,
    tools_home: str,
    cache: Path | str,
    manager: str = "",
) -> Recipe:
    """The recipe that gives a host the environment `requirements` run in here.

    `requirements` are the packages of the app, each `name` or `name[extras]`: one imported
    from a source checkout goes as a wheel of its code as it is now, one that is installed is
    asked for in the version it has. `imports` are the modules that have to import there for
    the environment to count as usable. `prefix` and `tools_home` are paths on the host (see
    `Host.home`); `cache` is a directory of this machine for the wheels.

    The environment is updated where it stands for as long as the Python is the same. Builds
    wheels the first time, and after the code changes: keep it off UI threads.
    """
    prefix, cache = str(prefix).rstrip("/"), Path(cache)
    code, wheels, wanted, tags = f"{prefix}/{_CODE}", [], [], []
    for requirement in requirements:
        dist, extras = re.fullmatch(r"\s*([\w.-]+)\s*(\[[^]]*])?\s*", requirement).groups()
        module = re.sub(r"[-.]+", "_", dist).lower()
        root = _checkout(module)
        if root is None:
            wanted.append(f"{dist}{extras or ''}=={metadata.version(dist)}")
            continue
        tags.append(_tree_hash(root, module))
        wheels.append(_wheel(root, tags[-1], cache))
        wanted.append(f"{code}/{wheels[-1].name}{extras or ''}")
    cache.mkdir(parents=True, exist_ok=True)
    constraints, loose = cache / "constraints.txt", cache / "constraints-pip.txt"
    constraints.write_text(_constraints(True), encoding="utf-8")
    loose.write_text(_constraints(False), encoding="utf-8")

    pip = (f"{prefix}/bin/python", "-m", "pip", "install", "--progress-bar", "off")
    # ponytail: one version the index lacks drops every pin of what conda installed here, not
    # just that one. Drop them one by one if a host ever ends up too far from this environment.
    pinned = (
        'pip() { "$0" -m pip install --progress-bar off --constraint "$1" "${@:2}"; }; d=$1; shift; '
        'pip "$d/constraints.txt" "$@" || '
        '{ echo "not every version here can be had there: pinning what pip installed only"; '
        'pip "$d/constraints-pip.txt" "$@"; }'
    )
    refresh = [
        RecipeStep(
            f"send the code of {name}",
            ("sh", "-c", 'rm -rf "$0" && mkdir -p "$0" && tar -x -C "$0"', code),
            600.0,
            send=(str(constraints), str(loose), *map(str, wheels)),
        ),
        RecipeStep(
            f"install {name} and what it needs",
            ("bash", "-c", pinned, f"{prefix}/bin/python", code, *wanted),
            3600.0,
        ),
    ]
    if wheels:  # the version of a checkout does not change with its code: pip would keep the old one
        refresh.append(
            RecipeStep(
                f"install the code of {name} as it is here",
                (*pip, "--no-deps", "--force-reinstall", *(f"{code}/{wheel.name}" for wheel in wheels)),
                600.0,
            )
        )
    conda = (f"python={platform.python_version()}", "pip")
    values = {"prefix": prefix, "tools_home": str(tools_home).rstrip("/")}
    short = lambda *parts: hashlib.sha256(repr(parts).encode()).hexdigest()[:16]  # noqa: E731
    return Recipe(
        name=name,
        prefix=prefix,
        candidates=(f"{prefix}/bin/python",),
        commands=(),
        check=("{exe}", "-c", "import " + ", ".join(imports)),
        steps=(*_env_steps(name, conda, (), ("conda-forge",), values, manager), *refresh),
        conda=conda,
        stamp=short(conda, wanted, tags, constraints.read_text(encoding="utf-8")),
        refresh=tuple(refresh),
        base=short(conda),
    )

def ensure(recipe: Recipe, host: Host = LOCAL, busy: Callable[[], str] | None = None) -> Install:
    """Make sure the tool is there: start installing it, or join the install already running.

    Returns at once: wait on ``future`` and read ``lines`` for the log. Nothing is installed
    where the tool is already usable. An install of ours made from another recipe is made
    again, or brought up to this one where it stands when the recipe says how (that is the
    update); an environment of the user is verified and left as it is.

    ``busy`` is asked only before an install of ours is changed: what it returns is why that
    cannot be done now (something is running from it), and the install fails saying so.
    """
    key = f"{recipe.name}@{host.address or 'local'}"
    with _LOCK:
        current = _INSTALLS.get(key)
        if current is not None and not current.future.done():
            return current
        install = _INSTALLS[key] = Install(key)
    # ponytail: one daemon thread per install; a pool if installs pile up.
    # Daemon on purpose: quitting the app must not wait 30 min for a conda solve.
    threading.Thread(target=_run_install, args=(install, recipe, host, busy), name=f"ms_env-{key}", daemon=True).start()
    return install

def locate(recipe: Recipe, host: Host = LOCAL, timeout: float = 60.0) -> tuple[str, str]:
    """(READY, executable) or (MISSING | UNREACHABLE, ""). Blocks on a subprocess: keep it off UI threads."""
    try:
        result = subprocess.run(
            host.argv(_locate_script(recipe)),
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return (UNREACHABLE if host.address else MISSING), ""
    found = result.stdout.strip().splitlines()
    if result.returncode == 0 and found:
        return READY, found[-1]
    return (UNREACHABLE if host.address and result.returncode == _SSH_CONNECT_FAILED else MISSING), ""

def recipe_steps(recipe: Recipe, host: Host = LOCAL, refresh: bool = False) -> list[Step]:
    """The whole install: clear what ms_env left at the prefix, run the recipe, stamp it.

    The first step fails on a prefix that holds something ms_env did not install. With
    `refresh` the prefix stays and only `recipe.refresh` runs before the stamp.
    """
    label, prefix = (f"{host.address}: " if host.address else ""), _prefix(recipe)
    clear = (
        f'p={prefix}\n'
        f'if [ -n "$(ls -A "$p" 2>/dev/null)" ] && [ ! -f "$p/{STAMP}" ] && [ ! -f "$p{_INSTALLING}" ]; then\n'
        '  echo "$p holds an environment ms_env did not install: it is checked, never changed." >&2; exit 1\n'
        "fi\n"
        f'mkdir -p "$(dirname "$p")" && : > "$p{_INSTALLING}" && rm -rf -- "$p"'
    )
    stamp = (
        f'p={prefix}\nmkdir -p "$p" && printf %s {shlex.quote(_stamp_text(recipe))} > "$p/{STAMP}" '
        f'&& rm -f "$p{_INSTALLING}"'
    )
    steps = (
        *(recipe.refresh if refresh else (RecipeStep(f"prepare {recipe.prefix}", ("sh", "-c", clear), 300.0), *recipe.steps)),
        RecipeStep(f"record the {recipe.name} install", ("sh", "-c", stamp), 60.0),
    )
    return [_step(step, label, host) for step in steps]

def remove(recipe: Recipe, host: Host = LOCAL) -> bool:
    """Delete the prefix when ms_env installed it. False: nothing of ours there, nothing deleted."""
    key = f"{recipe.name}@{host.address or 'local'}"
    with _LOCK:
        current = _INSTALLS.get(key)
        if current is not None and not current.future.done():
            raise RuntimeError(f"{recipe.name} is being installed on {host.address or 'this machine'}.")
    script = (
        f'p={_prefix(recipe)}\n[ -f "$p/{STAMP}" ] || [ -f "$p{_INSTALLING}" ] || {{ echo no; exit 0; }}\n'
        f'rm -rf -- "$p" "$p{_INSTALLING}"'
    )
    return host.run(script, timeout=600).decode().strip() != "no"

def status(recipe: Recipe, host: Host = LOCAL) -> str:
    """READY, MISSING or UNREACHABLE. Blocks like `locate`, which also says where the tool is."""
    return locate(recipe, host)[0]

def verify(recipe: Recipe, host: Host = LOCAL) -> list[str]:
    """What the environment at the recipe's prefix lacks of what the recipe asks for; [] when it meets it.

    The way to take an environment the user already has instead of installing one: every
    pinned conda package (looked up in `conda-meta`), every pinned pip package (asked to the
    python of the prefix) and the recipe's check on the executables of that prefix. Reads
    only. ponytail: a pip URL or path is not looked up, and `>=` counts as "is there".
    """
    lines = [f'p={_prefix(recipe)}', '[ -d "$p" ] || { echo "there is no directory $p"; exit 0; }',
             'has() { ls "$p"/conda-meta/$1 >/dev/null 2>&1; }']
    for name, version in map(_pin, recipe.conda):
        # conda's `=1.2` takes 1.2 and 1.2.x; a file there is name-version-build.json
        globs = [f"{name}-{version}-*.json", f"{name}-{version}.*-*.json"] if version else [f"{name}-[0-9]*-*.json"]
        wanted = shlex.quote(f"conda: {name}{'=' + version if version else ''} is asked for and is not there")
        lines.append(" || ".join([*(f"has {shlex.quote(glob)}" for glob in globs), f"echo {wanted}"]))
    pins = ["==".join(filter(None, _pin(spec))) for spec in recipe.pip if "/" not in spec]
    if pins:
        lines.append(
            f'"$p/bin/python" -c {shlex.quote(_PIP_PINS)} {shlex.join(pins)} 2>/dev/null'
            ' || echo "pip: no working python at $p/bin/python"'
        )
    lacks = host.run("\n".join(lines), timeout=120).decode().splitlines()
    inside = tuple(path for path in recipe.candidates if path.startswith(recipe.prefix + "/"))
    if recipe.check and status(dataclasses.replace(recipe, candidates=inside, commands=()), host) != READY:
        lacks.append(f"check: `{shlex.join(recipe.check)}` passes for none of {', '.join(inside) or 'its executables'}")
    return lacks

def find_environment_manager(*extra_variables: str) -> Path | None:
    """The program that creates environments here: micromamba, unless a variable names another.

    ``extra_variables`` are app-specific overrides, read before ``MOLSUITE_ENV_MANAGER``; any
    of them may name a conda or a mamba. None: the recipe downloads its own micromamba.
    """
    named = [str(os.environ.get(variable) or "").strip() for variable in (*extra_variables, "MOLSUITE_ENV_MANAGER")]
    for candidate in (*filter(None, named), shutil.which("micromamba")):
        path = Path(candidate or "").expanduser()
        if path.is_file() and os.access(path, os.X_OK):
            return path.resolve()
    return None

def run_step(step: Step, on_line: Callable[[str], None] | None = None) -> None:
    """Run one step, streaming its output. Raises StepFailed on failure."""
    emit = on_line or (lambda _line: None)
    if step.call is not None:
        message = step.call()
        if message:
            emit(message)
        return
    if not step.argv:
        return
    emit("$ " + " ".join(step.argv))
    try:
        process = subprocess.Popen(
            step.argv,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
        )
    except FileNotFoundError as exc:
        raise StepFailed(f"{step.name}: command not found ({exc.filename}).") from exc
    deadline = time.monotonic() + step.timeout
    assert process.stdout is not None
    for line in process.stdout:
        emit(line.rstrip())
        if time.monotonic() > deadline:
            process.kill()
            raise StepFailed(f"{step.name}: timed out after {step.timeout:.0f}s.")
    try:
        code = process.wait(timeout=max(1.0, deadline - time.monotonic()))
    except subprocess.TimeoutExpired:
        process.kill()
        raise StepFailed(f"{step.name}: timed out after {step.timeout:.0f}s.") from None
    if code != 0:
        raise StepFailed(f"{step.name}: exited with code {code}.")

def run_steps(
    steps: Iterable[Step],
    on_line: Callable[[str], None] | None = None,
    on_step: Callable[[int, Step], None] | None = None,
) -> None:
    for index, step in enumerate(steps):
        if on_step is not None:
            on_step(index, step)
        run_step(step, on_line)