Module refinery.lib.unrar.reader

Bit-level I/O for RAR decompression.

"""
Bit-level I/O for RAR decompression.
"""
from __future__ import annotations


class BitInput:
    """
    Bit-level reader matching unrar's BitInput class.
    MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.
    """
    __slots__ = ('buf', 'in_addr', 'in_bit')

    OVERREAD_SLACK = 4

    def __init__(self, data: bytes | bytearray | memoryview):
        self.buf = data
        self.in_addr = 0
        self.in_bit = 0

    def init(self):
        self.in_addr = 0
        self.in_bit = 0

    def load(self, data: bytes | bytearray | memoryview):
        """
        Continue with the given data from its start. A solid archive decodes each file with the
        same reader, so that whatever reads through it, like the PPM model, reads the new data.
        """
        self.buf = data
        self.init()

    def get_char(self) -> int:
        """
        Return the byte at the current byte position and advance past it, or zero at the end.
        """
        addr = self.in_addr
        if addr >= len(self.buf):
            return 0
        self.in_addr = addr + 1
        return self.buf[addr]

    def getbits(self) -> int:
        """
        Return 16 bits starting from current position (MSB-first, left-aligned).
        Does NOT advance the position.
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0

        val = (b0 << 16) | (b1 << 8) | b2
        return (val >> (8 - bit)) & 0xFFFF

    def getbits32(self) -> int:
        """
        Return 32 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0
        b3 = buf[addr + 3] if addr + 3 < blen else 0
        b4 = buf[addr + 4] if addr + 4 < blen else 0

        val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
        return (val >> (8 - bit)) & 0xFFFFFFFF

    def getbits64(self) -> int:
        """
        Return 64 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        buf = self.buf
        blen = len(buf)

        val = 0
        for k in range(addr, addr + 9):
            val = (val << 8) | (buf[k] if k < blen else 0)
        return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

    def addbits(self, bits: int):
        """
        Advance the position by the given number of bits.
        """
        total = (self.in_addr << 3) + self.in_bit + bits
        self.in_addr = total >> 3
        self.in_bit = total & 7

    @property
    def remaining(self) -> int:
        """
        Return approximate number of bytes remaining.
        """
        return max(0, len(self.buf) - self.in_addr)

    @property
    def overread(self) -> bool:
        """
        Whether the reader has advanced more than a small slack of bytes past the end of the
        input. Valid streams routinely read a few bytes past the end because getbits reads
        ahead; only advancing beyond the slack indicates a malformed or truncated stream.
        """
        return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

Classes

class BitInput (data)

Bit-level reader matching unrar's BitInput class. MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.

Expand source code Browse git

class BitInput:
    """
    Bit-level reader matching unrar's BitInput class.
    MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.
    """
    __slots__ = ('buf', 'in_addr', 'in_bit')

    OVERREAD_SLACK = 4

    def __init__(self, data: bytes | bytearray | memoryview):
        self.buf = data
        self.in_addr = 0
        self.in_bit = 0

    def init(self):
        self.in_addr = 0
        self.in_bit = 0

    def load(self, data: bytes | bytearray | memoryview):
        """
        Continue with the given data from its start. A solid archive decodes each file with the
        same reader, so that whatever reads through it, like the PPM model, reads the new data.
        """
        self.buf = data
        self.init()

    def get_char(self) -> int:
        """
        Return the byte at the current byte position and advance past it, or zero at the end.
        """
        addr = self.in_addr
        if addr >= len(self.buf):
            return 0
        self.in_addr = addr + 1
        return self.buf[addr]

    def getbits(self) -> int:
        """
        Return 16 bits starting from current position (MSB-first, left-aligned).
        Does NOT advance the position.
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0

        val = (b0 << 16) | (b1 << 8) | b2
        return (val >> (8 - bit)) & 0xFFFF

    def getbits32(self) -> int:
        """
        Return 32 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0
        b3 = buf[addr + 3] if addr + 3 < blen else 0
        b4 = buf[addr + 4] if addr + 4 < blen else 0

        val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
        return (val >> (8 - bit)) & 0xFFFFFFFF

    def getbits64(self) -> int:
        """
        Return 64 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        buf = self.buf
        blen = len(buf)

        val = 0
        for k in range(addr, addr + 9):
            val = (val << 8) | (buf[k] if k < blen else 0)
        return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

    def addbits(self, bits: int):
        """
        Advance the position by the given number of bits.
        """
        total = (self.in_addr << 3) + self.in_bit + bits
        self.in_addr = total >> 3
        self.in_bit = total & 7

    @property
    def remaining(self) -> int:
        """
        Return approximate number of bytes remaining.
        """
        return max(0, len(self.buf) - self.in_addr)

    @property
    def overread(self) -> bool:
        """
        Whether the reader has advanced more than a small slack of bytes past the end of the
        input. Valid streams routinely read a few bytes past the end because getbits reads
        ahead; only advancing beyond the slack indicates a malformed or truncated stream.
        """
        return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

Class variables

var OVERREAD_SLACK

The type of the None singleton.

Instance variables

var remaining

Return approximate number of bytes remaining.

Expand source code Browse git

@property
def remaining(self) -> int:
    """
    Return approximate number of bytes remaining.
    """
    return max(0, len(self.buf) - self.in_addr)

var overread

Whether the reader has advanced more than a small slack of bytes past the end of the input. Valid streams routinely read a few bytes past the end because getbits reads ahead; only advancing beyond the slack indicates a malformed or truncated stream.

Expand source code Browse git

@property
def overread(self) -> bool:
    """
    Whether the reader has advanced more than a small slack of bytes past the end of the
    input. Valid streams routinely read a few bytes past the end because getbits reads
    ahead; only advancing beyond the slack indicates a malformed or truncated stream.
    """
    return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

var buf

Expand source code Browse git

class BitInput:
    """
    Bit-level reader matching unrar's BitInput class.
    MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.
    """
    __slots__ = ('buf', 'in_addr', 'in_bit')

    OVERREAD_SLACK = 4

    def __init__(self, data: bytes | bytearray | memoryview):
        self.buf = data
        self.in_addr = 0
        self.in_bit = 0

    def init(self):
        self.in_addr = 0
        self.in_bit = 0

    def load(self, data: bytes | bytearray | memoryview):
        """
        Continue with the given data from its start. A solid archive decodes each file with the
        same reader, so that whatever reads through it, like the PPM model, reads the new data.
        """
        self.buf = data
        self.init()

    def get_char(self) -> int:
        """
        Return the byte at the current byte position and advance past it, or zero at the end.
        """
        addr = self.in_addr
        if addr >= len(self.buf):
            return 0
        self.in_addr = addr + 1
        return self.buf[addr]

    def getbits(self) -> int:
        """
        Return 16 bits starting from current position (MSB-first, left-aligned).
        Does NOT advance the position.
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0

        val = (b0 << 16) | (b1 << 8) | b2
        return (val >> (8 - bit)) & 0xFFFF

    def getbits32(self) -> int:
        """
        Return 32 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0
        b3 = buf[addr + 3] if addr + 3 < blen else 0
        b4 = buf[addr + 4] if addr + 4 < blen else 0

        val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
        return (val >> (8 - bit)) & 0xFFFFFFFF

    def getbits64(self) -> int:
        """
        Return 64 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        buf = self.buf
        blen = len(buf)

        val = 0
        for k in range(addr, addr + 9):
            val = (val << 8) | (buf[k] if k < blen else 0)
        return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

    def addbits(self, bits: int):
        """
        Advance the position by the given number of bits.
        """
        total = (self.in_addr << 3) + self.in_bit + bits
        self.in_addr = total >> 3
        self.in_bit = total & 7

    @property
    def remaining(self) -> int:
        """
        Return approximate number of bytes remaining.
        """
        return max(0, len(self.buf) - self.in_addr)

    @property
    def overread(self) -> bool:
        """
        Whether the reader has advanced more than a small slack of bytes past the end of the
        input. Valid streams routinely read a few bytes past the end because getbits reads
        ahead; only advancing beyond the slack indicates a malformed or truncated stream.
        """
        return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

var in_addr

Expand source code Browse git

class BitInput:
    """
    Bit-level reader matching unrar's BitInput class.
    MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.
    """
    __slots__ = ('buf', 'in_addr', 'in_bit')

    OVERREAD_SLACK = 4

    def __init__(self, data: bytes | bytearray | memoryview):
        self.buf = data
        self.in_addr = 0
        self.in_bit = 0

    def init(self):
        self.in_addr = 0
        self.in_bit = 0

    def load(self, data: bytes | bytearray | memoryview):
        """
        Continue with the given data from its start. A solid archive decodes each file with the
        same reader, so that whatever reads through it, like the PPM model, reads the new data.
        """
        self.buf = data
        self.init()

    def get_char(self) -> int:
        """
        Return the byte at the current byte position and advance past it, or zero at the end.
        """
        addr = self.in_addr
        if addr >= len(self.buf):
            return 0
        self.in_addr = addr + 1
        return self.buf[addr]

    def getbits(self) -> int:
        """
        Return 16 bits starting from current position (MSB-first, left-aligned).
        Does NOT advance the position.
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0

        val = (b0 << 16) | (b1 << 8) | b2
        return (val >> (8 - bit)) & 0xFFFF

    def getbits32(self) -> int:
        """
        Return 32 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0
        b3 = buf[addr + 3] if addr + 3 < blen else 0
        b4 = buf[addr + 4] if addr + 4 < blen else 0

        val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
        return (val >> (8 - bit)) & 0xFFFFFFFF

    def getbits64(self) -> int:
        """
        Return 64 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        buf = self.buf
        blen = len(buf)

        val = 0
        for k in range(addr, addr + 9):
            val = (val << 8) | (buf[k] if k < blen else 0)
        return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

    def addbits(self, bits: int):
        """
        Advance the position by the given number of bits.
        """
        total = (self.in_addr << 3) + self.in_bit + bits
        self.in_addr = total >> 3
        self.in_bit = total & 7

    @property
    def remaining(self) -> int:
        """
        Return approximate number of bytes remaining.
        """
        return max(0, len(self.buf) - self.in_addr)

    @property
    def overread(self) -> bool:
        """
        Whether the reader has advanced more than a small slack of bytes past the end of the
        input. Valid streams routinely read a few bytes past the end because getbits reads
        ahead; only advancing beyond the slack indicates a malformed or truncated stream.
        """
        return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

var in_bit

Expand source code Browse git

class BitInput:
    """
    Bit-level reader matching unrar's BitInput class.
    MSB-first bit ordering. Maintains InAddr (byte) and InBit (bit) positions.
    """
    __slots__ = ('buf', 'in_addr', 'in_bit')

    OVERREAD_SLACK = 4

    def __init__(self, data: bytes | bytearray | memoryview):
        self.buf = data
        self.in_addr = 0
        self.in_bit = 0

    def init(self):
        self.in_addr = 0
        self.in_bit = 0

    def load(self, data: bytes | bytearray | memoryview):
        """
        Continue with the given data from its start. A solid archive decodes each file with the
        same reader, so that whatever reads through it, like the PPM model, reads the new data.
        """
        self.buf = data
        self.init()

    def get_char(self) -> int:
        """
        Return the byte at the current byte position and advance past it, or zero at the end.
        """
        addr = self.in_addr
        if addr >= len(self.buf):
            return 0
        self.in_addr = addr + 1
        return self.buf[addr]

    def getbits(self) -> int:
        """
        Return 16 bits starting from current position (MSB-first, left-aligned).
        Does NOT advance the position.
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0

        val = (b0 << 16) | (b1 << 8) | b2
        return (val >> (8 - bit)) & 0xFFFF

    def getbits32(self) -> int:
        """
        Return 32 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        bit = self.in_bit
        buf = self.buf
        blen = len(buf)

        b0 = buf[addr] if addr < blen else 0
        b1 = buf[addr + 1] if addr + 1 < blen else 0
        b2 = buf[addr + 2] if addr + 2 < blen else 0
        b3 = buf[addr + 3] if addr + 3 < blen else 0
        b4 = buf[addr + 4] if addr + 4 < blen else 0

        val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
        return (val >> (8 - bit)) & 0xFFFFFFFF

    def getbits64(self) -> int:
        """
        Return 64 bits starting from current position (MSB-first).
        """
        addr = self.in_addr
        buf = self.buf
        blen = len(buf)

        val = 0
        for k in range(addr, addr + 9):
            val = (val << 8) | (buf[k] if k < blen else 0)
        return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

    def addbits(self, bits: int):
        """
        Advance the position by the given number of bits.
        """
        total = (self.in_addr << 3) + self.in_bit + bits
        self.in_addr = total >> 3
        self.in_bit = total & 7

    @property
    def remaining(self) -> int:
        """
        Return approximate number of bytes remaining.
        """
        return max(0, len(self.buf) - self.in_addr)

    @property
    def overread(self) -> bool:
        """
        Whether the reader has advanced more than a small slack of bytes past the end of the
        input. Valid streams routinely read a few bytes past the end because getbits reads
        ahead; only advancing beyond the slack indicates a malformed or truncated stream.
        """
        return self.in_addr > len(self.buf) + self.OVERREAD_SLACK

Methods

def init(self)

Expand source code Browse git

def init(self):
    self.in_addr = 0
    self.in_bit = 0

def load(self, data)

Continue with the given data from its start. A solid archive decodes each file with the same reader, so that whatever reads through it, like the PPM model, reads the new data.

Expand source code Browse git

def load(self, data: bytes | bytearray | memoryview):
    """
    Continue with the given data from its start. A solid archive decodes each file with the
    same reader, so that whatever reads through it, like the PPM model, reads the new data.
    """
    self.buf = data
    self.init()

def get_char(self)

Return the byte at the current byte position and advance past it, or zero at the end.

Expand source code Browse git

def get_char(self) -> int:
    """
    Return the byte at the current byte position and advance past it, or zero at the end.
    """
    addr = self.in_addr
    if addr >= len(self.buf):
        return 0
    self.in_addr = addr + 1
    return self.buf[addr]

def getbits(self)

Return 16 bits starting from current position (MSB-first, left-aligned). Does NOT advance the position.

Expand source code Browse git

def getbits(self) -> int:
    """
    Return 16 bits starting from current position (MSB-first, left-aligned).
    Does NOT advance the position.
    """
    addr = self.in_addr
    bit = self.in_bit
    buf = self.buf
    blen = len(buf)

    b0 = buf[addr] if addr < blen else 0
    b1 = buf[addr + 1] if addr + 1 < blen else 0
    b2 = buf[addr + 2] if addr + 2 < blen else 0

    val = (b0 << 16) | (b1 << 8) | b2
    return (val >> (8 - bit)) & 0xFFFF

def getbits32(self)

Return 32 bits starting from current position (MSB-first).

Expand source code Browse git

def getbits32(self) -> int:
    """
    Return 32 bits starting from current position (MSB-first).
    """
    addr = self.in_addr
    bit = self.in_bit
    buf = self.buf
    blen = len(buf)

    b0 = buf[addr] if addr < blen else 0
    b1 = buf[addr + 1] if addr + 1 < blen else 0
    b2 = buf[addr + 2] if addr + 2 < blen else 0
    b3 = buf[addr + 3] if addr + 3 < blen else 0
    b4 = buf[addr + 4] if addr + 4 < blen else 0

    val = (b0 << 32) | (b1 << 24) | (b2 << 16) | (b3 << 8) | b4
    return (val >> (8 - bit)) & 0xFFFFFFFF

def getbits64(self)

Return 64 bits starting from current position (MSB-first).

Expand source code Browse git

def getbits64(self) -> int:
    """
    Return 64 bits starting from current position (MSB-first).
    """
    addr = self.in_addr
    buf = self.buf
    blen = len(buf)

    val = 0
    for k in range(addr, addr + 9):
        val = (val << 8) | (buf[k] if k < blen else 0)
    return (val >> (8 - self.in_bit)) & 0xFFFFFFFFFFFFFFFF

def addbits(self, bits)

Advance the position by the given number of bits.

Expand source code Browse git

def addbits(self, bits: int):
    """
    Advance the position by the given number of bits.
    """
    total = (self.in_addr << 3) + self.in_bit + bits
    self.in_addr = total >> 3
    self.in_bit = total & 7