| # Copyright 2021 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """Python types for Assembly Input Bundles. |
| |
| Assembly Input Bundles are a set of artifacts that need to be delivered to out- |
| of-tree (OOT) assembly as a unit, but whose contents should be opaque to the |
| delivery system itself. |
| |
| """ |
| import functools |
| import os |
| import pathlib |
| import shutil |
| from collections import defaultdict |
| from dataclasses import dataclass, field |
| from typing import Any, Optional, TextIO, Union |
| |
| import serialization |
| from assembly import package_copier |
| from assembly.package_manifest import PackageManifest |
| from serialization import json_load |
| |
| from .common import FileEntry, FilePath, fast_copy_makedirs |
| from .image_assembly_config import KernelInfo |
| from .package_copier import PackageCopier |
| |
| __all__ = [ |
| "AIBCreator", |
| "AssemblyInputBundle", |
| "AssemblyInputBundleCreationException", |
| "ConfigDataEntries", |
| "DriverDetails", |
| "PackageDetails", |
| ] |
| |
| PackageManifestList = list[FilePath] |
| PackageName = str |
| ComponentName = str |
| DepSet = set[FilePath] |
| FileEntryList = list[FileEntry] |
| FileEntrySet = set[FileEntry] |
| ComponentShards = set[FilePath] |
| Merkle = str |
| BlobList = list[tuple[Merkle, FilePath]] |
| SubpackageManifests = dict[Merkle, FilePath] |
| ConfigDataEntries = dict[PackageName, set[FileEntry]] |
| |
| |
| class AssemblyInputBundleCreationException(Exception): |
| """To be raised when AIB creation fails for some reason""" |
| |
| ... |
| |
| |
| class DuplicatePackageException(AssemblyInputBundleCreationException): |
| """To be raised when an attempt is made to add multiple packages with the same name to the same |
| invocation of the AIBCreator""" |
| |
| ... |
| |
| |
| class PackageManifestParsingException(Exception): |
| """To be raised when an attempt to parse a json file into a PackageManifest object fails""" |
| |
| ... |
| |
| |
| @dataclass(order=True) |
| class DriverDetails: |
| """Details for constructing a driver manifest fragment from a driver package""" |
| |
| package: FilePath = field() # Path to the package manifest |
| components: set[FilePath] = field(default_factory=set) |
| |
| |
| @dataclass |
| class PackageDetails: |
| """Details for a package""" |
| |
| package: FilePath = field() # Path to the package manifest |
| set: str = field() # Package set that includes the package |
| |
| def __hash__(self) -> int: |
| """ |
| This intentionally only hashes the package manifest in order to |
| deduplicate packages across package sets. |
| """ |
| return hash(self.package) |
| |
| def __lt__(self, other: Any) -> bool: |
| return self.package < other.package |
| |
| |
| PackageDetailsList = list[PackageDetails] |
| |
| |
| @dataclass |
| class CompiledComponentDefinition: |
| """ |
| The definition of component to be compiled by Assembly |
| """ |
| |
| # Name of the component |
| component_name: str = field() |
| # Component shards to compile together |
| shards: set[FilePath] = field(default_factory=set) |
| |
| |
| @dataclass |
| class CompiledPackageDefinition: |
| """Primary definition of a compiled package which is created by Assembly""" |
| |
| # Name of the package |
| name: str = field() |
| # Dictionary mapping components to cml files by name |
| components: list[CompiledComponentDefinition] = field(default_factory=list) |
| # Other files to include in the compiled package |
| contents: set[FileEntry] = field(default_factory=set) |
| # CML files included by the component cml |
| includes: set[FilePath] = field(default_factory=set) |
| # Whether to extract the contents of this package into bootfs |
| bootfs_package: bool = field(default=False) |
| |
| |
| @dataclass |
| class CompiledPackageDefinitionFromGN: |
| """The CompilePackageDefinition which is written by GN for consuming by this tool. |
| |
| The key difference is that the 'includes' field has a different name and has |
| a different type of objects (FileEntry vs source paths). This is so that |
| they can be copied to their proper path within the include dir. |
| """ |
| |
| # Name of the package |
| name: str = field() |
| # Package manifests that include files to add to `contents` |
| packages: list[FilePath] = field(default_factory=list) |
| # Dictionary mapping components to cml files by name |
| components: list[CompiledComponentDefinition] = field(default_factory=list) |
| # Other files to include in the compiled package |
| contents: set[FileEntry] = field(default_factory=set) |
| # CML files included by the component cml |
| component_includes: set[FileEntry] = field(default_factory=set) |
| # Whether to extract the contents of this package into bootfs |
| bootfs_package: bool = field(default=False) |
| |
| |
| @dataclass |
| class AssemblyInputBundle: |
| """AssemblyInputBundle wraps a set of artifacts together for use by out-of-tree assembly, both |
| the manifest of the artifacts, and the artifacts themselves. |
| |
| The archived artifacts are placed into a nominal layout that is for readability, but the |
| JSON manifest itself is the arbiter of what artifacts are in what categories: |
| |
| file layout:: |
| |
| ./ |
| assembly_config.json |
| packages/ |
| name_of_aib.bootfs_files_package |
| <package name> |
| blobs/ |
| <merkle> |
| subpackages/ |
| <merkle> |
| bootfs/ |
| path/to/file/in/bootfs |
| config_data/ |
| <package name>/ |
| <file name> |
| compiled_packages/ |
| include/ |
| path/to/shard/file/in/tree |
| <compiled package name>/ |
| <component_name>/ |
| name.shard.cml |
| files/ |
| path/to/file/in/package |
| memory_buckets/ |
| path/to/bucket.json |
| kernel/ |
| kernel.zbi |
| multiboot.bin |
| |
| Files matching the patterns `packages/*/<package name>` and |
| `subpackages/<merkle>` are JSON representations of the |
| `PackageManifest` type (see `package_manifest.py`). (The `<merkle>` in |
| `subpackages/` is the merkle of the subpackage's metafar.) A named package |
| that is also referenced as a subpackage will have a copy of its manifest in |
| both directories. (This is an intentional duplication to allow the |
| simplification of the creation and use of the AIBs.) |
| |
| Each `PackageManifest` contains a `blobs` list of `BlobEntry` and an |
| optional `subpackages` list of `SubpackageEntry`. The `source_path` of a |
| `BlobEntry` references a file containing the blob bytes, in the `blobs/` |
| directory. The `manifest_path` of a `SubpackageEntry` references another |
| `PackageManifest` file in `subpackages/<merkle>`. |
| |
| assembly_config.json schema:: |
| |
| { |
| "packages": [ |
| { |
| "package": "package1", |
| "set": "base", |
| }, |
| { |
| "package": "package2", |
| "set": "base", |
| }, |
| { |
| "package": "package3", |
| "set": "cache", |
| }, |
| { |
| "package": "packageS1", |
| "set": "system", |
| }, |
| { |
| "package": "packageB1", |
| "set": "bootfs", |
| }, |
| ], |
| "base_drivers": [ "packageD1", ... ], |
| "boot_drivers": [ "packageD1", ... ], |
| "bootfs_files_package": "packages/packageB2", |
| "bootfs_files": [ |
| { "destination": "path/in/bootfs", source: "path/in/layout" }, |
| ... |
| ], |
| "config_data": { |
| "package1": [ |
| { "destination": "path/in/data/dir", "source": "path/in/layout" }, |
| { "destination": "path/in/data/dir", "source": "path/in/layout" }, |
| ... |
| ], |
| ... |
| }, |
| "blobs": [ "blobs/<hash1>", "blobs/<hash2>", ... ], |
| "kernel": { |
| "path": "kernel/kernel.zbi", |
| "args": [ "arg1", "arg2", ... ], |
| }, |
| "qemu_kernel": "kernel/multiboot.bin", |
| "boot_args": [ "arg1", "arg2", ... ], |
| "bootfs_shell_commands": { |
| "package1": |
| ["path/to/binary1", "path/to/binary2"] |
| }, |
| "shell_commands": { |
| "package1": |
| ["path/to/binary1", "path/to/binary2"] |
| }, |
| "packages_to_compile": [ |
| { |
| "name": "package_name", |
| "components": { |
| "component1": "path/to/component1.cml", |
| "component2": "path/to/component2.cml", |
| }, |
| "contents": [ |
| { |
| "source": "path/to/source", |
| "destination": "path/to/destination", |
| } |
| ] |
| }, |
| { |
| "name": "package_name", |
| "component_shards": { |
| "component1": [ |
| "path/to/shard1.cml", |
| "path/to/shard2.cml" |
| ] |
| } |
| } |
| ], |
| "memory_buckets": [ |
| "memory_buckets/path/to/bucket.json", |
| ], |
| } |
| |
| All items are optional. Files for `config_data` should be in the config_data section, |
| not in a package called `config_data`. |
| """ |
| |
| # Fields shared with ImageAssemblyConfig. |
| kernel: KernelInfo = field(default_factory=KernelInfo) |
| qemu_kernel: Optional[FilePath] = None |
| boot_args: set[str] = field(default_factory=set) |
| bootfs_files: set[FileEntry] = field(default_factory=set) |
| bootfs_packages: set[FilePath] = field(default_factory=set) |
| |
| packages: set[PackageDetails] = field(default_factory=set) |
| config_data: ConfigDataEntries = field(default_factory=dict) |
| blobs: set[FilePath] = field(default_factory=set) |
| base_drivers: list[DriverDetails] = field(default_factory=list) |
| boot_drivers: list[DriverDetails] = field(default_factory=list) |
| bootfs_shell_commands: dict[str, list[str]] = field( |
| default_factory=functools.partial(defaultdict, list) |
| ) |
| shell_commands: dict[str, list[str]] = field( |
| default_factory=functools.partial(defaultdict, list) |
| ) |
| packages_to_compile: list[CompiledPackageDefinition] = field( |
| default_factory=list |
| ) |
| bootfs_files_package: Optional[FilePath] = None |
| memory_buckets: set[FilePath] = field(default_factory=set) |
| |
| def __repr__(self) -> str: |
| """Serialize to a JSON string""" |
| return serialization.json_dumps(self, indent=2) |
| |
| def add_packages(self, packages: list[PackageDetails]) -> None: |
| for details in sorted(packages): |
| # This 'in' check only looks at the package manifest file path and |
| # ignores the package set. This is intentional in order to |
| # deduplicate packages across sets. |
| if details in self.packages: |
| raise ValueError(f"Duplicate package {details.package}") |
| self.packages.add(details) |
| |
| def all_file_paths(self) -> list[FilePath]: |
| """Return a list of all files that are referenced by this AssemblyInputBundle.""" |
| file_paths = [] |
| file_paths.extend([p.package for p in self.packages]) |
| if self.kernel.path is not None: |
| file_paths.append(self.kernel.path) |
| if self.qemu_kernel is not None: |
| file_paths.append(self.qemu_kernel) |
| for entries in self.config_data.values(): |
| file_paths.extend([entry.source for entry in entries]) |
| if self.blobs is not None: |
| file_paths.extend(self.blobs) |
| if self.memory_buckets: |
| file_paths.extend(self.memory_buckets) |
| |
| for package in self.packages_to_compile: |
| file_paths.extend(package.includes) |
| file_paths.extend([entry.source for entry in package.contents]) |
| for component in package.components: |
| file_paths.extend(component.shards) |
| |
| return file_paths |
| |
| def write_fini_manifest( |
| self, |
| file: TextIO, |
| base_dir: Optional[FilePath] = None, |
| rebase: Optional[FilePath] = None, |
| ) -> None: |
| """Write a fini-style manifest of all files in the AssemblyInputBundle |
| to the given |file|. |
| |
| fini manifests are in the format:: |
| |
| destination/path=path/to/source/file |
| |
| As all file paths in the AssemblyInputBundle are relative to the root of |
| the bundle, `destination/path` is the existing path. However, the path |
| to the source file cannot be known (absolutely) without additional |
| information. |
| |
| Arguments: |
| - file -- The |TextIO| file to write to. |
| - base_dir -- The folder to assume that file paths are relative from. |
| - rebase -- The folder to make the source paths relative to, if `base_dir` is also provided. |
| By default this is the cwd. |
| |
| If `base_dir` is given, it's used to construct the path to the source files, if not, the cwd |
| is assumed to be the path the files are from. |
| |
| If `rebase` is also given, the path to the source files are then made relative to it. |
| """ |
| file_paths = self.all_file_paths() |
| if base_dir is not None: |
| file_path_entries = [ |
| FileEntry( |
| os.path.relpath(os.path.join(base_dir, path), rebase), path |
| ) |
| for path in file_paths |
| ] |
| file_path_entries += [ |
| FileEntry( |
| os.path.join(base_dir, "assembly_config.json"), |
| "assembly_config.json", |
| ) |
| ] |
| else: |
| file_path_entries = [FileEntry(path, path) for path in file_paths] |
| |
| FileEntry.write_fini_manifest(file_path_entries, file) |
| |
| |
| class AIBCreator: |
| """The AIBCreator is a builder for AIBs that will copy all the files into |
| the correct layout for the AIB structure, rewriting package manifests |
| as needed to make them relative to the AIB manifest location. |
| |
| The AIBCreator has fields that match the AIB itself, but isn't an AIB |
| because the paths it contains are not valid for an AIB. |
| """ |
| |
| package_url_template = "{repository}/{package_name}" |
| |
| def __init__(self, outdir: FilePath) -> None: |
| # The directory to create the AIB in. The manifest will be placed in |
| # the root of this dir. |
| self.outdir = outdir |
| |
| # The packages (paths to package manifests) |
| self.packages: set[PackageDetails] = set() |
| |
| # The shell command configurations |
| self.shell_commands: dict[str, list[str]] = defaultdict(list) |
| self.bootfs_shell_commands: dict[str, list[str]] = defaultdict(list) |
| |
| # The kernel info |
| self.kernel = KernelInfo() |
| self.boot_args: set[str] = set() |
| |
| # The emulator kernel. |
| self.qemu_kernel: Optional[FilePath] = None |
| |
| # Bootfs info |
| self.bootfs_files: set[FileEntry] = set() |
| self.bootfs_files_package: Optional[FilePath] = None |
| self.bootfs_packages: set[FilePath] = set() |
| |
| # The config_data entries |
| self.config_data: FileEntryList = [] |
| |
| # Additional base drivers directly specified without requiring |
| # us to parse GN generated files |
| self.provided_base_driver_details: list[DriverDetails] = list() |
| |
| # Additional boot drivers directly specified without requiring |
| # us to parse GN generated files |
| self.provided_boot_driver_details: list[DriverDetails] = list() |
| |
| # A set containing all the unique packageUrls seen by the AIBCreator instance |
| self.package_urls: set[str] = set() |
| |
| # A list of CompiledPackageDefinitions from either a parsed json GN |
| # scope, or directly set by the legacy AIB creator. |
| self.compiled_packages: list[CompiledPackageDefinitionFromGN] = list() |
| |
| # Memory buckets to add to memory monitor. |
| self.memory_buckets: set[FilePath] = set() |
| |
| # The package copying mechanism. |
| self.package_copier: PackageCopier = PackageCopier(outdir) |
| |
| def build(self) -> tuple[AssemblyInputBundle, FilePath, DepSet]: |
| """ |
| Copy all the artifacts from the ImageAssemblyConfig into an AssemblyInputBundle that is in |
| outdir, tracking all copy operations in a DepFile that is returned with the resultant bundle. |
| |
| Some notes on operation: |
| - <outdir> is removed and recreated anew when called. |
| - hardlinks are used for performance |
| - the return value contains a list of all files read/written by the |
| copying operation (ie. depfile contents) |
| """ |
| |
| # Remove the existing <outdir>, and recreate it and the "subpackages" |
| # subdirectory. |
| if os.path.exists(self.outdir): |
| shutil.rmtree(self.outdir) |
| |
| # Track all files we read |
| deps: DepSet = set() |
| |
| # Init an empty resultant AssemblyInputBundle |
| result = AssemblyInputBundle() |
| |
| # Copy over the boot args and zbi kernel args, unchanged, into the resultant |
| # assembly bundle |
| result.boot_args = self.boot_args |
| kernel_args = self.kernel.args |
| if kernel_args: |
| result.kernel.args = kernel_args |
| |
| # Copy the manifests for the packages into the assembly bundle. |
| result.add_packages( |
| self._prepare_packages_for_copying(sorted(self.packages)) |
| ) |
| |
| # Copy the base drivers' packages. |
| result.base_drivers.extend( |
| sorted( |
| self._prepare_drivers_for_copying( |
| self.provided_base_driver_details, "base" |
| ) |
| ) |
| ) |
| |
| # Copy bootfs drivers' packages. |
| result.boot_drivers.extend( |
| sorted( |
| self._prepare_drivers_for_copying( |
| self.provided_boot_driver_details, "bootfs" |
| ) |
| ) |
| ) |
| |
| # Copy the package that contains the bootfs files, but only if it's not empty. |
| if self.bootfs_files_package: |
| deps.add(self.bootfs_files_package) |
| with open(self.bootfs_files_package, "r") as fname: |
| manifest = json_load(PackageManifest, fname) |
| if [ |
| path |
| for path in manifest.blobs_by_path().keys() |
| if path != "meta/" |
| ]: |
| # there are bootfs files, so include the package: |
| ( |
| bootfs_pkg_manifest_path, |
| _, |
| ) = self.package_copier.add_package( |
| self.bootfs_files_package |
| ) |
| result.bootfs_files_package = bootfs_pkg_manifest_path |
| |
| # Copy the memory bucket entries |
| _memory_buckets_entries = [ |
| FileEntry(src, os.path.basename(src)) for src in self.memory_buckets |
| ] |
| (memory_buckets_entries, memory_buckets_deps) = self._copy_file_entries( |
| _memory_buckets_entries, "memory_buckets" |
| ) |
| memory_buckets = [entry.source for entry in memory_buckets_entries] |
| deps.update(memory_buckets_deps) |
| result.memory_buckets.update(memory_buckets) |
| |
| # Add shell_commands field to assembly_config.json field in AIBCreator |
| result.bootfs_shell_commands = self.bootfs_shell_commands |
| result.shell_commands = self.shell_commands |
| |
| # Copy all the blobs to their dir in the out-of-tree layout |
| (all_blobs, pkg_copy_deps) = self.package_copier.perform_copy() |
| deps.update(pkg_copy_deps) |
| result.blobs = set(all_blobs) |
| |
| # Copy the bootfs entries |
| (bootfs, bootfs_deps) = self._copy_file_entries( |
| self.bootfs_files, "bootfs" |
| ) |
| deps.update(bootfs_deps) |
| result.bootfs_files.update(bootfs) |
| |
| # Rebase the path to the kernel into the out-of-tree layout |
| if self.kernel.path: |
| kernel_src_path: Any = self.kernel.path |
| kernel_filename = os.path.basename(kernel_src_path) |
| kernel_dst_path = os.path.join("kernel", kernel_filename) |
| result.kernel.path = kernel_dst_path |
| |
| # Copy the kernel itself into the out-of-tree layout |
| local_kernel_dst_path = os.path.join(self.outdir, kernel_dst_path) |
| deps.add(fast_copy_makedirs(kernel_src_path, local_kernel_dst_path)) |
| |
| # Rebase the path to the qemu kernel into the out-of-tree layout |
| if self.qemu_kernel: |
| kernel_src_path = self.qemu_kernel |
| kernel_filename = os.path.basename(kernel_src_path) |
| kernel_dst_path = os.path.join("kernel", kernel_filename) |
| result.qemu_kernel = kernel_dst_path |
| |
| # Copy the kernel itself into the out-of-tree layout |
| local_kernel_dst_path = os.path.join(self.outdir, kernel_dst_path) |
| deps.add(fast_copy_makedirs(kernel_src_path, local_kernel_dst_path)) |
| |
| # Track all the FileEntries for includes, to make sure that we don't get |
| # any duplicate destination paths with different source paths. |
| all_copied_include_entries: set[FileEntry] = set() |
| for package in self.compiled_packages: |
| components: list[CompiledComponentDefinition] = [] |
| for component_def in package.components: |
| copied_shards, component_deps = self._copy_component_shards( |
| component_def.shards, |
| package_name=package.name, |
| component_name=component_def.component_name, |
| ) |
| components.append( |
| CompiledComponentDefinition( |
| component_name=component_def.component_name, |
| shards=set(copied_shards), |
| ) |
| ) |
| deps.update(component_deps) |
| |
| # This assumes that package.includes has actually been passed to the |
| # AIB creator as set[FileEntry] instead of a set[FilePath]. This is |
| # not ideal, but it allows the reuse of the CompiledPackageDefinition |
| # type without any other changes. |
| # |
| # The FileEntries are only needed because of the SDK include paths |
| # used by some component shards. |
| # |
| # TODO(): Remove the use of the 'include' statement in component shards |
| # compiled by assembly, for all included cml files that aren't in the |
| # SDK itself (these can be found via a separate path to the SDK's set |
| # of cml include files). These files should either be incorporated |
| # into the body of the component shards in the AIB, or added to the |
| # AIBs as another shard for the same component. |
| # |
| # Once that's complete, this whole mechanism can be removed. |
| # |
| # _copy_component_includes will check for and validate inconsistent |
| # include entries. |
| ( |
| copied_include_entries, |
| component_includes_deps, |
| ) = self._copy_component_includes( |
| package.component_includes, all_copied_include_entries |
| ) |
| deps.update(component_includes_deps) |
| |
| # Save the copied entries so we can check for inconsistent |
| # duplicate include entries in other packages in this AIB |
| # in the following loop iterations |
| all_copied_include_entries.update(copied_include_entries) |
| |
| # The final copied includes to add to the AIB. |
| copied_includes = set( |
| map(lambda x: x.destination, copied_include_entries) |
| ) |
| |
| # Collect all the contents include the files from the input packages. |
| package_contents = [] |
| package_manifests = [] |
| for package_manifest_path in package.packages: |
| with open(package_manifest_path, "r") as f: |
| package_manifest = json_load(PackageManifest, f) |
| package_manifests += [package_manifest_path] |
| for blob in package_manifest.blobs: |
| # We do not include the meta.far, because assembly will |
| # generate a new one with all the contents. |
| if blob.path != "meta/": |
| package_contents.append( |
| FileEntry(str(blob.source_path), blob.path) |
| ) |
| contents = package.contents |
| contents.update(package_contents) |
| deps.update(package_manifests) |
| |
| # Copy the package contents entries |
| (copied_package_files, package_deps) = self._copy_file_entries( |
| contents, |
| os.path.join("compiled_packages", package.name, "files"), |
| ) |
| |
| copied_definition = CompiledPackageDefinition( |
| name=package.name, |
| components=components, |
| includes=copied_includes, |
| contents=set(copied_package_files), |
| bootfs_package=package.bootfs_package, |
| ) |
| result.packages_to_compile.append(copied_definition) |
| |
| deps.update(package_deps) |
| |
| # Copy the config_data entries into the out-of-tree layout |
| (config_data, config_data_deps) = self._copy_config_data_entries() |
| deps.update(config_data_deps) |
| result.config_data = config_data |
| |
| # Sort the shell commands alphabetically |
| result.bootfs_shell_commands = dict( |
| sorted(result.bootfs_shell_commands.items()) |
| ) |
| result.shell_commands = dict(sorted(result.shell_commands.items())) |
| |
| # Write the AIB manifest |
| assembly_config_path = os.path.join(self.outdir, "assembly_config.json") |
| with open(assembly_config_path, "w") as file: |
| serialization.json_dump(result, file, indent=2) |
| |
| return (result, assembly_config_path, deps) |
| |
| def _prepare_packages_for_copying( |
| self, |
| package_details_list: list[PackageDetails], |
| ) -> PackageDetailsList: |
| """Queue up the packages for copying, and return the list of package details, using the destination path""" |
| |
| # Resultant paths to package manifests |
| package_details: PackageDetailsList = [] |
| manifest_path_mapping: dict[FilePath, FilePath] = {} |
| |
| # Bail early if empty |
| if not package_details_list: |
| return package_details |
| |
| # Open each manifest, record the blobs, and then copy it to its destination, |
| # sorted by path to the package manifest. |
| for package_detail in package_details_list: |
| try: |
| (destination_path, manifest) = self.package_copier.add_package( |
| package_detail.package |
| ) |
| except package_copier.DuplicatePackageException as e: |
| raise DuplicatePackageException(e) |
| |
| self._validate_package_url(manifest, package_detail.set) |
| |
| # Track the package manifest in our set of packages |
| package_details.append( |
| PackageDetails(destination_path, package_detail.set) |
| ) |
| |
| # Create the mapping from the source to the copied path, so we can correct the driver |
| # details entries. |
| manifest_path_mapping[package_detail.package] = destination_path |
| |
| return package_details |
| |
| def _prepare_drivers_for_copying( |
| self, driver_details_list: list[DriverDetails], pkg_set: str |
| ) -> list[DriverDetails]: |
| """Queue up the package copying for each driver, returning a DriverDetails with the new destination path""" |
| |
| driver_details: list[DriverDetails] = [] |
| |
| for driver_detail in driver_details_list: |
| try: |
| (destination_path, manifest) = self.package_copier.add_package( |
| driver_detail.package |
| ) |
| except package_copier.DuplicatePackageException as e: |
| raise DuplicatePackageException(e) |
| |
| self._validate_package_url(manifest, pkg_set) |
| |
| driver_details.append( |
| DriverDetails( |
| destination_path, |
| driver_detail.components, |
| ) |
| ) |
| |
| return driver_details |
| |
| def _validate_package_url( |
| self, manifest: PackageManifest, pkg_set: str |
| ) -> None: |
| """Validate that a given package url only appears once in the set of packages being copied""" |
| package_url = AIBCreator.package_url_template.format( |
| repository=manifest.repository, |
| package_name=manifest.package.name, |
| ) |
| |
| # Validate that there's a single source for each package url. |
| if package_url not in self.package_urls: |
| self.package_urls.add(package_url) |
| else: |
| raise DuplicatePackageException( |
| f"There is a duplicate declaration of {package_url} in {pkg_set}" |
| ) |
| |
| def _copy_component_includes( |
| self, |
| component_includes: FileEntrySet, |
| existing_shard_includes: FileEntrySet, |
| ) -> tuple[FileEntrySet, DepSet]: |
| deps: DepSet = set() |
| shard_includes: FileEntrySet = set() |
| for entry in component_includes: |
| rebased_destination = os.path.join( |
| "compiled_packages", "include", entry.destination |
| ) |
| copy_destination = os.path.join(self.outdir, rebased_destination) |
| |
| rebased_entry = FileEntry(entry.source, rebased_destination) |
| |
| # Check whether we have previously specified a different |
| # source for the same include file |
| if rebased_destination in map( |
| lambda x: x.destination, |
| existing_shard_includes | shard_includes, |
| ): |
| if rebased_entry not in existing_shard_includes: |
| raise AssemblyInputBundleCreationException( |
| f"Include file already exists with a different source: {copy_destination}" |
| ) |
| else: |
| # Hardlink the file from the source to the destination |
| deps.add(fast_copy_makedirs(entry.source, copy_destination)) |
| |
| shard_includes.add(rebased_entry) |
| |
| return shard_includes, deps |
| |
| def _copy_component_shard( |
| self, component_shard: FilePath, package_name: str, component_name: str |
| ) -> tuple[FilePath, DepSet]: |
| deps: DepSet = set() |
| # The shard is copied to a path based on the name of the package, the |
| # name of the component, and the filename of the shard: |
| # f"compiled_packages/{package_name}/{component_name}/{filename}" |
| # |
| bundle_destination = os.path.join( |
| "compiled_packages", |
| package_name, |
| component_name, |
| os.path.basename(component_shard), |
| ) |
| |
| # The copy destination is the above path, with the bundle's outdir. |
| copy_destination = os.path.join(self.outdir, bundle_destination) |
| |
| # Hardlink the file from the source to the destination |
| deps.add(fast_copy_makedirs(component_shard, copy_destination)) |
| return bundle_destination, deps |
| |
| def _copy_component_shards( |
| self, |
| component_shards: Union[ComponentShards, list[FilePath]], |
| package_name: str, |
| component_name: str, |
| ) -> tuple[list[FilePath], DepSet]: |
| shard_file_paths: list[FilePath] = list() |
| deps: DepSet = set() |
| for shard in component_shards: |
| destination, copy_deps = self._copy_component_shard( |
| shard, package_name, component_name |
| ) |
| deps.update(copy_deps) |
| shard_file_paths.append(destination) |
| return shard_file_paths, deps |
| |
| def _copy_file_entries( |
| self, entries: Union[FileEntrySet, FileEntryList], subdirectory: str |
| ) -> tuple[FileEntryList, DepSet]: |
| results: FileEntryList = [] |
| deps: DepSet = set() |
| |
| # Bail early if nothing to do |
| if len(entries) == 0: |
| return (results, deps) |
| |
| for entry in entries: |
| rebased_destination = os.path.join(subdirectory, entry.destination) |
| copy_destination = os.path.join(self.outdir, rebased_destination) |
| |
| # Hardlink the file from source to the destination, relative to the |
| # directory for all entries. |
| deps.add(fast_copy_makedirs(entry.source, copy_destination)) |
| |
| # Make a new FileEntry, which has a source of the path within the |
| # out-of-tree layout, and the same destination. |
| results.append( |
| FileEntry( |
| source=rebased_destination, destination=entry.destination |
| ) |
| ) |
| |
| return (results, deps) |
| |
| def _copy_config_data_entries(self) -> tuple[ConfigDataEntries, DepSet]: |
| """ |
| Take a list of entries for the config_data package, copy them into the |
| appropriate layout for the assembly input bundle, and then return the |
| config data entries and the set of DepEntries from the copying |
| |
| This expects the entries to be destined for: |
| `meta/data/<package>/<path/to/file>` |
| |
| and creates a ConfigDataEntries dict of PackageName:FileEntryList. |
| """ |
| results: ConfigDataEntries = {} |
| deps: DepSet = set() |
| |
| if len(self.config_data) == 0: |
| return (results, deps) |
| |
| # Make a sorted list of a deduplicated set of the entries. |
| for entry in sorted(set(self.config_data)): |
| # Crack the in-package path apart |
| # |
| # "meta" / "data" / package_name / path/to/file |
| parts = pathlib.Path(entry.destination).parts |
| if parts[:2] != ("meta", "data"): |
| raise ValueError( |
| "Found an unexpected destination path: {}".format(parts) |
| ) |
| package_name = parts[2] |
| file_path = os.path.join(*parts[3:]) |
| |
| rebased_source_path = os.path.join( |
| "config_data", package_name, file_path |
| ) |
| copy_destination = os.path.join(self.outdir, rebased_source_path) |
| |
| # Hardlink the file from source to the destination |
| deps.add(fast_copy_makedirs(entry.source, copy_destination)) |
| |
| # Append the entry to the set of entries for the package |
| results.setdefault(package_name, set()).add( |
| FileEntry(source=rebased_source_path, destination=file_path) |
| ) |
| |
| return (results, deps) |
