Source code for ska_sdp_piper.app.executable_pipeline

import importlib.util
import logging
import os
import sys
from importlib.machinery import SourceFileLoader
from pathlib import Path

from ..piper.exceptions import PipelineNotFoundException
from ..piper.named_instance import NamedInstance
from .constants import MAIN_ENTRY_POINT, SHEBANG_HEADER

logger = logging.getLogger(__name__)




class ExecutablePipeline:
    """
    Creates an executable instance from the pipeline definition

    Attributes
    ----------
        _script_path: str
            Path to the pipeline definition
        installable_pipeline: module
            The pipeline definition loaded as a module
        executable_content: str
            The content of the executable file
    """

    def __init__(self, pipeline_name, script_path=None):
        """
        Initialise a executable pipeline object

        Parameters
        ---------
            script_path: str
                Path to the pipeline definition
        """
        self._pipeline_name = pipeline_name
        self._script_path = script_path or self.__executable_script_path()

        self.installable_pipeline = None
        self.executable_content = None



    def validate_pipeline(self):
        """
        Validates if the pipeline definition is syntactically correct python
        code.

        Raises
        ------
            Compilation error based on python interpretor.
            PipelineNotFoundException if the provided pipeline name
                doesn't exist
        """
        logger.info(f"Validating {self._script_path}")
        if not os.path.exists(self._script_path):
            raise FileNotFoundError(self._script_path)

        _module_name = "installable_pipeline"
        spec = importlib.util.spec_from_loader(
            _module_name, SourceFileLoader(_module_name, self._script_path)
        )
        self.installable_pipeline = importlib.util.module_from_spec(spec)
        sys.modules[_module_name] = self.installable_pipeline
        spec.loader.exec_module(self.installable_pipeline)

        valid_pipeline = NamedInstance.get_instance(self._pipeline_name)

        if valid_pipeline is None:
            raise PipelineNotFoundException(
                f"Pipeline {self._pipeline_name} not"
                f" found in {self._script_path}"
            )


    def __prepare_executable(self):
        """
        Prepares the content of the executable script file
        """
        file_content = ""
        logger.info("Preparing executable")
        with open(self._script_path, "r") as script_file:
            file_content = script_file.readlines()

        self.executable_content = (
            SHEBANG_HEADER.format(executable=sys.executable)
            + "".join(file_content)
            + MAIN_ENTRY_POINT.format(pipeline_name=self._pipeline_name)
        )



    def install(self):
        """
        Installs the executable script containing the executable script.
        The path is taken from the executable path obtained from sys.executable
        The script file is created with a execute privilage to user and group
        """
        self.__prepare_executable()

        logger.info("Installing executable")
        executable_path = self.__executable_script_path()

        with open(executable_path, "w") as outfile:
            outfile.write(self.executable_content)
        os.chmod(executable_path, 0o750)

        logger.info(f"Installed {executable_path}")




    def uninstall(self):
        """
        Removes the executable pipeline from the executable path.
        """
        executable_path = self.__executable_script_path()
        logger.info(f"Deleting {executable_path}")
        os.remove(executable_path)


    def __executable_script_path(self):
        """
        Returns
        ------
            the absolute path of the executable.
            The path is derived from sys.executable
        """
        executable_root = Path(sys.executable).parent.absolute()
        return f"{executable_root}/{self._pipeline_name}"