Python script to fix invalid metadata values using bravo (requires my fork of bravo)

fix_meta.py

#!/usr/bin/env python
import json
import os
import string
import sys
import urllib2
from array import array
from collections import defaultdict
from StringIO import StringIO
from multiprocessing import Pool

from twisted.python.filepath import FilePath

from bravo.geometry.section import Section
from bravo.nbt import *
from bravo.region import Region
from bravo.utilities.bits import unpack_nibbles, pack_nibbles

def _load_sections_from_tag(section_tag):
    sections = [Section() for _ in range(16)]

    for tag in section_tag.tags:
        index = tag["Y"].value
        section = Section()
        section.blocks = array("B")
        section.blocks.fromstring(tag["Blocks"].value)
        section.metadata = array("B", unpack_nibbles(tag["Data"].value))
        section.blocklight = array("B", unpack_nibbles(tag["BlockLight"].value))
        section.skylight = array("B", unpack_nibbles(tag["SkyLight"].value))
        sections[index] = section

    return sections

def _save_sections_to_tag(sections):
    section_tag = TAG_List(type=TAG_Compound)

    for i, s in enumerate(sections):
        if s:
            section = TAG_Compound()
            section.name = ""
            section["Y"] = TAG_Byte(i)
            section["Blocks"] = TAG_Byte_Array()
            section["Blocks"].value = s.blocks.tostring()
            section["Data"] = TAG_Byte_Array()
            section["Data"].value = pack_nibbles(s.metadata)
            section["BlockLight"] = TAG_Byte_Array()
            section["BlockLight"].value = pack_nibbles(s.blocklight)
            section["SkyLight"] = TAG_Byte_Array()
            section["SkyLight"].value = pack_nibbles(s.skylight)
            section_tag.tags.append(section)

    return section_tag

def process_region(region):
    region.load_pages()

    chunkCoords = region.positions.keys()
    total = len(chunkCoords)
    for coords in chunkCoords:
        x, z = coords

        data = region.get_chunk(x, z)
        tag = NBTFile(buffer=StringIO(data))

        sections = _load_sections_from_tag(tag["Level"]["Sections"])
        map(process_section, sections)
        tag["Level"]["Sections"] = _save_sections_to_tag(sections)

        b = StringIO()
        tag.write_file(buffer=b)
        data = b.getvalue()

        region.put_chunk(x, z, data)

def process_section(section):
    for index, block in enumerate(section.blocks):
        if section.metadata[index] not in valid_meta[block]:
            section.metadata[index] = valid_meta[block][0]


if __name__ == "__main__":
    # Unbuffer stdout
    sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)

    if not os.path.exists("blocks.json"):
        print "Downloading blocks.json...",
        with open("blocks.json", 'w') as blocks:
            blocks.write(urllib2.urlopen("https://github.com/Thinkofname/mc-autodocs/raw/master/blocks.json").read())
            print "done."

    print "Parsing block data...",
    valid_meta = defaultdict(list)
    with open("blocks.json") as blocks:
        data = json.load(blocks)
        for block, meta in map(lambda x: x.split(":", 2), data.keys()):
            valid_meta[int(block)].append(int(meta))
    # Sort metadata
    valid_meta = {k: sorted(v) for k, v in valid_meta.iteritems()}
    # Blocks without metadata (data value must be 0)
    no_meta = tuple(k for k, v in valid_meta.iteritems() if v == [0])
    print "done."

    levels = [FilePath(sys.argv[1]) for _ in range(3)]
    for i, path in enumerate((None, "DIM-1", "DIM1")):
        if path: levels[i] = levels[i].child(path)
        levels[i] = levels[i].child("region")

    pool = Pool()
    regions = map(Region, reduce(list.__add__, map(lambda l: l.globChildren("*.mca"), levels)))
    count = 0
    total = len(regions)

    print "Fixing metadata... 0/%d\033[F" % total
    for _ in pool.imap_unordered(process_region, regions):
        count += 1
        print "Fixing metadata... %d/%d\033[F" % (count, total)

    print "Fixing metadata... done.\033[K"