[1/4] feat: split composed stage chunks into the layer stack

.github/workflows/art-pipeline.yml

@@ -22,5 +22,6 @@ jobs:
       - run: python -m pip install -r requirements-art.txt
       - run: python -m unittest discover -s scripts -p 'test_*.py' -v
       - run: python scripts/creek_layers.py
+      - run: python scripts/split_layers.py --check
       - run: python scripts/collision_check.py
       - run: python scripts/check_docs.py

docs/04-WORLD-AND-LEVELS.md

@@ -55,7 +55,7 @@ Every chunk is a stack of registered layers, all the same size and position, dra
 | 8 | Particles | – | Falling leaves, dust, sparks – spawned by the game, not painted |
 | 9 | Anomaly FX | – | Cuts, stretches, 404 voids applied over the frame by a shader |
 
-The anomaly state replaces only the layers that change (usually 3 and 5, sometimes 2) and keeps the rest pixel-identical.
+The anomaly state replaces only the layers that change (usually 3 and 5, sometimes 2) and keeps the rest pixel-identical. `scripts/split_layers.py` cuts a composed chunk into these layers from ordered mask rules in the region's `layers.json` (rectangles, polygons, colours, another image's alpha). Every pixel goes to exactly one layer, so the stack rebuilds the art exactly; a layer that is identical in both states is stored once. Rusty Creek keeps its own `creek_layers.py`, because its plates are soft and overlap.
 
 ## Scale – the numbers everything follows
 

docs/09-ART-PIPELINE.md

@@ -99,6 +99,7 @@ python scripts/export_atlas.py <ids> [--preview]                  # approved boa
 python scripts/creek_layers.py [--build]                          # Rusty Creek layers: verify / rebuild
 python scripts/collision_check.py [--overlay]                     # collision: schema, reach, overlays
 python scripts/stage_art.py prepare --config art/stages/rusty-creek/stage.json --out tmp/stage-art/next --band 512
+python scripts/split_layers.py art/stages/<region> [--check]      # composed chunk -> z0–z7 layers, exact
 python scripts/navigation.py && python scripts/check_docs.py
 ```
 

scripts/split_layers.py

@@ -0,0 +1,187 @@
+"""Split composed stage chunks into the docs/04 layer stack, and prove the split is exact.
+
+    art/stages/<region>/layers.json   chunks, their composed art per state, ordered mask rules
+      -> every pixel of a chunk goes to exactly ONE layer (the last rule that claims it wins;
+         unclaimed pixels go to the chunk's default layer)
+      -> art/stages/<region>/layers/<chunk>.<layer>.png          when clean and anomaly agree
+         art/stages/<region>/layers/<chunk>.<state>.<layer>.png  when the anomaly changes it
+      -> check: stacking the layers z0..z7 rebuilds each composed state pixel for pixel
+
+Because the layers never overlap, recomposition is exact by construction; the check proves the
+files on disk still say so. A layer missing from a chunk is simply not written.
+
+Mask rules, applied in order (coordinates in chunk pixels, x1/y1 exclusive):
+    {"layer": "00-sky", "rect": [0, 0, 2160, 260]}
+    {"layer": "04-water", "poly": [[0, 600], [2160, 600], [2160, 720], [0, 720]]}
+    {"layer": "05-play", "colour": [70, 160, 150], "tolerance": 40, "rect": [0, 100, 2160, 620]}
+    {"layer": "06-near", "alpha": "images/06-near-rails-left.png"}   # where that image is opaque
+A rule can name "states": ["anomaly"] to apply to one state only.
+
+Usage:
+    python scripts/split_layers.py art/stages/hidden-garden     # write the layer files
+    python scripts/split_layers.py --check                      # CI: every region still rebuilds exactly
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from PIL import Image, ImageChops, ImageDraw
+
+ROOT = Path(__file__).resolve().parents[1]
+LAYERS = ("00-sky", "01-far", "02-mid", "03-rear", "04-water", "05-play", "06-near", "07-foreground")
+
+
+class SplitError(Exception):
+    """The config or the files on disk do not add up; the message says where."""
+
+
+def mask_for(rule: dict, size: tuple[int, int], source: Image.Image, folder: Path) -> Image.Image:
+    """The pixels one rule claims, as an L image (255 = claimed)."""
+    w, h = size
+    area = Image.new("L", size, 0)
+    draw = ImageDraw.Draw(area)
+    if "poly" in rule:
+        draw.polygon([tuple(p) for p in rule["poly"]], fill=255)
+    elif "rect" in rule or "colour" in rule:
+        x0, y0, x1, y1 = rule.get("rect", [0, 0, w, h])
+        draw.rectangle((x0, y0, x1 - 1, y1 - 1), fill=255)
+    if "colour" in rule:
+        r, g, b = rule["colour"]
+        tol = rule.get("tolerance", 30)
+        rgb = source.convert("RGB")
+        near = Image.new("L", size, 0)
+        near.putdata([255 if abs(pr - r) <= tol and abs(pg - g) <= tol and abs(pb - b) <= tol else 0
+                      for pr, pg, pb in rgb.getdata()])
+        area = ImageChops.multiply(area, near)
+    if "alpha" in rule:
+        other = Image.open(folder / rule["alpha"]).convert("RGBA")
+        if other.size != size:
+            raise SplitError(f"{rule['alpha']} is {other.size}, the chunk is {size}")
+        area = other.getchannel("A").point(lambda a: 255 if a else 0)
+    if not any(k in rule for k in ("rect", "poly", "colour", "alpha")):
+        raise SplitError(f"a rule needs rect, poly, colour or alpha: {rule}")
+    return area
+
+
+def assign(chunk: dict, state: str, source: Image.Image, folder: Path) -> Image.Image:
+    """An L image holding each pixel's layer index."""
+    default = chunk.get("default", "02-mid")
+    if default not in LAYERS:
+        raise SplitError(f"unknown default layer {default!r}")
+    owner = Image.new("L", source.size, LAYERS.index(default))
+    for rule in chunk["rules"]:
+        if rule["layer"] not in LAYERS:
+            raise SplitError(f"unknown layer {rule['layer']!r}; use one of {', '.join(LAYERS)}")
+        if state not in rule.get("states", [state]):
+            continue
+        owner.paste(LAYERS.index(rule["layer"]), mask=mask_for(rule, source.size, source, folder))
+    return owner
+
+
+def split(source: Image.Image, owner: Image.Image) -> dict[str, Image.Image]:
+    """Cut the source into one RGBA image per layer that owns any pixel."""
+    out = {}
+    for index, name in enumerate(LAYERS):
+        keep = owner.point(lambda v, i=index: 255 if v == i else 0)
+        if not keep.getbbox():
+            continue
+        layer = Image.new("RGBA", source.size, (0, 0, 0, 0))
+        layer.paste(source, mask=ImageChops.multiply(keep, source.getchannel("A")))
+        out[name] = layer
+    return out
+
+
+def stack(layers: dict[str, Image.Image], size: tuple[int, int]) -> Image.Image:
+    canvas = Image.new("RGBA", size, (0, 0, 0, 0))
+    for name in LAYERS:
+        if name in layers:
+            canvas = Image.alpha_composite(canvas, layers[name])
+    return canvas
+
+
+def same(a: Image.Image, b: Image.Image) -> bool:
+    return a.size == b.size and not ImageChops.difference(a.convert("RGBA"), b.convert("RGBA")).getbbox(alpha_only=False)
+
+
+def files_for(chunk_id: str, states: dict[str, dict[str, Image.Image]]) -> dict[str, Image.Image]:
+    """Layer file names: shared when every state has the identical layer, per state otherwise."""
+    files = {}
+    names = {n for layers in states.values() for n in layers}
+    for name in names:
+        images = [layers.get(name) for layers in states.values()]
+        if all(i is not None for i in images) and all(same(images[0], i) for i in images[1:]):
+            files[f"{chunk_id}.{name}.png"] = images[0]
+        else:
+            for state, layers in states.items():
+                if name in layers:
+                    files[f"{chunk_id}.{state}.{name}.png"] = layers[name]
+    return files
+
+
+def load(folder: Path, chunk_id: str, state: str, files: list[str]) -> dict[str, Image.Image]:
+    """The layers of one chunk state from disk: its own files, else the shared ones."""
+    out = {}
+    for name in LAYERS:
+        own, shared = f"{chunk_id}.{state}.{name}.png", f"{chunk_id}.{name}.png"
+        pick = own if own in files else shared if shared in files else None
+        if pick:
+            out[name] = Image.open(folder / "layers" / pick).convert("RGBA")
+    return out
+
+
+def run(region: Path, check: bool) -> list[str]:
+    config = json.loads((region / "layers.json").read_text(encoding="utf-8"))
+    report = []
+    for chunk_id, chunk in config["chunks"].items():
+        sources = {s: Image.open(region / p).convert("RGBA") for s, p in chunk["states"].items()}
+        sizes = {img.size for img in sources.values()}
+        if len(sizes) != 1:
+            raise SplitError(f"{chunk_id}: its states have different sizes {sizes}")
+        size = sizes.pop()
+        if check:
+            on_disk = sorted(p.name for p in (region / "layers").glob(f"{chunk_id}.*.png"))
+            for state, source in sources.items():
+                if not same(stack(load(region, chunk_id, state, on_disk), size), source):
+                    raise SplitError(f"{chunk_id}.{state}: the committed layers no longer rebuild "
+                                     f"{chunk['states'][state]}; run split_layers.py again")
+            report.append(f"{chunk_id}: {len(on_disk)} layer files rebuild {', '.join(sources)} exactly")
+            continue
+        states = {}
+        for state, source in sources.items():
+            layers = split(source, assign(chunk, state, source, region))
+            if not same(stack(layers, size), source):  # cannot happen unless split() is broken
+                raise SplitError(f"{chunk_id}.{state}: recomposition differs from the source")
+            states[state] = layers
+        files = files_for(chunk_id, states)
+        out = region / "layers"
+        out.mkdir(exist_ok=True)
+        for old in out.glob(f"{chunk_id}.*.png"):
+            if old.name not in files:
+                old.unlink()
+        for name, image in files.items():
+            image.save(out / name, optimize=True)
+        shared = sum(1 for n in files if n.count(".") == 2)
+        report.append(f"{chunk_id}: {len(files)} layer files ({shared} shared by all states), exact")
+    return report
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("regions", nargs="*", type=Path, help="region folders (default: every stage with a layers.json)")
+    parser.add_argument("--check", action="store_true", help="only verify the committed layers rebuild the art")
+    args = parser.parse_args()
+    regions = args.regions or sorted(p.parent for p in (ROOT / "art" / "stages").glob("*/layers.json"))
+    try:
+        for region in regions:
+            for line in run(region, args.check):
+                print(line)
+    except SplitError as error:
+        sys.exit(f"split_layers: {error}")
+
+
+if __name__ == "__main__":
+    main()

scripts/test_split_layers.py

@@ -0,0 +1,108 @@
+"""Offline checks for split_layers.py: exact recomposition, shared layers, --check, and Rusty Creek."""
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from PIL import Image, ImageChops, ImageDraw
+
+import split_layers as sl
+
+ROOT = Path(__file__).resolve().parents[1]
+CREEK = ROOT / 'art' / 'stages' / 'rusty-creek'
+
+
+def scene(anomaly=False):
+    """A 120x60 chunk: sky, a far hill, a green platform, water; the anomaly recolours the platform."""
+    image = Image.new('RGBA', (120, 60), (120, 180, 230, 255))
+    draw = ImageDraw.Draw(image)
+    draw.polygon([(0, 30), (40, 18), (80, 30)], fill=(90, 110, 120, 255))
+    draw.rectangle((20, 34, 99, 39), fill=(170, 60, 40, 255) if anomaly else (60, 160, 70, 255))
+    draw.rectangle((0, 48, 119, 59), fill=(40, 80, 140, 255))
+    return image
+
+
+class SplitLayersTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.region = Path(self.tmp.name)
+        scene().save(self.region / 'clean.png')
+        scene(anomaly=True).save(self.region / 'anomaly.png')
+        self.config = {'chunks': {'t-0-0': {
+            'states': {'clean': 'clean.png', 'anomaly': 'anomaly.png'},
+            'default': '00-sky',
+            'rules': [
+                {'layer': '01-far', 'poly': [[0, 30], [40, 18], [80, 30]]},
+                {'layer': '05-play', 'rect': [20, 34, 100, 40]},
+                {'layer': '04-water', 'rect': [0, 48, 120, 60]},
+            ]}}}
+        self.save()
+
+    def save(self):
+        (self.region / 'layers.json').write_text(json.dumps(self.config), encoding='utf-8')
+
+    def test_split_is_exact_and_shares_unchanged_layers(self):
+        report = sl.run(self.region, check=False)
+        names = sorted(p.name for p in (self.region / 'layers').glob('*.png'))
+        self.assertEqual(names, ['t-0-0.00-sky.png', 't-0-0.01-far.png', 't-0-0.04-water.png',
+                                 't-0-0.anomaly.05-play.png', 't-0-0.clean.05-play.png'])
+        self.assertIn('3 shared by all states', report[0])
+        self.assertTrue(sl.run(self.region, check=True)[0].endswith('rebuild clean, anomaly exactly'))
+
+    def test_layers_never_overlap(self):
+        source = scene()
+        layers = sl.split(source, sl.assign(self.config['chunks']['t-0-0'], 'clean', source, self.region))
+        seen = Image.new('L', source.size, 0)
+        for layer in layers.values():
+            alpha = layer.getchannel('A').point(lambda a: 1 if a else 0)
+            seen = ImageChops.add(seen, alpha)
+        self.assertEqual(seen.getextrema(), (1, 1))
+
+    def test_check_catches_a_stale_layer(self):
+        sl.run(self.region, check=False)
+        scene(anomaly=True).transpose(Image.Transpose.FLIP_LEFT_RIGHT).save(self.region / 'anomaly.png')
+        with self.assertRaisesRegex(sl.SplitError, 'no longer rebuild anomaly.png'):
+            sl.run(self.region, check=True)
+
+    def test_state_only_rules_and_colour_rules(self):
+        self.config['chunks']['t-0-0']['rules'].append(
+            {'layer': '06-near', 'colour': [170, 60, 40], 'tolerance': 5, 'states': ['anomaly']})
+        self.save()
+        sl.run(self.region, check=False)
+        names = {p.name for p in (self.region / 'layers').glob('*.png')}
+        self.assertIn('t-0-0.anomaly.06-near.png', names)
+        self.assertNotIn('t-0-0.anomaly.05-play.png', names)  # the recoloured platform moved to 06
+        self.assertIn('t-0-0.clean.05-play.png', names)
+
+    def test_bad_config_is_named(self):
+        self.config['chunks']['t-0-0']['rules'].append({'layer': '09-fx', 'rect': [0, 0, 1, 1]})
+        self.save()
+        with self.assertRaisesRegex(sl.SplitError, "unknown layer '09-fx'"):
+            sl.run(self.region, check=False)
+
+    def test_rusty_creek_matches_creek_layers(self):
+        """Split the composed left anomaly chunk with the creek's own layer alphas, in its draw order.
+        Every pixel lands in the layer creek_layers.py drew on top there, and the stack rebuilds the art.
+        (Colours can't match the plates: they are semi-transparent, so the composed art is a blend.)"""
+        assembly = json.loads((CREEK / 'assembly.json').read_text(encoding='utf-8'))
+        role_to_layer = {'01-distance': '01-far', '03-bank': '02-mid', '02-supports': '03-rear',
+                         '04-water': '04-water', '05-traversal': '05-play', '06-near-rails': '06-near',
+                         '07-foliage': '07-foreground'}
+        order = assembly['draw_order']
+        plates = {r: Image.open(CREEK / 'images' / f'{r}-left.png').convert('RGBA') for r in order}
+        rules = [{'layer': role_to_layer[r], 'alpha': f'images/{r}-left.png'} for r in order]
+        source = Image.open(CREEK / 'images' / 'after-left.png').convert('RGBA')
+        layers = sl.split(source, sl.assign({'default': '01-far', 'rules': rules}, 'anomaly', source, CREEK))
+        self.assertTrue(sl.same(sl.stack(layers, source.size), source))
+        covered = Image.new('L', source.size, 0)  # pixels a later-drawn plate already claims
+        for role in reversed(order):
+            drawn = plates[role].getchannel('A').point(lambda a: 255 if a else 0)
+            on_top = ImageChops.subtract(drawn, covered)
+            ours = layers.get(role_to_layer[role])
+            ours = ours.getchannel('A').point(lambda a: 255 if a else 0) if ours else Image.new('L', source.size, 0)
+            self.assertIsNone(ImageChops.difference(ours, on_top).getbbox(), role)
+            covered = ImageChops.lighter(covered, drawn)
+
+if __name__ == '__main__':
+    unittest.main()