libcmo21/PyCmo/VTUtils.py

import PyCmoMisc
import zlib, io, mmap, os
import typing

class RawFileReader():
    def __init__(self, filename: str):
        self.__size: int = os.path.getsize(filename)
        self.__fs = open(filename, 'rb')
        self.__mm: mmap.mmap = mmap.mmap(self.__fs.fileno(), 0, access = mmap.ACCESS_READ)
    
    def __del__(self):
        self.__mm.close()
        del self.__mm
        self.__fs.close()
        del self.__fs

    def GetSize(self) -> int:
        return self.__size

    def GetReader(self) -> mmap.mmap:
        return self.__mm

class LargeZlibFileReader():
    def __init__(self, raw_reader: RawFileReader, comp_size: int, uncomp_size: int):
        # set size
        self.__size: int = uncomp_size

        # create mmap
        self.__mm: mmap.mmap = mmap.mmap(-1, -1, access = mmap.ACCESS_WRITE)
        
        # decompress data
        reader = raw_reader.GetReader()
        parser: zlib._Decompress = zlib.decompressobj()

        buf = reader.read(io.DEFAULT_BUFFER_SIZE)
        while buf:
            self.__mm.write(parser.decompress(buf))
            buf = reader.read(io.DEFAULT_BUFFER_SIZE)
        self._mm.write(parser.flush())

    def __del__(self):
        self.__mm.close()
        del self.__mm

    def GetSize(self) -> int:
        return self.__size

    def GetReader(self) -> mmap.mmap:
        return self.__mm

class SmallZlibFileReader():
    def __init__(self, raw_reader: RawFileReader, comp_size: int, uncomp_size: int):
        # set size
        self.__size: int = uncomp_size

        # create io
        self.__ss: io.BytesIO = io.BytesIO()

        # decompress data
        reader = raw_reader.GetReader()
        parser: zlib._Decompress = zlib.decompressobj()

        buf = reader.read(io.DEFAULT_BUFFER_SIZE)
        while buf:
            self.__ss.write(parser.decompress(buf))
            buf = reader.read(io.DEFAULT_BUFFER_SIZE)
        self.__ss.write(parser.flush())

        # reset cursor
        self.__ss.seek(0, io.SEEK_SET)

    def __del__(self):
        del self.__ss

    def GetSize(self) -> int:
        return self.__size

    def GetReader(self) -> io.BytesIO:
        return self.__ss

UniversalFileReader = typing.Union[RawFileReader, LargeZlibFileReader, SmallZlibFileReader]