Docker images¶

def apt_install(packages: tuple[AptPackage, ...]) -> str:
    """An ``apt-get install`` of ``packages``, one per line."""
    names: str = " \\\n".join(f"    {package.name}" for package in packages)

    return f"apt-get install -y --no-install-recommends \\\n{names}"

def ensure_image(image: DockerImage, docker: DockerCli, *, pull: bool = True) -> None:
    """Make Docker hold ``image``: reuse a matching local tag, else pull a published one, else build its recipe; checked once per tag per process.

    A tag without a recipe is pulled when Docker lacks it and never built, so a failed pull raises `DockerError`.

    Args:
        image: The image to make available.
        docker: The client to run Docker through.
        pull: Try the registry before building a published recipe.
    """

    def pulled_from_recipe(digest: str) -> bool:
        try:
            docker.pull(image.tag, platform=image.platform)
        except DockerError as error:
            log.bind(tag=image.tag, error=str(error)).warning("Could not pull the image; building it instead.")
            return False

        if docker.image_label(image.tag, RECIPE_LABEL) == digest:
            return True

        log.bind(tag=image.tag).warning("The pulled image was built from another recipe; building it instead.")
        return False

    with _BUILD_LOCK:
        if image.tag in _READY:
            return

        if image.dockerfile is None:
            if not docker.has_image(image.tag):
                docker.pull(image.tag, platform=image.platform)

            _READY.add(image.tag)
            return

        digest: str = recipe_digest(image.dockerfile)
        if docker.image_label(image.tag, RECIPE_LABEL) == digest or (pull and image.published and pulled_from_recipe(digest)):
            _READY.add(image.tag)
            return

        log.bind(tag=image.tag).info("Building a Docker image; this takes a few minutes, once.")
        docker.build(
            image.tag,
            image.dockerfile,
            labels=[f"{RECIPE_LABEL}={digest}", *image.labels],
            platform=image.platform,
        )

        probed: str = docker.output(["run", "--rm", *image.platform_args, image.tag, *image.probe]) if image.probe else ""
        log.bind(tag=image.tag, probe=probed).success("Built the Docker image.")
        _READY.add(image.tag)

def install_run(sections: Sequence[Section]) -> str:
    """A Dockerfile ``RUN`` heredoc that runs ``sections`` as one strict, non-interactive shell script."""
    script: str = "\n\n".join(section.render() for section in sections)

    return f"RUN <<EOF\nset -eux\nexport DEBIAN_FRONTEND=noninteractive\n\n{script}\nEOF"

def dockerfile() -> str:
    """The image's Dockerfile, generated from `INSTALL`."""
    return f"""\
# syntax=docker/dockerfile:1
# ^ Enables RUN heredocs (BuildKit).

# Elmer FEM (headless) on Ubuntu 24.04, installed from CSC's official PPA.
FROM {BASE_IMAGE}

# The repository this recipe comes from, and the licenses of the Elmer it installs.
LABEL {SOURCE_LABEL}="{SOURCE}" \\
      {LICENSES_LABEL}="{LICENSES}"

# Installs headless Elmer FEM from CSC's official package archive.
# One RUN, so the cleanup at the end actually shrinks the image.
{install_run(INSTALL)}

# Open MPI refuses to launch as root by default, and containers run as root.
# These allow parallel runs, e.g.:  mpirun -np 4 ElmerSolver_mpi
ENV OMPI_ALLOW_RUN_AS_ROOT=1 \\
    OMPI_ALLOW_RUN_AS_ROOT_CONFIRM=1

# Working directory for the simulation case (case.sif + mesh directory).
# Typically bind-mounted:  docker run -v "$PWD":/case <image> ElmerSolver case.sif
WORKDIR /case
"""

def add_arguments(parser: argparse.ArgumentParser) -> None:
    """The ``femforge doctor`` options."""
    _ = parser.add_argument("--json", action="store_true", help="Print the report as JSON instead of a table.")
    _ = parser.add_argument("--runs-dir", default="runs", help="The runs directory whose disk is checked (default: runs).")

def diagnose(runs_dir: Path, artifacts: Path | None = None, *, docker: DockerCli | None = None, recipe: ImageRecipe = ELMER_IMAGE) -> DoctorReport:
    """Run every check against this machine.

    Args:
        runs_dir: The runs directory whose disk is checked.
        artifacts: The artifact store whose disk is checked; None skips that check.
        docker: The Docker client to probe through; None makes one.
        recipe: The image recipe whose image is checked.
    """
    client: DockerCli = docker if docker is not None else DockerCli()
    docker_check, info = _check_docker(client)
    docker_usable: bool = info is not None
    image_check: Check = _check_elmer_image(client, recipe) if docker_usable else Check(name=CheckName.ELMER_IMAGE, status=CheckStatus.SKIP, detail="Needs Docker.")
    native_check, tools = _check_native_elmer(docker_usable)

    return DoctorReport(
        checks=[
            _check_python(),
            _check_gmsh(),
            _check_cad_kernel(),
            docker_check,
            image_check,
            native_check,
            _check_disk(CheckName.RUNS_DISK, runs_dir.resolve(), "Free space, for example with `femforge purge`."),
            _check_disk(CheckName.ARTIFACTS_DISK, artifacts.resolve() if artifacts is not None else None, "Free space, for example with `femforge purge`."),
            _check_docker_disk(info),
            _check_solve_path(tools is not None, docker_usable),
        ]
    )

def render(report: DoctorReport, console: Console) -> None:
    """Print the checks as a table, then the numbered fixes and the count of blockers and warnings."""
    table = Table("Check", "Status", "Detail")
    for check in report.checks:
        table.add_row(check.name.value, Text(check.status.value.upper(), style=check.status.style), Text(check.detail))

    console.print(table)

    fixes: list[Check] = [check for check in report.checks if check.fix is not None]
    for number, check in enumerate(fixes, start=1):
        console.print(Text(f"{number}. {check.name.value}: {check.fix}"))

    blockers: str = f"{report.blockers} blocker{'' if report.blockers == 1 else 's'}"
    warnings: str = f"{report.warnings} warning{'' if report.warnings == 1 else 's'}"
    console.print(f"{blockers}, {warnings}")

def run(args: argparse.Namespace, console: Console, artifacts: str | None) -> int:
    """Run the checks and report them; the exit code is 1 when any check failed.

    Args:
        args: The parsed ``femforge doctor`` options.
        console: Where the table goes.
        artifacts: The artifact store root from the environment, or None.
    """
    report: DoctorReport = diagnose(Path(cast("str", args.runs_dir)), Path(artifacts) if artifacts else None)
    if cast("bool", args.json):
        print(report.model_dump_json(indent=2))
    else:
        render(report, console)

    return 1 if report.blockers else 0