testing¶

class LocalPipelineRunner:
    """
    Simulates the execution of a pipeline in a local Python environment.

    Restriction: only linear pipelines are supported where the pipeline input variables are only used by one component,
    each component uses only the outputs of the previous components, and the pipeline output only consists of variables from the last component.

    If the caller specifies no `path`, the working directory is temporary and is removed unless an error occurs.
    If the caller specifies a working directory with a `path` argument, the working directory is kept.
    This behavior can be overridden using boolean parameter `cleanup`.

    Currently, the pipeline runner supports both the current `process_input(data: dict)` entrypoint signature and the legacy
    `run(data: str)` signature. If both entrypoints are present, `process_input()` takes precedence. Please note however that
    `run()` is deprecated, and support for it will be removed in future versions of the pipeline runner.

    Args:
        packageZip (path-like): Path to the pipeline configuration package.
        path (path-like): Path to the working directory. If unset, a temporary directory is created.
        cleanup (bool): If set, the working directory is kept when True, and deleted when False. \
            If unset, a temporary working directory is removed, and an explicit working directory is kept. \
            When an error occurs in a component, the working directory is kept regardless of this value.
    """

    def __init__(self, packageZip: os.PathLike, path: Optional[os.PathLike] = None, cleanup: Optional[bool] = None, loglevel = logging.INFO):
        """
        Creates a new component LocalPipelineRunner for the provided pipeline configuration package.

        Only works with a pipeline configuration package. Does not work with e.g. an edge configuration package.

        Args:
            packageZip (path-like): Path to the pipeline configuration package.
            path (path-like): Path to the working directory. If unset a temporary directory will be created.
            cleanup (bool): If set, the working directory will be kept when True, and deleted when False. \
                If unset, a temporary working directory will be removed, and an explicit working directory will be kept. \
                When an error occurs in a component, the working directory will be kept regardless of this value.
        """
        self.package_zip: Path = Path(packageZip)
        self.path = Path(path).resolve().absolute() if path is not None else None
        self.components = {}
        self.parameters = {}
        self.cleanup = cleanup
        self.log_level = loglevel
        self.workdir: Path
        self.docker = VesselBaseDocker()

        self.report_writer = PipelineRunnerReportWriter()
        report_writer_handler = ReportWriterHandler(self.report_writer)
        _logger.addHandler(report_writer_handler)

    def __enter__(self):
        self.report_writer.set_package_zip_path(self.package_zip)
        timestamp = re.sub(r"[-:]", "", datetime.now(timezone.utc).isoformat(sep="_", timespec="seconds"))
        if self.path is not None:
            self.workdir = Path(self.path)
            self.workdir.mkdir(parents=True, exist_ok=True)
            self.cleanup = self.cleanup if self.cleanup is not None else False
        else:
            self.workdir = Path(tempfile.mkdtemp(prefix=f"LocalPipelineRunner_{timestamp}_"))
            self.cleanup = self.cleanup if self.cleanup is not None else True

        unzip_components = False
        with zipfile.ZipFile(self.package_zip) as zf:
            if 'runtime_config.yml' in zf.namelist():
                self.workdir = self.workdir / self.package_zip.stem
                zf.extractall(path=self.workdir)
                unzip_components = True
            else:
                zf.extractall(path=self.workdir)
                self.workdir = self.workdir / zf.namelist()[0]

        try:
            with open(self.workdir / "pipeline_config.yml") as cf:
                config = yaml.load(cf, Loader=yaml.FullLoader)

            components = config["dataFlowPipeline"].get("components", [])
            for component in components:
                component["context"] = None
                component["env_dir"] = self.workdir / component['name']
                self.components[component["name"]] = component

            if unzip_components:
                for component in components:
                    component_zip = f"{component['name']}_{component['version']}.zip"
                    with zipfile.ZipFile(self.workdir / 'components' / component_zip) as zf:
                        zf.extractall(path=self.workdir / component["name"])

            for parameter in config["dataFlowPipeline"].get("pipelineParameters", {}):
                self.parameters[parameter["name"]] = parameter

        except Exception:
            raise RuntimeError(INVALID_PIPELINE_PACKAGE_MESSAGE)

        return self

    def __exit__(self, exception_type, value, traceback):

        self.update_telemetry_data()
        self.report_writer.write_report()

        if self.cleanup:
            _logger.info("Removing local pipeline runner environment...")
            shutil.rmtree(self.workdir.parent)
        else:
            _logger.info(f"Leaving local pipeline runner environment in its final state at '{self.workdir}'")

    def collect_telemetry_data(self) -> dict:
        """
        Collects telemetry data about the platform, environment, industrial AI packages, and pipeline.

        Returns:
            dict: A dictionary containing the telemetry data.
        """
        telemetry_data = {}

        telemetry_data["platform"] = {}
        telemetry_data["platform"]["os"] = platform.system()
        telemetry_data["platform"]["release"] = platform.release()
        telemetry_data["platform"]["python_version"] = platform.python_version()

        _logger.info(f"locals: {locals()}")
        telemetry_data["environment"] = {}
        env_info = "Jupyter" if any(k for k in locals() if k in ["__IPYTHON__", "get_ipython"]) else MSG_NOT_FOUND
        env_info = "GitLab CI/CD" if "GITLAB_CI" in os.environ else env_info
        env_info = "Azure DevOps Pipelines" if "TF_BUILD" in os.environ else env_info
        env_info = "GitHub Actions" if "GITHUB_ACTIONS" in os.environ else env_info
        telemetry_data["environment"]["info"] = env_info

        telemetry_data["industrial_ai"] = {}
        try:
            telemetry_data["industrial_ai"]["simaticai"] = importlib_metadata.version("simaticai")
        except importlib_metadata.PackageNotFoundError:
            telemetry_data["industrial_ai"]["simaticai"] = MSG_NOT_FOUND

        telemetry_data["pipeline"] = {}
        telemetry_data["pipeline"]["python_versions"] = list(set(self.components[component]['runtime']['version'] for component in self.components
                                                                 if self.components[component]['runtime']['type'] == 'python'))
        telemetry_data["pipeline"]["file_extensions"] = []
        for component_dir in [Path(self.workdir) / c for c in Path(self.workdir).rglob("*") if c.name in self.components.keys()]:
            excluded_dirs = set([component_dir / '.venv', component_dir / '__pyache__'])
            suffixes = list(set(f.suffix for f in component_dir.rglob("*")
                                if not (any(excluded_dirs.intersection(f.parents)) or f.suffix in ["", ".zip", ".yml", ".yaml", ".html"])))
            for suffix in suffixes:
                if suffix not in telemetry_data["pipeline"]["file_extensions"]:
                    telemetry_data["pipeline"]["file_extensions"].append(suffix)

        return telemetry_data

    def update_telemetry_data(self):
        """
        Update the telemetry data and the package.

        This method updates the telemetry data by loading the existing data from a YAML file,
        or collecting new telemetry data if the file doesn't exist. It then updates the
        "last_test_run" field of the telemetry data with the current timestamp. The updated
        telemetry data is then written back to the YAML file.

        If the package contains a different version of the telemetry data file, a new package
        is created with the updated telemetry data. Otherwise, the existing package is
        overwritten with the new package containing the updated telemetry data.
        """
        _logger.info("Updating telemetry data and the package")
        telemetry_path = self.workdir / "telemetry_data.yml"
        if telemetry_path.is_file():
            telemetry_data = yaml.safe_load(telemetry_path.read_text())
        else:
            telemetry_data = self.collect_telemetry_data()

        telemetry_data["pipeline"]["last_test_run"] = datetime.now().isoformat()
        telemetry_path.write_text(yaml.dump(telemetry_data))

        config_package = False
        with zipfile.ZipFile(self.package_zip, 'r') as zip_read:
            for file in zip_read.namelist():
                if TELEMETRY_YAML in file and file != TELEMETRY_YAML:
                    config_package = True
                    break

        new_zip_path = Path(self.package_zip).parent / f"{Path(self.package_zip).stem}_tested.zip"

        with zipfile.ZipFile(self.package_zip, 'r') as zip_read:
            with zipfile.ZipFile(new_zip_path, 'w') as zip_write:
                for file in zip_read.namelist():
                    if TELEMETRY_YAML not in file:
                        filepath = zip_read.extract(file, path=self.workdir)
                        zip_write.write(filepath, arcname=file)
                    else:
                        zip_write.write(telemetry_path, arcname=file)

        if config_package:
            shutil.copy(new_zip_path, self.package_zip)
        else:
            new_sha_path = Path(self.package_zip).parent / f"{Path(self.package_zip).stem}_tested.sha256"
            new_sha_path.write_text(calc_sha(new_zip_path))

    def _install_requirements(self, component, package_dir, no_index: bool = True):
        _logger.info("Installing requirements...")
        pip_report_file = component["env_dir"] / "pip_report.json"
        package_dir_path = _relative_to(package_dir, component["env_dir"])
        pip_report_file_path = _relative_to(pip_report_file, component["env_dir"])
        cmd = [
            str(component['python_path']),
            "-m", "pip", "install",
            "--no-warn-script-location",
            "-f", f"{package_dir_path}",
            "-r", REQUIREMENTS_TXT,
        ]

        if no_index:
            cmd += [ "--no-index" ]

        use_no_deps = component.get("runtime", {}).get("installTransitiveDependencies", True) is False
        if use_no_deps:
            cmd += [ "--no-deps" ]

        result = subprocess.run(cmd, cwd=component["env_dir"], text=True, stderr=subprocess.PIPE)

        if 0 == result.returncode and use_no_deps:
            _logger.warning(f"Component `{component['name']}` is requested to be installed without transitive dependencies. "
                            "This may lead to runtime errors if the dependencies are not handled manually.")

        # generate pip report with a dry run to get the list of installed packages (in case of CPU Python packages)
        if 0 == result.returncode and component.get("hwType", None) == "CPU":
            cmd += ["--dry-run", "--ignore-installed", "--report", f"{pip_report_file_path}"]
            subprocess.run(cmd, cwd=component["env_dir"], text=True, stderr=subprocess.PIPE)
            self.report_writer.add_installed_packages(component["name"], pip_report_file)
        return result

    def _install_from_packages_zip(self, component, package_dir):
        result = self._install_requirements(component, package_dir, True)
        return 0 == result.returncode

    def _install_from_pypi_org(self, component, package_dir):
        return self._install_requirements(component, package_dir, False)

    def _init_component_venv(self, component: dict):
        """
        Creates a virtual environment in which the given component can run.

        Args:
            component (str): name of the selected component.
        """
        _logger.info(f"Creating virtual environment for component '{component['name']}'...")
        context_dir = component["env_dir"] / ".venv"

        if self.docker.is_vessel:
            _logger.info("Creating virtual environment in docker...")
            component["context"], component["python_path"] = self.docker._create_venv(component["env_dir"], component["runtime"]["version"])
            component["python_path"] = Path(component["python_path"]).resolve()
            _logger.info(f"Component context (docker): {component['context']}")
        else:
            builder = venv.EnvBuilder(with_pip=True, symlinks=False)
            builder.create(str(context_dir))
            component["context"] = builder.ensure_directories(context_dir)
            component['python_path'] = Path(component["context"].env_exe).resolve()
        _logger.debug(f"Component python_path: {component['python_path']}")

        _logger.info("Upgrading pip...")
        cmd = [str(component['python_path']), "-m", "pip", "install", "pip", "--upgrade"]
        result = subprocess.run(cmd, cwd=component["env_dir"], text=True, stderr=subprocess.PIPE)
        if result.returncode != 0:
            self.cleanup = False
            _logger.warning(f"Error upgrading pip:\n{result.stderr}")

        cmd = [str(component['python_path']), "-m", "pip", "install", "--upgrade", "setuptools"]
        result = subprocess.run(cmd, cwd=component["env_dir"], text=True, stderr=subprocess.PIPE)
        if result.returncode != 0:
            self.cleanup = False
            _logger.warning(f"Error installing setuptools:\n{result.stderr}")

        try:
            result = self._install_logmodule(component["python_path"], component["env_dir"])
        except Exception as err:
            _logger.error(err)
            self.cleanup = False
            raise RuntimeError("The 'simaticai' Python package is either not installed or does not contain package 'log_module'.") from None
        if result.returncode != 0:
            self.cleanup = False
            raise RuntimeError(f"Error installing log_module:\n{result.stderr}")

        req_list = Path(component["env_dir"] / "requirements.list")
        req_list.touch(exist_ok=True)
        if Path(component["env_dir"] / REQUIREMENTS_TXT).is_file():
            dependencies, extra_index, index_url = parse_requirements(component["env_dir"] / REQUIREMENTS_TXT)
            requirements = "#".join(dependencies.keys())
            req_list.write_text(requirements)
        else:
            _logger.info(f"'{REQUIREMENTS_TXT}' was not found. No additional dependencies were installed.")
            return

        package_dir = component["env_dir"] / PYTHON_PACKAGES
        package_zip = component["env_dir"] / PYTHON_PACKAGES_ZIP

        _logger.info(f"Extracting {PYTHON_PACKAGES_ZIP}")
        if package_zip.is_file():
            with zipfile.ZipFile(package_zip) as zf:
                zf.extractall(path=package_dir.absolute())
        else:
            _logger.info(f"There is no {PYTHON_PACKAGES_ZIP} to extract.")
            package_dir.mkdir(parents=True, exist_ok=True)

        success = self._install_from_packages_zip(component, package_dir)
        if not success:
            msg = f"Warning! Could not install dependencies from {PYTHON_PACKAGES_ZIP}. "
            msg += "Trying to install them from pypi.org. The resulting Python environment "
            msg += "may be significantly different than the targeted Python environment on the Edge Device!"
            _logger.warning(msg)

            if self.docker.is_vessel and component["runtime"]["type"] == "python":
                raise RuntimeError("The component is running in a docker container. The installation of dependencies from pypi.org is not supported.")
            else:
                second_install_result = self._install_from_pypi_org(component, package_dir)
                if 0 != second_install_result.returncode:
                    self.cleanup = False
                    raise RuntimeError(f"Error installing requirements:\n{second_install_result.stderr}")

    @staticmethod
    def _install_logmodule(python_path, env_dir):
        _logger.info("Installing LogModule...")
        try:
            package_paths = importlib_metadata.files("simaticai")
            assert package_paths is not None
            logger_wheel = [p for p in package_paths if 'log_module' in str(p)][0].locate()
        except Exception:
            from importlib.metadata import Distribution
            direct_url = Distribution.from_name("simaticai").read_text("direct_url.json")
            assert direct_url is not None
            direct_url = json.loads(direct_url)['url']
            direct_url = direct_url.replace('file://','')
            direct_url = Path(direct_url) / 'simaticai' / 'data'
            paths = list(direct_url.rglob('*.whl'))
            logger_wheel = [p for p in paths if 'log_module' in str(p)][0].resolve()

        cmd = [
            str(python_path), "-m", "pip",
            "install",
            "--no-warn-script-location", logger_wheel, "joblib"
        ]
        return subprocess.run(cmd, cwd=env_dir, text=True, stderr=subprocess.PIPE)

    def run_component(self, name: str, data: Optional[Union[dict, list, DataStream]]) -> Optional[Union[dict, list]]:
        """
        Runs the component in its virtual environment with the given input.
        This environment is created according to `requirements.txt` in the package.
        Additionally 'joblib' and the mock `log_module` is automatically installed in this virtual environment.
        The input data can be a single input record in a dictionary or a batch of input records in a list of dictionaries,
        or a DataStream object which will produce the appropriate input data.
        The supplied input data is saved as `inputs.joblib` in the component runtime directory, and the output is saved as `output.joblib`.

        Args:
            name (str): The name of the component to be executed.
            data (dict or list): One or more input records for the component.

        Returns:
            dict / list: A list of dictionaries for outputs if there were no errors and field `ready` is true.
            If the input was a single dict, then a single dict (the first item of the list) or None if there is no output.

        """
        assert name in self.components, f"Invalid component name: {name}"
        component = self.components[name]

        assert component["runtime"]["type"] in ["python", "gpuruntime"], f"Can not run component '{name}': Runtime type is nor 'python' or 'gpuruntime'"

        input_payload_path: Path  = component["env_dir"] / "input.joblib"
        output_payload_path: Path = component["env_dir"] / "output.joblib"

        batch_input: bool  = component["batch"]["inputBatch"]  == "Yes" if component.get("batch") is not None else False
        batch_output: bool = component["batch"]["outputBatch"] == "Yes" if component.get("batch") is not None else False

        # Validate and serialize input payload
        assert data is not None, f"Can not run component '{name}' without input."
        result_is_list = True
        if isinstance(data, list):
            input_payload = data
        elif isinstance(data, DataStream):
            input_payload = [item for item in data]
        else:
            result_is_list = False
            input_payload = [data]
        validate_payload(input_payload, component["inputType"], batch_input)
        joblib.dump(input_payload, input_payload_path)
        self.report_writer.set_input_payload_length(name, len(input_payload))
        _logger.info(f"Input payload saved as '{input_payload_path}'")

        # Assemble command for runnig component
        if component['runtime']['type'] == 'python':
            # Version check for Python
            e_major, e_minor, _, _, _ = sys.version_info
            c_major, c_minor, *_ = tuple(str(component["runtime"]["version"]).split('.'))

            if not self.docker.is_vessel and (f"{e_major}.{e_minor}" != f"{c_major}.{c_minor}"):
                msg = f"The local python version ({e_major}.{e_minor}) and the python version defined for the component ({c_major}.{c_minor}) are different."
                msg += " Testing will be done using dependencies that corresponds to the python version of your development environment."
                msg += " Pipeline behavior on AI Inference Server might be different."
                _logger.warning(msg)

            if component["context"] is None:
                self._init_component_venv(component)
                shutil.copy(_runner_path / 'run_component.py', component["env_dir"])

            req_list_path = _relative_to(component["env_dir"] / "requirements.list", component["env_dir"])
            json_params = json.dumps({param["name"]: param["defaultValue"] for param in self.parameters.values()})
            args = [
                "-m", 'run_component',
                "-m", Path(component["entrypoint"]).stem,
                "-p", f"{json_params}",
                "-r", f"{req_list_path}",
            ]

        else:
            # gpuruntime step requires Python environment with onnxruntime installed
            if component["context"] is None:
                shutil.copy(_runner_path / 'gpuruntime_requirements.txt', component["env_dir"] / REQUIREMENTS_TXT)
                self._init_component_venv(component)
                shutil.copy(_runner_path / 'run_gpuruntime_component.py', component["env_dir"])
                shutil.copy(_runner_path.parent / 'model_config_pb2.py', component["env_dir"])

            args = [
                "-m", 'run_gpuruntime_component',
                "-m", component["entrypoint"],
                "-c", "config.pbtxt",
            ]

        args += [
            "-i", input_payload_path.name,
            "-o", output_payload_path.name,
            "-ll", logging.getLevelName(self.log_level)
        ]

        # Run the component in the created Python environment
        _logger.info(f"Running component '{name}'...")
        _logger.info(f"{component['python_path'] =} to {component['env_dir'] =}")

        cmd = [str(component['python_path'])] + args
        _logger.info(f"Running command: {subprocess.list2cmdline(cmd)}")
        p = subprocess.Popen(cmd, cwd=component["env_dir"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
        if p.stdout is not None:
            for line in p.stdout:
                _logger.info(line.strip())
        returncode = p.wait()
        if returncode not in [RETURN_CODE_OK, RETURN_CODE_DEPRECATED]:
            self.cleanup = False
            raise RuntimeError(f"""\
There was an error while running component '{name}'.
You can check the test results in directory '{component['env_dir']}'
""")
        else:
            response_wrapped = True if returncode == RETURN_CODE_DEPRECATED else False

        # Deserialize and validate output payload
        _logger.info(f"Loading output payload from '{output_payload_path}'")
        output_payload = joblib.load(output_payload_path)
        output_payload = output_payload if isinstance(output_payload, list) else [output_payload]
        self.report_writer.set_output_payload_length(name, len(output_payload))
        if response_wrapped:
            output_payload = [json.loads(item['output']) for item in output_payload if item['ready']]
        else:
            output_payload = [output for output in output_payload if output is not None]
        validate_payload(output_payload, component["outputType"], batch_output)

        if result_is_list:
            return output_payload
        if 0 < len(output_payload):
            return output_payload[0]
        return None

    def update_parameters(self, parameters: dict):
        """
        Validates and updates pipeline parameters.
        The elements of the dictionary must match the parameters specified in the pipeline configuration package.
        If any of the names or types does not match, all parameters will remain untouched.

        Args:

            parameters (dict): names and values of parameters to update

        Raises:

            AssertionError:
                When:
                - either `name` is not in the configured parameters,
                - or `defaultValue` type is different from the configured one
        """
        for key,value in parameters.items():
            if key not in self.parameters.keys():
                if key.upper().startswith("__AI_IS"):
                    self.parameters[key] = {'defaultValue': False, 'name': key, 'type': 'Boolean'}
                else:
                    raise AssertionError(f"Pipeline has no parameters with the name '{key}' and type '{type_map.get(type(value).__name__)}'")
            elif self.parameters[key]["type"] != type_map.get(type(value).__name__):
                raise AssertionError(f"Pipeline has no parameters with the name '{key}' and type '{type_map.get(type(value).__name__)}'")

        for key, value in parameters.items():
            self.parameters[key]["defaultValue"] = value

    def run_pipeline(self, payload: Optional[Union[dict, list, DataStream]] = {}) -> Optional[Union[dict, list]]:
        """
        Runs all the components sequentially, assuming the output of a component is only consumed by the next.
        The input data can be a single input record in a dictionary or a batch of input records in a list of dictionaries.
        For each component the supplied input data is saved as `input.joblib` in the component runtime directory,
        and the output is saved as `output.joblib`.

        Args:
            payload (dict or list): One or more input records for the pipeline.

        Returns:
            The output of the last component.
        """

        for name in self.components.keys():
            payload = self.run_component(name, payload)

        return payload

class ComponentRunner():
    """
    Class to run a Pipeline Component in a virtual environment.
    Supported Component Types:
    - PythonComponent
    - GPURuntimeComponent

    Args:
    component: PythonComponent or GPURuntimeComponent
    workdir: Path to the directory where the component should be run. If None, the current working directory is used.
    cleanup: If True, the workdir is deleted after the context manager exits.

    """
    def __init__(self, component, workdir=None, cleanup=False):
        self.component = component
        self.parameters = { '__AI_IS_IMAGE_SET_VISUALIZATION': False }
        self.cleanup = cleanup

        self._logger = self._set_logger()

        self._create_workdir(component, workdir)
        self._copy_resources()
        self._create_venv()
        self._install_requirements()

    def __enter__(self):
        return self

    def __exit__(self, exception_type, exception_value, traceback):
        if exception_type is not None:
            self._logger.error(f"Exception occurred: {exception_type.__name__}: {exception_value}")
            return False
        if self.cleanup:
            shutil.rmtree(self.workdir)

    def _set_logger(self):
        """
        Set the logger for the ComponentRunner.
        """
        logger = logging.getLogger(__name__)
        log_level = os.environ.get("loglevel", "INFO").upper()
        logger.setLevel(log_level)
        handler = logging.StreamHandler()
        handler.setLevel(log_level)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        return logger

    def _create_workdir(self, component, workdir):
        """
        Create a workdir for the component.
        """
        if workdir is None:
            self.workdir = Path.cwd() / component.name
        else:
            self.workdir = Path(workdir).resolve().absolute() / component.name

        self.workdir.mkdir(parents=True, exist_ok=True)
        self._logger.info(f"Created workdir: {self.workdir}")

    def _copy_resources(self):
        """
        Copy the required resources of the component to the workdir.
        In case of a PythonComponent, also creates the dependencies into requirements.txt.
        In case of a GPURuntimeComponent, also creates a requirements.txt with the required onnx and onnxruntime dependencies.
        Also adds the runner script to the workdir.
        In case of the source folder same as the workdir, the resources are not copied,
        so any changes in source will affect the workdir.
        """
        if isinstance(self.component, PythonComponent):
            same_resources = []
            for from_path in self.component.resources.keys():
                try:
                    to_path = self.workdir / self.component.resources[from_path]
                    to_path.mkdir(parents=True, exist_ok=True)
                    shutil.copy(from_path, to_path)
                except shutil.SameFileError:
                    same_resources.append(from_path.name)
            if same_resources:
                self._logger.info("Resources are already in workdir, which are not copied:")
                for resource in same_resources:
                    self._logger.debug(f"  - {resource}")

            # download manually added Python packages and requirements.txt
            self.component.python_dependencies.save(self.workdir)
            shutil.copy(PYTHON_RUNNER_PATH, self.workdir)
        else:
            model_dir = Path(self.workdir / "1")
            model_dir.mkdir(parents=True, exist_ok=True)
            shutil.copy(self.component.model_path, model_dir / "model.onnx")
            Path(self.workdir / "config.pbtxt").write_text(f"{self.component.auto_config}")

            shutil.copy(GPU_RUNNER_PATH, self.workdir / "run_component.py")
            shutil.copy(GPU_REQUIREMENTS_PATH, self.workdir / "requirements.txt")
            shutil.copy(GPU_CONFIG_PATH, self.workdir)

        self._logger.info("Resources are copied into the workdir.")

    def _create_venv(self):
        """
        Create a Python virtual environment in the workdir.
        The created virtual environment is stored in the context_dir and
        has the same version as the current Python interpreter.
        """
        context_dir = self.workdir / ".venv"
        builder = venv.EnvBuilder(with_pip=True, symlinks=False)
        builder.create(context_dir)

        self.context = builder.ensure_directories(context_dir)
        self.python_path = Path(self.context.env_exe).resolve()

        self._logger.info(f"Python virtualenv created in folder '{self.context.env_dir}'")

    def _install_requirements(self):
        """
        Installs the required Python dependencies.
        """
        cmd = [str(self.python_path), '-m', 'pip', 'install', 'joblib', '-r', 'requirements.txt']
        result = subprocess.run(cmd, cwd=self.workdir, capture_output=True)
        if result.returncode != 0:
            raise RuntimeError(result.stderr.decode())

        result_lines = result.stdout.decode().split("\n")
        installed_message = " ".join([line for line in result_lines if "Successfully installed" in line])
        self._logger.info(installed_message)

    def set_parameters(self, parameter_name, parameter_value):
        """
        Set the parameters for the component.
        """
        self.parameters[parameter_name] = parameter_value

    def run(self, input_payload: Optional[Union[dict, list]]) -> Optional[Union[dict, list]]:
        """
        Run the component with the input payload.

        Parameters:
        input_payload: Input payload for the component.

        Returns:
        Output payload from the component.

        Side Effects:
        - The input payload is saved in the workdir.
        - The output payload is loaded from the workdir.
        """
        input_payload_path: Path  = self.workdir / "input.joblib"
        output_payload_path: Path = self.workdir / "output.joblib"

        batch_input: bool  = self.component.batch.inputBatch
        if isinstance(input_payload, DataStream):
            input_payload = [item for item in input_payload]
        elif isinstance(input_payload, dict):
            input_payload = [input_payload]

        input_variables = [{'name': name, 'type': c_input['type']} for name, c_input in self.component.inputs.items()]
        self._validate_payload(input_payload, input_variables, batch_input)

        joblib.dump(input_payload, input_payload_path)

        cmd = self._create_command(input_payload_path, output_payload_path)

        self._logger.info("Running command: " + " ".join(cmd))

        result = subprocess.run(cmd, cwd=self.workdir, capture_output=True)
        if result.returncode != 0:
            raise RuntimeError(result.stderr.decode())

        self._logger.debug(result.stderr.decode())
        output_payload = joblib.load(output_payload_path)

        batch_output: bool = self.component.batch.outputBatch
        output_payload = output_payload if isinstance(output_payload, list) else [output_payload]
        output_payload = [output for output in output_payload if output is not None]
        output_variables = [{'name': name, 'type': c_output['type']} for name, c_output in self.component.outputs.items()]
        if isinstance(self.component, PythonComponent):
            output_variables += [{'name': name, 'type': 'String'} for name in self.component.metrics]
        self._validate_payload(output_payload, output_variables, batch_output)

        return output_payload

    def _check_instance(self, element, element_name: str, instance):
        if not isinstance(element, instance):
            self._logger.error(f"{element_name} must be an instance of {instance.__name__}")
            raise ValueError(f"{element_name} must be an instance of {instance.__name__}")

    def _check_variable_types(self, variable: dict, value):
        if variable["type"] == "String":
            self._check_instance(value, "String value", str)

        if variable["type"] == "StringArray":
            self._check_instance(value, "StringArray value", list)
            for i in value:
                self._check_instance(i, "StringArray item", str)

        if variable["type"] == "Object":
            self._check_instance(value, "Object value", dict)
            values = list(value.values())
            if len(values) != 2:
                self._logger.error("Object value must have exactly 2 items")
                raise ValueError("Object value must have exactly 2 items")
            ok = isinstance(values[0], str) and isinstance(values[1], bytes) or isinstance(values[1], str) and isinstance(values[0], bytes)
            if not ok:
                self._logger.error("Object value must have exactly one 'str' and one 'bytes' field")
                raise ValueError("Object value must have exactly one 'str' and one 'bytes' field")

    def _validate_payload(self, payload: list, variables: list, batch: bool):
        self._check_instance(payload, "Payload", list)
        for payload_element in payload:
            if batch:
                self._check_instance(payload_element, "Batch payload element", list)
            else:
                payload_element = [payload_element]

            for item in payload_element:
                self._validate_payload_item(item, variables)

    def _validate_payload_item(self, item: dict, variables: list):
        self._check_instance(item, "Payload item", dict)
        for variable in variables:
            name = variable["name"]
            value = item.get(name, None)
            if value is None:
                self._logger.warning(f"WARNING! Variable '{name}' is missing from input, output or metric")
                continue

            self._check_variable_types(variable, value)

        payload_names = set(item.keys())
        variable_names = { variable["name"] for variable in variables }
        variable_names.add('timestamp')
        extra_variables = payload_names - variable_names
        if len(extra_variables):
            self._logger.warning(f"WARNING! These variables are not declared but are part of the payload: {extra_variables}")

    def _create_command(self, input_payload_path, output_payload_path):
        """
        Create the command to run the component.

        """
        cmd = [str(self.python_path), "-m", 'run_component']
        if isinstance(self.component, PythonComponent):
            cmd += [
                "-m", Path(self.component.entrypoint).stem,
                "-p", json.dumps(self.parameters)
            ]
        if isinstance(self.component, GPURuntimeComponent):
            model_path = Path(self.workdir / "1" / "model.onnx").absolute().resolve()
            config_path = Path(self.workdir / "config.pbtxt").absolute().resolve()
            cmd += [
                "-m", str(model_path),
                "-c", str(config_path),
            ]

        cmd += [
            "-i", input_payload_path.name,
            "-o", output_payload_path.name,
            "-ll", logging.getLevelName(self._logger.getEffectiveLevel())
        ]
        return cmd