Europa/Nuvolari/tool/generate_mock_frames.py

"""Generate the synthetic radar frames that back MockRadarSource.

MockRadarSource is a first-class adapter, not test scaffolding: it is what makes
the app runnable offline, with no credentials, on a fresh clone. Its frames have
to be committed, so this script exists to say where they came from and to let
them be regenerated rather than hand-edited.

The output deliberately mimics the real contract the Python worker will publish
in milestone 8 — same manifest shape, same relative frame URLs, same legend — so
swapping MockRadarSource for DpcRadarSource is a configuration change and not a
different code path.

Pure standard library: no Pillow, no numpy. The PNG writer below is about thirty
lines because a dependency for this would be a dependency for everyone building
the app.

Usage:
    python tool/generate_mock_frames.py
"""

from __future__ import annotations

import json
import math
import pathlib
import struct
import zlib

# Piedmont, matching app/assets/regions/piemonte.json. Kept in sync by
# test/data/radar/mock_radar_source_test.dart, which compares the two.
BBOX = (6.55, 43.95, 9.30, 46.55)

WIDTH = 512
HEIGHT = 512
FRAME_COUNT = 24
FRAME_INTERVAL_MINUTES = 5

# A fixed instant, so regenerating the frames produces identical bytes and the
# committed assets do not churn on every run. 2026-01-15 12:00:00 UTC.
BASE_TIMESTAMP_MS = 1768478400000

OUT_DIR = pathlib.Path(__file__).resolve().parent.parent / "app" / "assets" / "mock"

# Reflectivity colour ramp, in dBZ. Published in the manifest so the app draws
# the legend from the data rather than from a hardcoded copy that could drift.
LEGEND_STOPS: list[tuple[float, str]] = [
    (5, "#4FA3D1"),
    (10, "#2E7DBE"),
    (20, "#35A64A"),
    (30, "#E9D22B"),
    (35, "#EC8B2A"),
    (40, "#D63B28"),
    (45, "#A61E1E"),
    (50, "#B028B0"),
]

MIN_DBZ = LEGEND_STOPS[0][0]


def hex_to_rgb(value: str) -> tuple[int, int, int]:
    value = value.lstrip("#")
    return int(value[0:2], 16), int(value[2:4], 16), int(value[4:6], 16)


def colour_for(dbz: float) -> tuple[int, int, int, int]:
    """Maps a reflectivity value to RGBA, transparent below the lowest stop."""
    if dbz < MIN_DBZ:
        return (0, 0, 0, 0)

    chosen = LEGEND_STOPS[0][1]
    for threshold, colour in LEGEND_STOPS:
        if dbz >= threshold:
            chosen = colour
        else:
            break

    r, g, b = hex_to_rgb(chosen)
    # Fade the weakest returns so the edge of a cell does not look like a wall.
    alpha = 140 if dbz < 10 else 215
    return (r, g, b, alpha)


def write_png(path: pathlib.Path, pixels: list[bytes]) -> None:
    """Writes an 8-bit RGBA PNG. `pixels` is one bytes object per row."""

    def chunk(kind: bytes, data: bytes) -> bytes:
        return (
            struct.pack(">I", len(data))
            + kind
            + data
            + struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
        )

    raw = b"".join(b"\x00" + row for row in pixels)  # filter type 0 per scanline
    png = (
        b"\x89PNG\r\n\x1a\n"
        + chunk(b"IHDR", struct.pack(">IIBBBBB", WIDTH, HEIGHT, 8, 6, 0, 0, 0))
        + chunk(b"IDAT", zlib.compress(raw, 9))
        + chunk(b"IEND", b"")
    )
    path.write_bytes(png)


def reflectivity(x: int, y: int, frame: int) -> float:
    """A precipitation field that drifts west to east and decays as it goes.

    Two overlapping cells rather than one, so the timeline shows something that
    changes shape and not just a disc sliding across the screen.
    """
    progress = frame / (FRAME_COUNT - 1)

    total = 0.0
    cells = (
        # (start x, start y, end x, end y, peak dBZ, radius in px)
        (0.05, 0.35, 0.85, 0.30, 52.0, 95.0),
        (-0.15, 0.62, 0.70, 0.72, 41.0, 70.0),
    )
    for sx, sy, ex, ey, peak, radius in cells:
        cx = (sx + (ex - sx) * progress) * WIDTH
        cy = (sy + (ey - sy) * progress) * HEIGHT
        # Cells grow, peak around the middle of the loop, then weaken.
        intensity = peak * (0.45 + 0.55 * math.sin(math.pi * progress))
        distance = math.hypot(x - cx, y - cy)
        if distance < radius:
            falloff = math.cos(distance / radius * math.pi / 2) ** 2
            total = max(total, intensity * falloff)

    return total


def main() -> None:
    frames_dir = OUT_DIR / "frames"
    frames_dir.mkdir(parents=True, exist_ok=True)

    frames = []
    for index in range(FRAME_COUNT):
        rows = []
        for y in range(HEIGHT):
            row = bytearray()
            for x in range(WIDTH):
                row += bytes(colour_for(reflectivity(x, y, index)))
            rows.append(bytes(row))

        timestamp = BASE_TIMESTAMP_MS + index * FRAME_INTERVAL_MINUTES * 60_000
        name = f"{timestamp}.png"
        write_png(frames_dir / name, rows)
        frames.append({"ts": timestamp, "url": f"frames/{name}"})
        print(f"  {name}")

    manifest = {
        "region": "piemonte",
        "product": "VMI",
        "generatedAt": frames[-1]["ts"],
        "bbox": list(BBOX),
        "crs": "EPSG:3857",
        "frames": frames,
        "legend": {
            "unit": "dBZ",
            "stops": [
                {"value": value, "color": colour} for value, colour in LEGEND_STOPS
            ],
        },
        # Mock data is ours, so it carries no third-party obligation. The live
        # adapter publishes the Radar-DPC credit here instead.
        "attribution": "Dati dimostrativi generati da tool/generate_mock_frames.py",
    }
    (OUT_DIR / "manifest.json").write_text(
        json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
    )

    print(f"\n{FRAME_COUNT} frames + manifest.json written to {OUT_DIR}")


if __name__ == "__main__":
    main()