Source code for fm_weck.image_mgr

# This file is part of fm-weck: executing fm-tools in containerized environments.
# https://gitlab.com/sosy-lab/software/fm-weck
#
# SPDX-FileCopyrightText: 2024 Dirk Beyer <https://www.sosy-lab.org>
#
# SPDX-License-Identifier: Apache-2.0

import logging
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING

try:
    from fm_tools.fmtoolversion import FmImageConfig
except ImportError:
    if not TYPE_CHECKING:

        class FmImageConfig:
            def __init__(self, full_images, base_images, required_packages):
                raise ImportError("fm_tools is not imported.")


from fm_weck.exceptions import NoImageError

if TYPE_CHECKING:
    from fm_weck.engine import Engine

CONTAINERFILE = Path(__file__).parent / "resources" / "Containerfile"
logger = logging.getLogger(__name__)

logger = logging.getLogger(__name__)




class ImageMgr(object):
    """
    The image manager singleton is responsible for preparing the images for the container.
    """

    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(ImageMgr, cls).__new__(cls)
        return cls._instance



    def prepare_image(self, engine: "Engine", image: FmImageConfig) -> str:
        if image.full_images:
            return image.full_images[0]

        if image.base_images and not image.required_packages:
            return image.base_images[0]

        if not image.base_images:
            raise NoImageError("No base image specified")

        logger.info(
            "Building image from from base image %s with packages %s", image.base_images[0], image.required_packages
        )
        image_cmd = engine.image_from(CONTAINERFILE)
        image_cmd.base_image(image.base_images[0])
        image_cmd.packages(image.required_packages)
        from yaspin import yaspin

        with yaspin(text="Building image", color="cyan") as spinner:
            tag = image_cmd.build()
            spinner.ok("✅ ")

        return tag




    def prepare_image_for_zenodo(self, engine: "Engine", image_name: str) -> tuple[str, str, Path]:
        """
        Prepare an image for Zenodo upload, by retreiving the image tarball.

        :return: A tuple containing the normalized image name and tarball path.

        Raises:
            ValueError: If the image is invalid or does not exist.
        """
        if ":" not in image_name:
            raise ValueError(f"Image specification should contain a tag: '{image_name}:tag'.")
        elif image_name.split(":")[1] == "latest":
            raise ValueError("Image specification should not use the 'latest' tag.")

        build_cmd = engine.image_from(CONTAINERFILE)

        if build_cmd.inspect_image(image_name) != 0:
            self.print_dockerfile_warning()
            raise ValueError(f"Image '{image_name}' is invalid or does not exist. Check your chosen engine.")

        with tempfile.NamedTemporaryFile(delete=False) as image_tar:
            image_tar_path = Path(image_tar.name)
            build_cmd.save_image(image_name, image_tar_path)
            return image_name.split(":")[0], image_name.split(":")[1], image_tar_path




    def check_if_doi_exists_locally(self, engine: "Engine", doi: str) -> str:
        """
        Check if an image tagged with the given DOI already exists locally.

        :return: The DOI if it does not exist locally, otherwise empty string.
        """
        build_cmd = engine.image_from(CONTAINERFILE)
        tag = doi.split(".")[-1]

        if build_cmd.tag_exists(tag):
            return ""
        else:
            return doi




    def load_image(self, engine: "Engine", image_tar: Path, tag: str | None = None) -> None:
        image_cmd = engine.image_from(CONTAINERFILE)
        image_cmd.load_image(image_tar)
        if tag:
            image_name = image_tar.stem.split(".")[0]
            image_cmd.tag_image(image_name, tag)




    def tag_image(self, engine: "Engine", image: str, tag: str) -> None:
        image_cmd = engine.image_from(CONTAINERFILE)
        image_cmd.tag_image(image, tag)




    @staticmethod
    def print_dockerfile_warning():
        print(
            "\033[31m" + "\nWarning:" + "\033[0m\n"
            "Building images from Dockerfiles is not supported for this process. "
            "Please use pre-built images instead.\n"
            "To build your own image, you can use Docker or Podman, ensuring that "
            "the image is fully functional and tested before permanently "
            "storing it on Zenodo.\n"
            "This approach helps maintain Zenodo's quality and avoids cluttering it "
            "with unfinished or untested images.\n"
            "We encourage verifying and finalizing images before uploading them to Zenodo.\n"
        )