deployment¶

def find_dependencies(name: str, dependencies: dict) -> dict:
    """
    @Deprecated, reason: uses 'pip show' which only works for installed packages on the current platform.

    Collects all dependencies of the Python module given with its `name` in the current Python environment.

    All inherited dependencies will be added to the `dependencies` dictionary with the installed version of the module.
    The method executes an OS command like `python -m pip show scikit-learn`.

    Args:
        name (str): Name of the Python module to be searched through for its dependencies.
        dependencies (dict): Dictionary to collect the dependencies with the module name as key, and the installed version as value.

    Returns:
        dict: The `dependencies` dictionary with the collected module names and versions.
    """

    cmd_line = [sys.executable, '-m', 'pip', 'show', name]
    result = subprocess.run(cmd_line, stdout=subprocess.PIPE, text=True)

    if result.returncode != 0:
        print(f"Dependency {name} is not found and cannot be added.")
        return dependencies

    version = None
    for line in result.stdout.splitlines():

        version_matches = _version_matcher.match(line)
        if version_matches:
            version = version_matches.groups()[0].strip()

        transitive_matches = _transitive_matcher.match(line)
        if transitive_matches:
            transitives = transitive_matches.groups()[0].split(", ")
            for dependency in transitives:
                if dependency not in dependencies:
                    find_dependencies(dependency, dependencies)

    if name not in dependencies:
        spec = pep508.Spec(name, [], [('==', version)] if version else [], None)
        dependencies[name] = spec
        print("Found:", spec)
    return dependencies

def create_delta_package(origin_edge_package_zip_path: str, new_edge_package_zip_path: str) -> os.PathLike:
    """
    Creates a Delta Edge Configuration Package from two given Edge Configuration Packages.
    The created Delta Configuration Package is applicable to import into AI Inference Server,
    if the Original Edge Configuration Package is already imported there.
    The Delta Configuration Package only contains the additions and modifications
    in the New Edge Configuration Package compared to the Original one.
    That also means that no file deletion is possible in a deployed pipeline via this option.
    Please make sure that both of the given zip files come from a trusted source!

    Usage:
    ~~~python
    delta_package_path = deployment.create_delta_package('Edge-Config-edge-1.0.0.zip', 'Edge-Config-edge-1.1.0.zip')
    ~~~

    This method can be used from the command line, too.
    ```
    python -m simaticai create_delta_package <origin_package.zip> <modified_package.zip>
    ```

    Once the package is calculated, you will have an `Edge-Config-edge-delta-1.1.0.zip` file beside the updated package zip file.
    <ul>This package will contain
    <li><ul>the three configuration file for the package;
        <li>pipeline_config.yml</li>
        <li>runtime_config.yml</li>
        <li>datalink_metadata.yml</li>
    </li></ul>
    <li>the newly added files,</li>
    <li>and the updated files.</li>
    </ul>

    The package will not contain any information on the deleted files and they will be copied from the original pipeline.

    **Caution!**
    *If you change the version of a component in the pipeline, the delta package will contain all the files of the component because AI Inference Server identifies
    a component with a different version as a different component!*

    Args:
        origin_edge_package_zip_path (str): Path to the origin edge configuration package zip file.
        new_edge_package_zip_path (str): Path to the new edge configuration package zip file.

    Returns:
        os.PathLike: The path of the created delta edge package zip file.

    Raises:

        AssertionError:
            When:
            - either of the given edge packages is a delta package or
            - the names of the given edge packages differ or
            - the versions of the given edge packages are equal.
    """

    try:
        # checking whether the user has made a statement about telemetry collection
        is_telemetry_allowed()
    except RuntimeError as consent_is_undecided:
        raise consent_is_undecided

    workdir = Path(tempfile.mkdtemp(prefix="aisdk_deltapack-"))
    delta_dir  = Path(workdir / "delta")
    delta_dir.mkdir(parents=True)

    origin_dir = _extract_edge_package(origin_edge_package_zip_path, Path(workdir / "orig"))
    new_dir    = _extract_edge_package(new_edge_package_zip_path, Path(workdir / "new"))

    origin_package_info = _get_pipeline_info(origin_dir / PIPELINE_CONFIG)
    new_package_info = _get_pipeline_info(new_dir / PIPELINE_CONFIG)

    _validate_delta_package_inputs(origin_package_info, new_package_info)

    files_in_new_package = new_dir.rglob("*")
    for f in files_in_new_package:
        if f.is_dir():
            continue
        orig_file_path = origin_dir / f.relative_to(new_dir)
        if not orig_file_path.exists():
            _copy_file(f, new_dir, delta_dir)
        else:
            checksum_original = calc_sha(orig_file_path)
            checksum_new = calc_sha(f)
            if checksum_original != checksum_new:
                _copy_file(f, new_dir, delta_dir)

    _change_pipeline_config(delta_dir / PIPELINE_CONFIG, origin_package_info["dataFlowPipelineVersion"])

    new_edge_package_zip_path = Path(new_edge_package_zip_path)
    delta_path = _zip_delta_package(delta_dir, new_edge_package_zip_path)

    shutil.rmtree(workdir, ignore_errors=True)
    return Path(delta_path)