"""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()