Blades of Steel PPU data (CHR/nametable) decoder and encoder

blades_of_steel_ppudata.py

#!/usr/bin/env python3

#
# Python script for decoding or re-encoding PPU data
# (CHR or nametable) bundles from Blades of Steel.
#
# See bottom of this file for example usage.
#

rom_filename = "Blades of Steel (U).nes"
rom = open(rom_filename,"rb").read()
banks = (len(rom)-16) // 0x4000
print("ROM: " + rom_filename)
print("%d bytes, %d banks" % (len(rom),banks))

def rom_addr(addr,bank=banks-1): # UNROM banked addressing
    if (addr >= 0xC000):
        bank = banks-1 # fixed bank
    return 16 + ((bank * 0x4000) | (addr & 0x3FFF))

def rom_get(addr,bank=banks-1):
    return rom[rom_addr(addr,bank)]

bundle_pointer_table = 0xC89D
bundle_bank_table    = 0xC8BB

# for high level info about the bundles
# returns (address, bank, size) of bundle read
def inspect_bundle(x, silent=False, verbose=False): # X = bundle index * 2
    if verbose:
        silent = False
    ptr = \
         rom_get(bundle_pointer_table+0+x) + \
        (rom_get(bundle_pointer_table+1+x) << 8)
    b = rom_get(bundle_bank_table+(x//2))
    if not silent:
        print("Bundle %2d ($%02X) at $%01X:%04X" % (x//2,x,b,ptr))
    ppu_out = 0
    block_size = 0
    pos = 0
    finished = False
    while not finished:
        ppu_out = rom_get(ptr+pos,b) + (rom_get(ptr+pos+1,b) << 8)
        pos += 2
        while True:
            c = rom_get(ptr+pos,b)
            pos += 1
            if c == 0xFF or c == 0x7F:
                if not silent:
                    print("Block: $%04X %d bytes" % (ppu_out, block_size))
                block_size = 0
                if c == 0xFF:
                    finished = True
                break
            run = c & 0x7F
            if run == 0:
                run = 256
                if (verbose):
                    print("Invalid run length at $%01X:%04X (%02X)" % (b,ptr+pos-1,c))
            if c < 0x80:
                d = rom_get(ptr+pos,b)
                pos += 1
                block_size += run
                if (verbose):
                    print("RLE %d x $%02X" % (run, d))
            else:
                d = []
                for i in range(run):
                    d.append(rom_get(ptr+pos+i,b))
                if (run == 256):
                    d[255] = c
                else:
                    pos += run
                block_size += run
                if (verbose):
                    print("Raw %d" % (run))
                if (run == 256):
                    raise Exception("invalid raw run length at $%01X:%04X" % (b,ptr+pos-1))
    if not silent:
        print ("%d bytes decoded" % (pos))
    return (ptr, b, pos)

# to extract the actual bundle data
def decode_bundle(x):
    ptr = \
         rom_get(bundle_pointer_table+0+x) + \
        (rom_get(bundle_pointer_table+1+x) << 8)
    b = rom_get(bundle_bank_table+(x//2))
    blocks = []
    pos = 0
    finished = False
    while not finished:
        ppu_out = rom_get(ptr+pos,b) + (rom_get(ptr+pos+1,b) << 8)
        block = [ppu_out,[]]
        pos += 2
        while True:
            c = rom_get(ptr+pos,b)
            pos += 1
            if c == 0xFF or c == 0x7F:
                blocks.append(block)
                if c == 0xFF:
                    finished = True
                break
            run = c & 0x7F
            if run == 0:
                run = 256
            if c < 0x80:
                d = rom_get(ptr+pos,b)
                pos += 1
                block[1].extend([d] * run)
            else:
                d = []
                for i in range(run):
                    d.append(rom_get(ptr+pos+i,b))
                if (run == 256):
                    d[255] = c
                else:
                    pos += run
                block[1].extend(d)
                if (run == 256):
                    raise Exception("invalid raw run length at $%01X:%04X" % (b,ptr+pos-1))
    return blocks

# to decode a bundle from a linear block of data
def decode_raw_bundle(data):
    blocks = []
    pos = 0
    finished = False
    while not finished:
        ppu_out = data[pos] + (data[pos+1] << 8)
        block = [ppu_out,[]]
        pos += 2
        while True:
            c = data[pos]
            pos += 1
            if c == 0xFF or c == 0x7F:
                blocks.append(block)
                if c == 0xFF:
                    finished = True
                break
            run = c & 0x7F
            if run == 0:
                run = 256
            if c < 0x80:
                d = data[pos]
                pos += 1
                block[1].extend([d] * run)
            else:
                d = []
                for i in range(run):
                    d.append(data[pos+i])
                if (run == 256):
                    d[255] = c
                else:
                    pos += run
                block[1].extend(d)
                if (run == 256):
                    raise Exception("invalid raw run length at $%04X" % (pos-1))
    return blocks

# for reading the raw data from the bundle in the ROM
# use inspect_bundle to get the needed parameters
def read_bundle_raw(addr, bank, size):
    d = []
    for i in range(size):
        d.append(rom_get(addr+i, bank))
    return d

# encode a single block into a packet of bytes
def encode_block(block):

    def consecutive(i): # how many matching consecutive bytes start at i
        c = d[i]
        count = 0
        while (i+count) < len(d) and c == d[i+count]:
            count += 1
        return min(count,126)
            
    (ppu_out, d) = block
    do = [ (ppu_out & 0xFF), (ppu_out >> 8) ]
    while len(d) > 0:
        run = consecutive(0)
        if (run > 2): # RLE
            do.append(run)
            do.append(d[0])
            d = d[run:]
            continue
        run = min(len(d),126)
        for i in range(1,run):
            if consecutive(i) > 2:
                run = i # break here for next RLE
                break
        # non-RLE
        do.append(0x80 | run)
        do.extend(d[0:run])
        d = d[run:]
        continue
    return do

# encode a bundle into a packet of bytes
def encode_bundle(blocks):
    d = []
    for b in blocks:
        d.extend(encode_block(b))
        d.append(0x7F) # next block control byte
    d[len(d)-1] = 0xFF # convert next block control to end of blocks
    return d

# for inspecting data
def string_hex(d, per_line=32):
    s = ""
    for i in range(len(d)):
        if (i % per_line) == 0:
            s += "\t"
        else:
            s += " "
        s += "%02X" % (d[i])
        if (i % per_line) == (per_line-1):
            s += "\n"
    return s

# for inspecting blocks
def print_block(block):
    print ("PPU $%04X" % (block[0]))
    print (string_hex(block[1]))
    return

# for inspecting bundles
def print_bundle(blocks):
    for b in blocks:
        print_block(b);


for x in range(15):
    print()
    (addr, bank, size) = inspect_bundle(x*2)
    b = decode_bundle(x*2)
    d = encode_bundle(b)
    #print("Re-encoded: %d bytes" % len(d))
    print()
    print("Original decoded:")
    print_bundle(b)
    #print("Re-encoded decoded:")
    #print_bundle(decode_raw_bundle(d))
    print("Original data:")
    print(string_hex(read_bundle_raw(addr,bank,size)))
    #print("Re-encoded data:")
    #print(string_hex(d))