diffing.py

#!/usr/bin/env python3

import argparse
import pathlib
import re
import signal
import sys

import anvil
from anvil.chunk import _states_from_section, bin_append
from anvil.versions import VERSION_20w17a

AIR_ID = "minecraft:air"
AIR_IDS = {AIR_ID, "air"}


def parse_args():
    parser = argparse.ArgumentParser(
        description="Report block coordinates where old region blocks were replaced by air in the new region."
    )
    parser.add_argument("old_mca", help="Old region file path")
    parser.add_argument("new_mca", help="New region file path")
    parser.add_argument(
        "target_substring",
        nargs="?",
        default=None,
        help="Optional substring to filter old block IDs (e.g. pale_oak)",
    )
    return parser.parse_args()


class SectionDecoder:
    __slots__ = (
        "palette",
        "data",
        "mask",
        "bits",
        "stretches",
        "section_y",
    )

    def __init__(self, chunk=None, section=None):
        if section is None:
            self.section_y = None
            self.palette = []
            self.data = []
            self.bits = 4
            self.mask = 0xF
            self.stretches = False
            return

        self.section_y = section["Y"].value
        self.palette = self._build_palette(chunk, section)
        self.data = self._build_data(section)
        self.bits = max((len(self.palette) - 1).bit_length(), 4)
        self.mask = (1 << self.bits) - 1
        self.stretches = chunk.version < VERSION_20w17a

    @staticmethod
    def _build_palette(chunk, section):
        if "block_states" not in section and "Palette" not in section:
            return []

        try:
            raw_palette = chunk.get_palette(section)
        except KeyError:
            return []

        if not raw_palette:
            return []
        return [block.id for block in raw_palette]

    @staticmethod
    def _build_data(section):
        if "block_states" not in section and "BlockStates" not in section:
            return []

        try:
            return _states_from_section(section)
        except KeyError:
            return []

    def block_id(self, index):
        if not self.palette or not self.data:
            return AIR_ID

        if self.stretches:
            state_index = index * self.bits // 64
            if state_index >= len(self.data):
                return AIR_ID
            shift = (self.bits * index) % 64
            value = self.data[state_index] >> shift
            if 64 - shift < self.bits:
                if state_index + 1 >= len(self.data):
                    return AIR_ID
                value = bin_append(
                    self.data[state_index + 1] & ((1 << (self.bits - shift)) - 1),
                    value,
                    self.bits - shift,
                )
        else:
            state_index = index // (64 // self.bits)
            if state_index >= len(self.data):
                return AIR_ID
            shift = (index % (64 // self.bits)) * self.bits
            value = self.data[state_index] >> shift

        palette_id = value & self.mask
        result = self.palette[palette_id] if palette_id < len(self.palette) else AIR_ID
        return result.lower() if result is not None else AIR_ID


def _get_sections(chunk):
    sections = chunk.data.get("sections")
    if not sections:
        sections = chunk.data.get("Sections", [])
    return {section["Y"].value: section for section in sections}


def compare_chunks(old_chunk, new_chunk, target_substr):
    if getattr(old_chunk, "data", None) == getattr(new_chunk, "data", None):
        return

    old_sections = _get_sections(old_chunk)
    new_sections = _get_sections(new_chunk)

    for section_y, old_section in old_sections.items():
        new_section = new_sections.get(section_y)
        if new_section is None:
            new_decoder = SectionDecoder()
        else:
            if old_section == new_section:
                continue
            new_decoder = SectionDecoder(new_chunk, new_section)

        old_decoder = SectionDecoder(old_chunk, old_section)

        y_base = section_y * 16
        range_z = range(16)
        range_x = range(16)
        for y in range(16):
            row_base = (y * 16) << 4
            world_y = y_base + y
            for z in range_z:
                cell_base = row_base + (z << 4)
                for x in range_x:
                    index = cell_base + x
                    old_id = old_decoder.block_id(index)
                    if old_id not in AIR_IDS:
                        new_id = new_decoder.block_id(index)
                        if new_id in AIR_IDS:
                            if target_substr and target_substr not in old_id:
                                continue
                            yield x, world_y, z, old_id, new_id


def parse_region_coords(path):
    name = pathlib.Path(path).stem
    match = re.fullmatch(r"r\.(-?\d+)\.(-?\d+)", name)
    if not match:
        return 0, 0
    return int(match.group(1)), int(match.group(2))


def main():
    args = parse_args()
    target = args.target_substring
    if target:
        target = target.lower()

    old_region_x, old_region_z = parse_region_coords(args.old_mca)
    new_region_x, new_region_z = parse_region_coords(args.new_mca)

    if (old_region_x, old_region_z) != (new_region_x, new_region_z):
        sys.stderr.write(
            f"Warning: old and new region files have different region coordinates: "
            f"old=({old_region_x},{old_region_z}) new=({new_region_x},{new_region_z})\n"
        )

    old_region = anvil.Region.from_file(args.old_mca)
    new_region = anvil.Region.from_file(args.new_mca)

    found = 0
    for cx in range(32):
        for cz in range(32):
            try:
                old_chunk = old_region.get_chunk(cx, cz)
                new_chunk = new_region.get_chunk(cx, cz)
            except Exception:
                continue

            for x, y, z, old_id, new_id in compare_chunks(old_chunk, new_chunk, target):
                wx = old_region_x * 512 + cx * 16 + x
                wz = old_region_z * 512 + cz * 16 + z
                print(f"{wx} {y} {wz}: {old_id} -> {new_id}")
                found += 1

    if found == 0:
        print("No blocks found where old != air and new == air.")


if __name__ == "__main__":
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)
    try:
        main()
    except BrokenPipeError:
        sys.exit(0)