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()
command = ["zstd", "-T0", f"-{config.compression_level}"]
if config.compression_level > 19:
command.append("--ultra")
proc = await asyncio.create_subprocess_exec(
*command, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE
)
result, _ = await proc.communicate(uncompressed_data)
if proc.returncode:
raise CalledProcessError(returncode=proc.returncode, cmd=command)
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:
cmd = ["zstd", "-d", "-k"]
proc = await asyncio.create_subprocess_exec(
*cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE
)
result, _ = await proc.communicate(await resource.get_data())
if proc.returncode:
raise CalledProcessError(returncode=proc.returncode, cmd=cmd)
await resource.create_child(tags=(GenericBinary,), data=result)