#!/usr/bin/env python3 """Rebuild a full-resolution reticle PNG from its tiles, byte for byte. This file is deliberately standalone: it imports nothing outside the Python standard library and nothing from the rest of this package, because a verbatim copy of it is published next to the tiles so that whoever downloads them can put the big image back together without installing anything. ``tests/test_tiles.py`` enforces both properties. How it can be a plain concatenation ----------------------------------- PNG stores each scanline with a filter type byte. Filter 2 (*Up*) predicts a pixel from the one directly above it - ``F[y][x] = raw[y][x] - raw[y-1][x] (mod 256)`` - and never looks at the pixel to its left. So a tile filtered with *Up* holds exactly the bytes the full image would hold for the columns that tile covers, and the full image's filtered scanline is the tiles-in-that-row's filtered scanlines laid end to end. The one row where that breaks is the first row of each tile, whose "row above" is the implicit all-zero row inside the tile but a real row of pixels in the full image. The writer therefore puts filter 0 (*None*, the raw bytes) on exactly those rows in both the tiles and the full image, so they agree there too. That makes rebuilding pure byte movement: no unfiltering, no re-filtering, no pixel arithmetic at all. Every expensive step stays inside zlib. """ from __future__ import annotations import argparse import hashlib import json import struct import sys import zlib from pathlib import Path PNG_SIG = b"\x89PNG\r\n\x1a\n" # The full image is only reproduced bit for bit if it is deflated exactly as it was # written, so these are part of the format and are repeated in the manifest. ZLIB_LEVEL, ZLIB_WBITS, ZLIB_MEMLEVEL, ZLIB_STRATEGY = 9, 15, 9, zlib.Z_DEFAULT_STRATEGY # A new IDAT chunk is started once this many compressed bytes have accumulated. Writer # and reader must agree, because chunk boundaries are part of the file's bytes. IDAT_LIMIT = 1 << 20 FILTER_NONE, FILTER_UP = 0, 2 def chunk(typ: bytes, data: bytes) -> bytes: """One PNG chunk: length, type, payload, CRC32 of type+payload.""" return struct.pack(">I", len(data)) + typ + data + struct.pack(">I", zlib.crc32(typ + data)) class PngWriter: """Writes an 8-bit greyscale PNG from already-filtered scanlines.""" def __init__(self, fh, width: int, height: int): self.fh, self.width, self.height = fh, width, height self.rows = 0 self.stream = hashlib.sha256() # over the filtered scanlines, pre-deflate self.png = hashlib.sha256() # over the file as written self._co = zlib.compressobj(ZLIB_LEVEL, zlib.DEFLATED, ZLIB_WBITS, ZLIB_MEMLEVEL, ZLIB_STRATEGY) self._pending = bytearray() self._write(PNG_SIG) self._write(chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 0, 0, 0, 0))) def _write(self, b: bytes) -> None: self.png.update(b) self.fh.write(b) def row(self, data: bytes) -> None: """Append one filtered scanline: a filter type byte followed by ``width`` bytes.""" if len(data) != self.width + 1: raise ValueError(f"scanline {self.rows} is {len(data)} bytes, expected {self.width + 1}") self.rows += 1 self.stream.update(data) self._pending += self._co.compress(data) if len(self._pending) >= IDAT_LIMIT: self._flush_idat() def _flush_idat(self) -> None: self._write(chunk(b"IDAT", bytes(self._pending))) self._pending = bytearray() def close(self) -> None: if self.rows != self.height: raise ValueError(f"wrote {self.rows} scanlines, expected {self.height}") self._pending += self._co.flush() if self._pending: self._flush_idat() self._write(chunk(b"IEND", b"")) def iter_filtered_rows(path: Path, sha256: str | None = None): """Yield each filtered scanline of an 8-bit greyscale PNG, without unfiltering it. zlib hands back arbitrary-sized pieces, so scanlines are cut out of a rolling buffer. """ digest = hashlib.sha256() with open(path, "rb") as f: sig = f.read(8) digest.update(sig) if sig != PNG_SIG: raise ValueError(f"{path.name}: not a PNG") dec = zlib.decompressobj() buf, pos, stride, seen = bytearray(), 0, None, 0 height = None while True: head = f.read(8) if len(head) < 8: raise ValueError(f"{path.name}: truncated before IEND") digest.update(head) length, typ = struct.unpack(">I4s", head) body = f.read(length) crc = f.read(4) digest.update(body) digest.update(crc) if len(body) != length or len(crc) != 4: raise ValueError(f"{path.name}: file ends inside a " f"{typ.decode('ascii', 'replace')} chunk") if struct.unpack(">I", crc)[0] != zlib.crc32(typ + body): raise ValueError(f"{path.name}: CRC mismatch in {typ.decode('ascii', 'replace')} chunk") if typ == b"IHDR": if len(body) != 13: raise ValueError(f"{path.name}: IHDR is {len(body)} bytes, expected 13") w, height, depth, color, comp, filt, interlace = struct.unpack(">IIBBBBB", body) if (depth, color, interlace) != (8, 0, 0): raise ValueError(f"{path.name}: expected 8-bit greyscale, non-interlaced, " f"got depth={depth} colour={color} interlace={interlace}") stride = w + 1 elif typ == b"IDAT": if stride is None: raise ValueError(f"{path.name}: IDAT before IHDR") try: buf += dec.decompress(body) except zlib.error as e: raise ValueError(f"{path.name}: damaged compressed data: {e}") from e while len(buf) - pos >= stride: yield bytes(buf[pos:pos + stride]) seen += 1 pos += stride if pos > (1 << 22): del buf[:pos] pos = 0 elif typ == b"IEND": break try: buf += dec.flush() except zlib.error as e: raise ValueError(f"{path.name}: compressed data ends early: {e}") from e while len(buf) - pos >= stride: yield bytes(buf[pos:pos + stride]) seen += 1 pos += stride if seen != height: raise ValueError(f"{path.name}: {seen} scanlines, IHDR says {height}") if len(buf) - pos: raise ValueError(f"{path.name}: {len(buf) - pos} trailing bytes after the last scanline") if sha256 and digest.hexdigest() != sha256: raise ValueError(f"{path.name}: sha256 {digest.hexdigest()}, manifest says {sha256}") def rebuild(manifest: dict, src: Path, out_fh, verify_tiles: bool = True, log=None) -> PngWriter: """Write the full image described by ``manifest`` using tiles found in ``src``.""" grid, w, h = manifest["grid"], manifest["width"], manifest["height"] rows, cols = grid["rows"], grid["cols"] by_cell = {(t["row"], t["col"]): t for t in manifest["tiles"]} writer = PngWriter(out_fh, w, h) for r in range(rows): band = [by_cell[(r, c)] for c in range(cols)] if sum(t["w"] for t in band) != w: raise ValueError(f"row {r} tiles are {sum(t['w'] for t in band)} px wide, image is {w}") readers = [iter_filtered_rows(src / t["file"], t["sha256"] if verify_tiles else None) for t in band] if log: log(f" row {r + 1}/{rows}: {', '.join(t['file'] for t in band[:2])}, ...") for y in range(band[0]["h"]): # a short tile must read as a bad file, not as a bare StopIteration parts = [] for t, it in zip(band, readers): row = next(it, None) if row is None: raise ValueError(f"{t['file']}: ran out of scanlines at row {y} of " f"{band[0]['h']}") parts.append(row) # the filter byte is per scanline, not per tile: every tile in the row carries # the same one, so take it once and drop the copies with the pixel bytes writer.row(parts[0][:1] + b"".join(p[1:] for p in parts)) for it in readers: # force each tile's end-of-stream checks for _ in it: raise ValueError("tile has more scanlines than the manifest says") writer.close() return writer def find_manifest(where: Path) -> Path: if where.is_file(): return where found = sorted(where.glob("*_tiles.json")) if not found: raise SystemExit(f"no *_tiles.json in {where}") if len(found) > 1: raise SystemExit("more than one manifest here; name the one you want:\n " + "\n ".join(p.name for p in found)) return found[0] def main(argv=None) -> int: ap = argparse.ArgumentParser( description="Rebuild a full-resolution reticle PNG from its tiles.", epilog="With no arguments it looks for a single *_tiles.json beside itself.") ap.add_argument("manifest", nargs="?", default=".", type=Path, help="the *_tiles.json file, or a directory holding one (default: .)") ap.add_argument("-o", "--output", type=Path, help="where to write (default: the manifest's name)") ap.add_argument("--skip-tile-checksums", action="store_true", help="do not verify each tile's sha256 while reading it") ap.add_argument("-q", "--quiet", action="store_true") args = ap.parse_args(argv) path = find_manifest(args.manifest) manifest = json.loads(path.read_text()) src = path.parent out = args.output or src / manifest["image"] log = (lambda m: None) if args.quiet else (lambda m: print(m, file=sys.stderr, flush=True)) log(f"{manifest['image']}: {manifest['width']} x {manifest['height']} from " f"{len(manifest['tiles'])} tiles") tmp = out.with_name("." + out.name + ".tmp") try: with open(tmp, "wb") as fh: writer = rebuild(manifest, src, fh, not args.skip_tile_checksums, log) tmp.replace(out) except BaseException: tmp.unlink(missing_ok=True) raise want = manifest.get("sha256", {}) got_png, got_stream = writer.png.hexdigest(), writer.stream.hexdigest() if want.get("png") == got_png: log(f"wrote {out} ({out.stat().st_size:,} bytes), sha256 matches the manifest") return 0 if want.get("filtered_stream") == got_stream: # Every pixel is right - PNG's filtered stream fully determines the image - but this # machine's zlib packs deflate differently, so the container bytes are not the same. log(f"wrote {out} ({out.stat().st_size:,} bytes)") log("The image is exactly right: its uncompressed PNG data stream matches the manifest.") log(f"The file's bytes differ because zlib {zlib.ZLIB_RUNTIME_VERSION} compresses") log("differently from the one that produced the published file (zlib-ng, say).") log("Anything that reads PNGs will see identical pixels; only the sha256 differs.") return 0 print(f"FAILED: rebuilt image does not match the manifest.\n" f" png sha256 {got_png}\n expected {want.get('png')}\n" f" stream sha256 {got_stream}\n expected {want.get('filtered_stream')}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())