unpacker.py

ofrak.component.unpacker

Unpacker (AbstractComponent, ABC)

Unpackers are components that unpack resources, splitting them into one or more children.

children: Tuple[Optional[ofrak.model.tag_model.ResourceTag], ...] property readonly

A list of ResourceTags that an unpacker can unpack a resource into.

unpack(self, resource, config) async

Unpack the given resource.

Users should not call this method directly; rather, they should run Resource.run or Resource.unpack.

Parameters:

Name	Type	Description	Default
resource	Resource	The resource that is being unpacked	required
config	~CC	Optional config for unpacking. If an implementation provides a default, this default will always be used when config would otherwise be None. Note that a copy of the default config will be passed, so the default config values cannot be modified persistently by a component run.	required

Source code in ofrak/component/unpacker.py

@abstractmethod
async def unpack(self, resource: Resource, config: CC) -> None:
    """
    Unpack the given resource.

    Users should not call this method directly; rather, they should run
    [Resource.run][ofrak.resource.Resource.run] or
    [Resource.unpack][ofrak.resource.Resource.unpack].

    :param resource: The resource that is being unpacked
    :param config: Optional config for unpacking. If an implementation provides a default,
    this default will always be used when config would otherwise be None. Note that a copy of
    the default config will be passed, so the default config values cannot be modified
    persistently by a component run.
    """
    raise NotImplementedError()

_validate_unpacked_children(self, resource) private

Validate that the unpacked resources match the type defined by Unpacker.children.

Parameters:

Name	Type	Description	Default
resource	Resource		required

Exceptions:

Type	Description
ValueError	if the unpacked child does not match the type defined in Unpacker.children

Source code in ofrak/component/unpacker.py

def _validate_unpacked_children(self, resource: Resource) -> None:
    """
    Validate that the unpacked resources match the type defined by
    [Unpacker.children][ofrak.component.unpacker.Unpacker.children].

    :param resource:
    :raises ValueError: if the unpacked child does not match the type defined in
      [Unpacker.children][ofrak.component.unpacker.Unpacker.children]
    """
    component_context = resource.get_component_context()
    resource_context = resource.get_resource_context()
    untagged_descendants_allowed = None in self.children
    for descendant_id in component_context.resources_created:
        descendant_model = resource_context.resource_models.get(descendant_id)
        if descendant_model is None:
            raise ValueError(
                f"Cannot find descendant {descendant_id.decode()} for resource "
                f"{resource.get_id().decode()}."
            )
        descendant_has_tags = 0 != len(descendant_model.get_tags())

        if descendant_has_tags:
            if any(
                descendant_model.has_tag(descendant_tag)
                for descendant_tag in self.children
                if descendant_tag is not None
            ):
                continue
            else:
                valid_tag_patterns: List[str] = [
                    descendant_tag.__name__
                    for descendant_tag in self.children
                    if descendant_tag is not None
                ]
                if untagged_descendants_allowed:
                    valid_tag_patterns.append("untagged resource")
                expected_patterns = ", ".join(valid_tag_patterns)

                raise ValueError(
                    f"Unpacker {self.get_id().decode()} created resource {descendant_id.hex()} "
                    f"but its tags {descendant_model.get_tags()} do not match any of the "
                    f"expected patterns this unpacker should create: {expected_patterns}"
                )
        elif untagged_descendants_allowed:
            continue
        else:
            raise ValueError(
                f"Unpacker {self.get_id().decode()} created resource {descendant_id.hex()} but "
                f"its tags {descendant_model.get_tags()} do not match any of the expected "
                f"patterns this unpacker should create: "
                f"{', '.join([str(descendant_tag) for descendant_tag in self.children])}"
            )

UnpackerError (RuntimeError)

Base exception raised by unpackers.