packaging¶

Pipeline packaging.

This module contains classes and functionality for creating and validating pipeline configuration packages.

`constants`¶

Common constants used in 'packaging' module.

`python_dependencies`¶

Python dependencies

This class handles specifying and validating Python dependencies.

`PythonDependencies` ¶

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/python_dependencies.py

class PythonDependencies():

    def __init__(self, python_version='3.11', dir: Union[str, os.PathLike] = None):
        """
        This class handles Python dependencies

        Dependencies from remote repositories can be added via a requirements.txt file,
        or by calling the add_dependencies method. Dependencies can also be added as
        a single wheel or source distribution file, or as a collection in a zip archive
        via the add_python_packages method.

        The class can be converted to string, which will contain the dependencies
        in PEP508 format.
        """
        self.python_version = python_version
        self.dependencies = {}
        self.python_packages = Path(tempfile.mkdtemp(dir=dir, prefix="dependencies_"))
        self.temp = Path(tempfile.mkdtemp(dir=dir, prefix="temp_"))

        self.optimize_dependencies = True
        self.index_url = None
        self.extra_index = []

    def __str__(self):
        """
        PEP508 representation of the dependencies.
        """
        result = ""
        if self.index_url is not None:
            result += '# Index URL\n'
            result += f"{self.index_url}\n"
        if len(self.extra_index) > 0:
            result += "# Extra index urls\n"
            for url in self.extra_index:
                result += f"{url}\n"
        result += "# Runtime dependencies\n"
        for spec in self.dependencies.values():
            result += str(spec) + "\n"
        return result

    def __repr__(self):
        return self.__str__()

    def clear(self):
        self.dependencies.clear()
        self.extra_index = []
        self.index_url = None
        shutil.rmtree(self.python_packages, ignore_errors=True)
        shutil.rmtree(self.temp, ignore_errors=True)
        self.python_packages.mkdir(mode=0o700, exist_ok=True)
        self.temp.mkdir(mode=0o700, exist_ok=True)
        _logger.warning("Previously added dependencies have been removed.")

    def set_requirements(self, requirements_path: Union[str, os.PathLike]):
        self.clear()

        if Path(requirements_path).suffix == '.toml':
            dependencies, extra_index, index_url = pep508.parse_pyproject_toml(requirements_path)
        else:
            dependencies, extra_index, index_url = pep508.parse_requirements(requirements_path)

        for name, spec in dependencies.items():
            _check_package_for_dependency_limitations(spec.name)
            if not any(spec.name.lower() == dep.lower() for dep in self.dependencies):
                self.dependencies[name] = spec
                _logger.info(f"Runtime dependency added: {spec}")
            else:
                _logger.warning(f"Dependency already exists: {spec}")
        if index_url is not None:
            self.index_url = index_url
            _logger.info(f"Index url added: {index_url}")
        for url in extra_index:
            self.extra_index.append(url)
            _logger.info(f"Extra index url added: {url}")

    def add_dependencies(self, packages: list):
        for package in packages:
            if isinstance(package, tuple):
                name, version = package
                spec = pep508.parse_line(f"{name}=={version}")
            else:
                spec = pep508.parse_line(f"{package}")

            _check_package_for_dependency_limitations(spec.name)

            if not any(spec.name.lower() == dep.lower() for dep in self.dependencies):
                self.dependencies[spec.name] = spec
                _logger.info(f"Runtime dependency added: {spec}")
            else:
                _logger.warning(f"Dependency already exists: {spec}")

    def add_python_packages(self, path: Union[str, os.PathLike]) -> None:
        path: Path = Path(path)
        if not path.is_file():
            raise AssertionError(f"The file must be available on path {path.resolve()}")
        specs = []
        tmp = None
        if is_wheel_file(path):
            name, version = get_wheel_name_version(path)
            specs.append((name, version, path))
        elif is_pure_python_source(path):
            name, version = get_sdist_name_version(path)
            specs.append((name, version, path))
        elif zipfile.is_zipfile(path):
            tmp = Path(tempfile.mkdtemp(dir=self.temp))
            zip = zipfile.ZipFile(path)
            for pkg in zip.namelist():
                zip.extract(pkg, path=tmp)
                file = tmp / Path(pkg).name
                if is_wheel_file(file):
                    name, version = get_wheel_name_version(file)
                    specs.append((name, version, file))
                elif is_pure_python_source(file):
                    name, version = get_sdist_name_version(file)
                    specs.append((name, version, file))
                else:
                    _logger.warning(f"File skipped because it is not a wheel or pure python source: {pkg}")
        else:
            _logger.warning(f"File skipped because it is not a wheel or pure python source or a zip file: {path}")
        for name, version, path in specs:
            if name is not None:
                if version is None:
                    spec = pep508.parse_line(f"{name}")
                else:
                    spec = pep508.parse_line(f"{name}=={version}")

                _check_package_for_dependency_limitations(spec.name)

                if not any(spec.name.lower() == dep.lower() for dep in self.dependencies):
                    shutil.copy(path, self.python_packages)
                    self.dependencies[spec.name] = spec
                    _logger.info(f"Runtime dependency added: {spec}")
                else:
                    _logger.warning(f"Dependency already exists: {spec}")
        if tmp is not None:
            shutil.rmtree(tmp, ignore_errors=True)

    def _download_or_copy_dependency(self, name, version):
        dependency_url = urllib.parse.urlparse(version)  # raises ValueError if the url is invalid
        dependency_path = Path(urllib.parse.unquote(dependency_url.path))
        filename = dependency_path.name

        if "file" == dependency_url.scheme:
            # Possible Exceptions here: FileNotFoundError, PermissionError, OSError, IsADirectoryError, SameFileError
            if not dependency_path.is_file():
                raise FileNotFoundError(f"The dependency '{name}' can not be found on path '{dependency_path}'")

            if (self.python_packages / filename).is_file():
                _logger.warning(f"Dependency '{name}' will not be copied because it already exists in '{self.python_packages}' folder.")
            else:
                _logger.info(f"Dependency '{name}' will be copied to '{self.python_packages}' folder.")
                shutil.copy(dependency_path.resolve(), self.python_packages)
        else:
            # Possible Exceptions here: requests.exceptions.RequestException
            _logger.info(f"Dependency '{name}@{version}' will be downloaded from the repository.")

            response = requests.get(version)
            response.raise_for_status()

            with open(self.python_packages / filename, "wb") as f:
                f.write(response.content)
        return self.python_packages / filename

    def save(self, folder_path):
        # Downloads dependencies specified with url from remote repositories or copies them from local file system
        # Does not work with packages in source distribution format
        for name, dependency in self.dependencies.copy().items():
            if isinstance(dependency.version, str):
                try:
                    path = self._download_or_copy_dependency(name, dependency.version)
                    _, version = get_wheel_name_version(path)
                    self.dependencies[name] = pep508.parse_line(f"{name}=={version}")

                except requests.exceptions.RequestException as request_error:
                    raise RuntimeError(f"Failed to download dependency '{dependency.name}=={dependency.version}' from the repository.") from request_error

        requirements_file_path = folder_path / REQUIREMENTS_TXT
        with open(requirements_file_path, "w") as f:
            f.write(str(self))

        shutil.make_archive(
            base_name = folder_path / PYTHON_PACKAGES,
            root_dir = self.python_packages,
            format = 'zip',
            verbose = True,
            logger = _logger)

    def _check_if_index_url_is_set_to_pytorch_cpu(self):
        if self.index_url is None:
            return False
        if self.index_url.strip().startswith("--index-url") and _PYTORCH_CPU_REPO_URL in self.index_url:
            return True
        return False

    def enable_dependency_optimization(self):
        self.optimize_dependencies = True

    def disable_dependency_optimization(self):
        self.optimize_dependencies = False

    def validate(self):
        for spec in self.dependencies.values():
            _check_package_for_dependency_limitations(spec.name)

        found_gpu_dependency = any(dep in _GPU_DEPENDENCIES for dep in self.dependencies)
        if not found_gpu_dependency:
            return

        if self.optimize_dependencies:
            if self.index_url is None:
                self.index_url = f"--index-url {_PYTORCH_CPU_REPO_URL}"
                added_pypi_warning = ""
                if not any([_PYPI_REPO_URL in item for item in self.extra_index]):
                    self.extra_index.insert(0, f"--extra-index-url {_PYPI_REPO_URL}")
                    added_pypi_warning = ADDED_PYPI_WARNING_MSG
                _logger.warning(f"WARNING! {REPO_MODIFICATION_WARNING_MSG} {added_pypi_warning}")

            elif self._check_if_index_url_is_set_to_pytorch_cpu():
                if not any([_PYPI_REPO_URL in item for item in self.extra_index]):
                    self.extra_index.insert(0, f"--extra-index-url {_PYPI_REPO_URL}")
                    _logger.warning(f"WARNING! {REPO_MODIFICATION_WARNING_MSG} {ADDED_PYPI_WARNING_MSG}")

            else:
                user_defined_index_url = self.index_url.replace("--index-url", "--extra-index-url", 1)
                self.index_url = f"--index-url {_PYTORCH_CPU_REPO_URL}"
                self.extra_index.insert(0, user_defined_index_url)
                _logger.warning(f"WARNING! {REPO_MODIFICATION_WARNING_MSG} {INDEX_URL_MOVED_WARNING_MSG}")

        else:
            if not self._check_if_index_url_is_set_to_pytorch_cpu():
                _logger.warning(
                    "WARNING! The resulting package could contain unused GPU dependencies "
                    "which considerably increase the file size.")

`init(python_version='3.11', dir=None)` ¶

This class handles Python dependencies

Dependencies from remote repositories can be added via a requirements.txt file, or by calling the add_dependencies method. Dependencies can also be added as a single wheel or source distribution file, or as a collection in a zip archive via the add_python_packages method.

The class can be converted to string, which will contain the dependencies in PEP508 format.

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/python_dependencies.py

def __init__(self, python_version='3.11', dir: Union[str, os.PathLike] = None):
    """
    This class handles Python dependencies

    Dependencies from remote repositories can be added via a requirements.txt file,
    or by calling the add_dependencies method. Dependencies can also be added as
    a single wheel or source distribution file, or as a collection in a zip archive
    via the add_python_packages method.

    The class can be converted to string, which will contain the dependencies
    in PEP508 format.
    """
    self.python_version = python_version
    self.dependencies = {}
    self.python_packages = Path(tempfile.mkdtemp(dir=dir, prefix="dependencies_"))
    self.temp = Path(tempfile.mkdtemp(dir=dir, prefix="temp_"))

    self.optimize_dependencies = True
    self.index_url = None
    self.extra_index = []

`str()` ¶

PEP508 representation of the dependencies.

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/python_dependencies.py

def __str__(self):
    """
    PEP508 representation of the dependencies.
    """
    result = ""
    if self.index_url is not None:
        result += '# Index URL\n'
        result += f"{self.index_url}\n"
    if len(self.extra_index) > 0:
        result += "# Extra index urls\n"
        for url in self.extra_index:
            result += f"{url}\n"
    result += "# Runtime dependencies\n"
    for spec in self.dependencies.values():
        result += str(spec) + "\n"
    return result

`wheelhouse`¶

Methods for downloading and validating dependencies

This module collects all the necessary methods for downloading wheel or source distributions, and validation methods for checking if the whole collection could be installed in the AI Inference Server's Python runtime environment.

`assert_none_parameters(**kwargs)` ¶

Checks if any of the given parameters are None.

Returns:

Type	Description
`bool`	True if all parameters are not None

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def assert_none_parameters(**kwargs) -> bool:
    """
    Checks if any of the given parameters are None.

    Returns:
        True if all parameters are not None
    Raises:
        AssertionError: otherwise.
    """

    none_values = [k for k, v in kwargs.items() if v is None]
    if 0 < len(none_values):
        none_values = ", ".join(none_values)
        raise AssertionError(f"Parameters can not be None: {none_values}")
    return True

`is_wheel_file(path)` ¶

Checks whether the file on the given path is a wheel file.

Parameters:

Name	Type	Description	Default
`path`	`path - like`	The relative or absolute path of the wheel file.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def is_wheel_file(path: os.PathLike) -> bool:
    """
    Checks whether the file on the given `path` is a wheel file.

    Args:
        path (path-like): The relative or absolute path of the wheel file.
    Returns:
        bool: True if the zipfile contains a WHEEL text file, False otherwise.
    """
    if zipfile.is_zipfile(path):
        _wheel = zipfile.ZipFile(path)
        return 'WHEEL' in [f.split("/")[-1] for f in _wheel.namelist()]

    return False

`is_source_file(path)` ¶

Checks whether the file on the given path is a python source distribtion file.

Parameters:

Name	Type	Description	Default
`path`	`path - like`	The relative or absolute path of the zip or tar.gz archive file.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def is_source_file(path: os.PathLike) -> bool:
    """
    Checks whether the file on the given `path` is a python source distribtion file.

    Args:
        path (path-like): The relative or absolute path of the zip or tar.gz archive file.
    Returns:
        bool: True if the archive file contains a PKG-INFO text file, False otherwise.
    """

    if zipfile.is_zipfile(path):
        _archive = zipfile.ZipFile(path)
        return 'PKG-INFO' in [f.split("/")[-1] for f in _archive.namelist()]

    if tarfile.is_tarfile(path):
        with tarfile.open(path) as _archive:
            return 'PKG-INFO' in [f.split("/")[-1] for f in _archive.getnames()]

    return False

`is_pure_python_source(archive_path)` ¶

Checks whether the given source distribution contains only Python sources.

This method handles source distributions in a unified way. It searches for 'PKG-INFO' files, collects the programming languages used in the source, and returns True if only the Python language was used.

Parameters:

Name	Type	Description	Default
`archive_path`	`path - like`	The relative or absolute path of the zip or tar.gz archive file.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def is_pure_python_source(archive_path: Union[str, os.PathLike]):
    """
    Checks whether the given source distribution contains only Python sources.

    This method handles source distributions in a unified way.
    It searches for 'PKG-INFO' files, collects the programming languages
    used in the source, and returns True if only the Python language was used.

    Args:
        archive_path (path-like): The relative or absolute path of the zip or tar.gz archive file.
    Returns:
        bool: True if the archive file contains only Python sources.
    """
    headers = _extract_pkg_info(archive_path)
    if headers is None:
        return False
    classifiers = map(lambda header: header.get_all("classifier"), headers)
    classifiers = map(lambda classifier: [] if classifier is None else classifier, classifiers)
    classifiers = chain.from_iterable(classifiers)
    programming_languages = filter(lambda line: line.startswith('Programming Language ::'), classifiers)
    programming_languages = map(lambda line: line.split("::")[1].strip().lower(), programming_languages)
    return all(map(lambda txt: txt == 'python', programming_languages))

`get_wheel_name_version(archive_path)` ¶

Extracts the package name and version from a wheel file.

Parameters:

Name	Type	Description	Default
`archive_path`	`path - like`	The relative or absolute path of the wheel archive file.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def get_wheel_name_version(archive_path: Union[str, os.PathLike]):
    """
    Extracts the package name and version from a wheel file.

    Args:
        archive_path (path-like): The relative or absolute path of the wheel archive file.
    Returns:
        (str, str): The name and version of the wheel if successful, (None,None) otherwise.
    """
    with zipfile.ZipFile(archive_path, "r") as archive:
        files = archive.namelist()
        METADATA = list(filter(lambda filepath: filepath.endswith('METADATA'), files))
        if 0 < len(METADATA):
            headers = map(
                lambda filename: archive.read(filename).decode("utf-8"),
                METADATA)
            headers = map(lambda txt: Parser().parsestr(text=txt, headersonly=True), headers)
            headers = list(headers)[0]
            name = headers.get("name")
            version = headers.get("version")
            return (name, version)
    return (None, None)

`get_sdist_name_version(archive_path)` ¶

Extracts the package name and version from a source distribution file.

Parameters:

Name	Type	Description	Default
`archive_path`	`path - like`	The relative or absolute path of the zip or tar.gz archive file.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def get_sdist_name_version(archive_path: Union[str, os.PathLike]):
    """
    Extracts the package name and version from a source distribution file.

    Args:
        archive_path (path-like): The relative or absolute path of the zip or tar.gz archive file.
    Returns:
        (str, str): The name and version of the source distribution if successful, (None,None) otherwise.
    """
    headers = _extract_pkg_info(archive_path)
    if headers is None:
        return (None, None)
    headers = headers[0]
    name = headers.get("name")
    version = headers.get("version")
    return (name, version)

`check_directory_has_only_wheels_and_pure_sdist(python_packages_folder)` ¶

Checks all files in a directory if they are wheel files or pure Python source distributions.

Parameters:

Name	Type	Description	Default
`python_packages_folder`	`path - like`	The relative or absolute path of the directory to be checked.	required

Source code in docs/industrial-ai-suite/sdk/simaticai/packaging/wheelhouse.py

def check_directory_has_only_wheels_and_pure_sdist(python_packages_folder: Union[str, os.PathLike]):
    """
    Checks all files in a directory if they are wheel files or pure Python source distributions.

    Args:
        python_packages_folder (path-like): The relative or absolute path of the directory to be checked.
    Raises:
        AssertionError: If the directory contains other files than wheels or pure Python source distributions.
    """
    not_pure = []
    for file in list(python_packages_folder.iterdir()):
        if not (is_wheel_file(file) or is_pure_python_source(file)):
            not_pure.append(file.name)
    if 0 < len(not_pure):
        not_pure = "\n".join(not_pure)
        raise AssertionError(dedent(f"""
            One or more source dependencies are not pure Python sources.
            You need to convert them to wheel files for the target platform manually.
            List of not pure Python source distributions:
            {not_pure}
            """))

packaging¶

constants¶

python_dependencies¶

PythonDependencies ¶

__init__(python_version='3.11', dir=None) ¶

__str__() ¶

wheelhouse¶

assert_none_parameters(**kwargs) ¶

is_wheel_file(path) ¶

is_source_file(path) ¶

is_pure_python_source(archive_path) ¶

get_wheel_name_version(archive_path) ¶

get_sdist_name_version(archive_path) ¶

check_directory_has_only_wheels_and_pure_sdist(python_packages_folder) ¶

`constants`¶

`python_dependencies`¶

`PythonDependencies` ¶

`init(python_version='3.11', dir=None)` ¶

`str()` ¶

`wheelhouse`¶

`assert_none_parameters(**kwargs)` ¶

`is_wheel_file(path)` ¶

`is_source_file(path)` ¶

`is_pure_python_source(archive_path)` ¶

`get_wheel_name_version(archive_path)` ¶

`get_sdist_name_version(archive_path)` ¶

`check_directory_has_only_wheels_and_pure_sdist(python_packages_folder)` ¶