zstd.py

ofrak.core.zstd

ZstdData (GenericBinary)

A zstd binary blob.

ZstdPacker (Packer)

Pack data into a compressed zstd file.

pack(self, resource, config=None) async

Pack the given resource.

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

Parameters:

Name	Type	Description	Default
resource	Resource		required
config	Optional[ofrak.core.zstd.ZstdPackerConfig]	Optional config for packing. 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.	None

Source code in ofrak/core/zstd.py

async def pack(self, resource: Resource, config: Optional[ZstdPackerConfig] = None):
    if config is None:
        config = ZstdPackerConfig(compression_level=19)
    zstd_view = await resource.view_as(ZstdData)
    child_file = await zstd_view.get_child()
    uncompressed_data = await child_file.resource.get_data()

    with tempfile.NamedTemporaryFile() as uncompressed_file:
        uncompressed_file.write(uncompressed_data)
        uncompressed_file.flush()
        output_filename = tempfile.mktemp()

        command = ["zstd", "-T0", f"-{config.compression_level}"]
        if config.compression_level > 19:
            command.append("--ultra")
        command.extend([uncompressed_file.name, "-o", output_filename])
        proc = await asyncio.create_subprocess_exec(
            *command,
        )
        returncode = await proc.wait()
        if proc.returncode:
            raise CalledProcessError(returncode=returncode, cmd=command)
        with open(output_filename, "rb") as f:
            result = f.read()

        compressed_data = result
        original_size = await zstd_view.resource.get_data_length()
        resource.queue_patch(Range(0, original_size), compressed_data)

ZstdPackerConfig (ComponentConfig) dataclass

ZstdPackerConfig(compression_level: int)

ZstdUnpacker (Unpacker)

Unpack (decompress) a zstd file.

unpack(self, resource, config=None) 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	ComponentConfig	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.	None

Source code in ofrak/core/zstd.py

async def unpack(self, resource: Resource, config: ComponentConfig = None) -> None:
    with tempfile.NamedTemporaryFile(suffix=".zstd") as compressed_file:
        compressed_file.write(await resource.get_data())
        compressed_file.flush()
        output_filename = tempfile.mktemp()

        cmd = [
            "zstd",
            "-d",
            "-k",
            compressed_file.name,
            "-o",
            output_filename,
        ]
        proc = await asyncio.create_subprocess_exec(
            *cmd,
        )
        returncode = await proc.wait()
        if proc.returncode:
            raise CalledProcessError(returncode=returncode, cmd=cmd)
        with open(output_filename, "rb") as f:
            result = f.read()

        await resource.create_child(tags=(GenericBinary,), data=result)