How to use the pyboy.plugins.rewind.CompressedFixedAllocBuffers function in pyboy

To help you get started, we’ve selected a few pyboy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Baekalfen / PyBoy / tests / test_rewind.py View on Github external
def test_all(self):
        for buf in [FixedAllocBuffers(), CompressedFixedAllocBuffers(), DeltaFixedAllocBuffers()]:
            A = [1] * 16
            B = [2] * 16
            C = [4] * 16
            # A = [x % 0x100 for x in range(0, 20)]
            # B = [x % 0x100 for x in range(10, 40)]
            # C = [x % 0x100 for x in range(5, 13)]
            D = [8] * 16

            for E in [A, B, C, D]:
                write_bytes(buf, E)
                buf.new()

            for E in [D, C, B, A]:
                assert buf.seek_frame(-1)
                tests = [(x, buf.read()) for x in E]
                assert all(list(map(lambda x: x[0] == x[1], tests)))
github Baekalfen / PyBoy / tests / test_rewind.py View on Github external
def test_compressed_buffer(self):
        buf = CompressedFixedAllocBuffers()

        write_bytes(buf, [0 for _ in range(10)])
        # Zeros are not written before a flush
        assert all(map(lambda x: x == FILL_VALUE, buf.buffer[:12]))

        # Now, we should see one pair of '0' and length
        buf.flush()
        assert all(map(lambda x: x == FILL_VALUE, buf.buffer[2:12]))
        assert buf.buffer[0] == 0
        assert buf.buffer[1] == 10

        # Second flush should do nothing
        buf.flush()
        assert all(map(lambda x: x == FILL_VALUE, buf.buffer[2:12]))
        assert buf.buffer[0] == 0
        assert buf.buffer[1] == 10
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
def commit(self):
        self.internal_pointer = 0
        self.injected_zero_frame = 0
        CompressedFixedAllocBuffers.commit(self)
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
# If the bytes is zero, it means that the next byte will be the counter
                self.zeros = FixedAllocBuffers.read(self)
                self.zeros -= 1
            return byte

    def new(self):
        FixedAllocBuffers.new(self)

    def commit(self):
        FixedAllocBuffers.commit(self)

    def seek_frame(self, v):
        return FixedAllocBuffers.seek_frame(self, v)


class DeltaFixedAllocBuffers(CompressedFixedAllocBuffers):
    """
    I chose to keep the code simple at the expense of some edge cases acting different from the other buffers.
    When seeking, the last frame will be lost. This has no practical effect, and is only noticeble in unittesting.
    """
    def __init__(self):
        CompressedFixedAllocBuffers.__init__(self)
        self.internal_pointer = 0
        self.prev_internal_pointer = 0
        # The initial values needs to be 0 to act as the "null-frame" and make the first frame a one-to-one copy
        # TODO: It would work with any values, but it makes it easier to debug
        self.internal_buffer = array.array("B", [0] * FIXED_BUFFER_MIN_ALLOC)
        self.internal_buffer_dirty = False

        # A side effect of the implementation will create a zero-frame in the beginning. Keep track of this,
        # as we don't want to use the section until we rollover the circular buffer.
        self.base_frame = 0
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
def flush_internal_buffer(self):
        # self.current_section += 1
        for n in range(self.prev_internal_pointer):
            CompressedFixedAllocBuffers.write(self, self.internal_buffer[n])
            # Make a null-frame so we can XOR the newest frame back in
            self.internal_buffer[n] = 0
        self.internal_buffer_dirty = False
        CompressedFixedAllocBuffers.new(self)
        self.injected_zero_frame = self.current_section
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
def __init__(self):
        CompressedFixedAllocBuffers.__init__(self)
        self.internal_pointer = 0
        self.prev_internal_pointer = 0
        # The initial values needs to be 0 to act as the "null-frame" and make the first frame a one-to-one copy
        # TODO: It would work with any values, but it makes it easier to debug
        self.internal_buffer = array.array("B", [0] * FIXED_BUFFER_MIN_ALLOC)
        self.internal_buffer_dirty = False

        # A side effect of the implementation will create a zero-frame in the beginning. Keep track of this,
        # as we don't want to use the section until we rollover the circular buffer.
        self.base_frame = 0
        # The frame we inject in the end to flush the last frame out
        self.injected_zero_frame = 0
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
def read(self):
        old_val = CompressedFixedAllocBuffers.read(self)
        data = old_val ^ self.internal_buffer[self.internal_pointer]
        self.internal_buffer[self.internal_pointer] = data
        self.internal_pointer += 1
        return data
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
def write(self, data):
        self.internal_buffer_dirty = True
        old_val = self.internal_buffer[self.internal_pointer]
        xor_val = data ^ old_val
        self.internal_buffer[self.internal_pointer] = data
        self.internal_pointer += 1
        return CompressedFixedAllocBuffers.write(self, xor_val)
github Baekalfen / PyBoy / pyboy / plugins / rewind.py View on Github external
if frames < 0:
            frames = -1
        else:
            frames = 1

        # Flush internal buffer to underlying memory. Otherwise, the newest frame, won't be seekable.
        if self.internal_buffer_dirty:
            self.flush_internal_buffer()
        self.internal_pointer = 0

        if frames > 0 and self.injected_zero_frame - 1 == self.current_section:
            return False
        elif frames < 0 and self.current_section - 1 == self.base_frame:
            return False

        return CompressedFixedAllocBuffers.seek_frame(self, frames)