decode_countyparcel.py

import sys
import json
from io import BytesIO
import geomet.wkb

def bytes2dict(x):
    '''Decode a dictionary encoded in bytes using the dicAsBytes method.'''
    b = BytesIO(x)
    count = read_int(b)
    result = {}
    for _ in range(count):
        key_len = read_int(b)
        key = b.read(key_len).decode('utf8')
        val_len = read_int(b)
        result[key] = b.read(val_len).decode('utf8')
    return result

def b2int(x):
    '''Convert a group of little-endian bytes to an int.'''
    return int.from_bytes(x, 'little')

def read_int(f):
    '''Read a 4-byte int'''
    return b2int(f.read(4))

def parse(f):
    '''Generate geojson features from the raw input.'''
    # that's apparently a bom?
    f.read(4)
    while read_int(f) > 0:
        id_len = read_int(f)
        b2int(f.read(id_len))
        layer_len = read_int(f)
        layer = f.read(layer_len).decode('utf8')
        # Read the wkb-encoded geometry.
        wkb_len = read_int(f)
        wkb = f.read(wkb_len)
        geom = geomet.wkb.loads(wkb)
        # Read the params
        params_len = read_int(f)
        params = f.read(params_len)
        params_dict = bytes2dict(params)
        params_dict['layer'] = layer
        yield {
            "type": "Feature",
            "properties": params_dict,
            "geometry": geom
        }

def main():
    with open(sys.argv[1], 'rb') as f:
        print('{ "type": "FeatureCollection", "features": [')
        # Save on memory by hackily printing one-by-one
        first = True
        for entry in parse(f):
            if not first:
                print(', ', end='')
            print(json.dumps(entry))
            first = False
        print(']}')

if __name__ == '__main__':
    main()