p7zip.py

ofrak.core.p7zip

P7zFilesystem (GenericBinary, FilesystemRoot) dataclass

Filesystem stored in a 7z archive.

P7zPacker (Packer)

Pack files into a compressed 7z archive.

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 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/p7zip.py

async def pack(self, resource: Resource, config=None):
    p7zip_v: P7zFilesystem = await resource.view_as(P7zFilesystem)
    temp_flush_dir = await p7zip_v.flush_to_disk()
    temp_flush_dir = os.path.join(temp_flush_dir, ".")
    with tempfile.TemporaryDirectory() as temp_dir:
        temp_name = os.path.join(temp_dir, "temp.7z")
        command = ["7z", "a", temp_name, temp_flush_dir]
        subprocess.run(command, check=True, capture_output=True)
        with open(temp_name, "rb") as f:
            new_data = f.read()
        # Passing in the original range effectively replaces the original data with the new data
        resource.queue_patch(Range(0, await resource.get_data_length()), new_data)

P7zUnpacker (Unpacker)

Unpack (decompress) a 7z 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		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/p7zip.py

async def unpack(self, resource: Resource, config=None):
    p7zip_v = await resource.view_as(P7zFilesystem)
    resource_data = await p7zip_v.resource.get_data()
    with tempfile.NamedTemporaryFile() as temp_file:
        temp_file.write(resource_data)
        temp_file.flush()
        with tempfile.TemporaryDirectory() as temp_flush_dir:
            command = ["7z", "x", f"-o{temp_flush_dir}", temp_file.name]
            subprocess.run(command, check=True, capture_output=True)
            await p7zip_v.initialize_from_disk(temp_flush_dir)