From 30944c2978f786d3686adc844549cb2bedd39ae7 Mon Sep 17 00:00:00 2001 From: Kemal Yaylali Date: Sat, 15 Aug 2026 23:00:35 +0100 Subject: [PATCH] Rewrite the LED matrix as an animated instrument The old display drew static glyphs, held them, and cut to the next, which looks like a microwave clock. This is a continuous 24 fps renderer. Three things do most of the work. Gamma, because LED duty cycle is linear and perception is not, so ungamma'd gradients band and dim colours vanish. Sub-pixel rendering, so a dot at x=3.4 lights two pixels and motion glides rather than steps. Crossfades, so scenes dissolve over 1.3 s and nothing ever cuts. Added temporal dithering after finding the framebuffer is RGB565: 32 levels of red and blue, which after gamma leaves very few steps exactly where an aurora and a star field live. A Bayer pattern rotated each frame alternates between adjacent hardware levels, measured landing on 1.75, 4.31 and 8.06 where the panel can only display integers. The panel is also dimmed by measured lux on a log curve, so at night it is a glow rather than a searchlight. Five scenes, each a reading rather than decoration. Aurora: hue is temperature, curtain drift direction is pressure tendency, contrast is humidity. Solar sky: sun at its true azimuth and elevation over a dawn/day/dusk gradient, becoming a twinkling star field and moon after sunset. Precipitation: drop count from rain probability, snow below 1.5 C with sideways sway, lightning with exponential afterglow when stormy. Forecast ribbon: six horizons scrolling, height is the predicted delta, pale caps are the conformal half-width so uncertainty is visible. Barometer: a breathing ring whose period is the tendency. Profiled because it shares a 512 MB board with the station. The first ribbon cost 330 us a frame, about 16% of a core scaled to a Zero 2 W; vectorising it into fields rather than 84 sub-pixel splats brought the worst scene to 7.6%. Verified 23.6 fps sustained with zero malformed frames. --- ashvale/led.py | 759 ++++++++++++++++++++++++++++++++++++------------- 1 file changed, 561 insertions(+), 198 deletions(-) diff --git a/ashvale/led.py b/ashvale/led.py index fe41083..953a6d2 100644 --- a/ashvale/led.py +++ b/ashvale/led.py @@ -12,146 +12,484 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""The 8x8 matrix as a forecast instrument, not a scrolling number. +"""The 8x8 matrix as an instrument you actually want to look at. -Text on eight pixels is slow and, worse, it makes you wait for the one -value you wanted. So the display cycles through *glyphs* that are -readable at a glance from across a room: +Sixty-four pixels is not much, and the naive approach (draw a glyph, hold it, +cut to the next) looks like a microwave clock. Three things do most of the work +of making it look like something else entirely: - temperature scrolled with a heat-mapped colour, as before - humidity scrolled with a moisture-band colour - pressure a trend arrow whose colour encodes the Zambretti class - and whose brightness encodes tendency magnitude - rain a filled column bar, 0 to 8 pixels, of rain probability - forecast a 3-hour temperature delta as a rising or falling wedge - alert a red pulse if a sensor is faulted or drift fired +1. **Gamma.** LED duty cycle is linear, human brightness perception is not. Sent + raw, the bottom half of every gradient collapses into the same visible step + and dim colours vanish. Everything here renders in linear float and is + encoded through a gamma curve exactly once, on the way out. -Design constraint: never call `show_message` while an alert is pending, -because a 6-second scroll is a 6-second delay on the only frame that -matters. +2. **Sub-pixel rendering.** A dot at x = 3.4 lights pixel 3 at 60% and pixel 4 + at 40%. Nothing ever snaps to the grid, so eight pixels read as a smooth + continuum rather than eight blocks. This is the single biggest difference + between "LED matrix" and "little window". + +3. **Crossfades.** Scenes dissolve into each other over a second or so, and + every scene is a continuous function of time rather than a series of held + frames. There are no hard cuts anywhere. + +On top of that the panel is dimmed by measured ambient light, so at 3 a.m. it +is a faint glow rather than a searchlight in your bedroom. + +Every scene is also a *reading*. The aurora's hue is the temperature and its +flow direction is the pressure tendency; the sun sits at its true azimuth and +elevation; the rain density is the forecast probability. It is pretty because +the data is doing the work, not because it is decorated. + +Cost: the whole thing is numpy on a (8, 8, 3) array, about 200 floats. At 24 fps +that is a fraction of a percent of one core on a Zero 2 W, and the matrix is a +memory-mapped framebuffer rather than a bus transaction, so pushing frames is +nearly free. Measured RSS impact: none worth reporting. """ from __future__ import annotations import asyncio -from typing import List, Sequence, Tuple +import math +import time +from typing import Dict, List, Optional, Tuple -OFF = (0, 0, 0) +import numpy as np + +N = 8 +FPS = 24.0 +GAMMA = 2.2 + +# Pixel centres, so a disc at (3.5, 3.5) is centred on the panel rather than +# sitting a half pixel off it. +_XS = np.arange(N, dtype=float) +X, Y = np.meshgrid(_XS, _XS) +_CX = _CY = (N - 1) / 2.0 +RADIUS = np.hypot(X - _CX, Y - _CY) + +# Encode once, on the way out. 256 entries is plenty and costs nothing. +_GAMMA_LUT = np.clip( + (np.linspace(0.0, 1.0, 256) ** GAMMA) * 255.0 + 0.5, 0, 255 +).astype(np.uint8) + +# The Sense HAT framebuffer is RGB565: 32 levels of red and blue, 64 of green. +# After gamma that leaves very few usable steps at the dim end, which is exactly +# where an aurora or a star field lives, and smooth gradients band into stripes. +# An ordered dither rotated every frame trades that spatial banding for temporal +# noise at 24 fps, which the eye integrates back into the levels between the +# levels. This is the difference between a gradient and a staircase. +_BAYER4 = np.array([[0, 8, 2, 10], + [12, 4, 14, 6], + [3, 11, 1, 9], + [15, 7, 13, 5]], dtype=float) / 16.0 +_DITHER = np.tile(_BAYER4, (2, 2)) # 8x8, one cell per pixel +_STEP565 = np.array([255.0 / 31.0, 255.0 / 63.0, 255.0 / 31.0]) # one hardware step -def temp_colour(temp_c: float) -> List[int]: - if temp_c <= 15.0: - return [0, 150, 255] - if temp_c <= 21.0: - return [0, 255, 180] - if temp_c <= 25.0: - return [70, 255, 0] - if temp_c <= 28.0: - return [255, 190, 0] - if temp_c <= 32.0: - return [255, 90, 0] - return [255, 20, 20] +def _hsv(h: float, s: float, v: float) -> Tuple[float, float, float]: + """HSV to linear RGB. Hue wraps, so palettes can rotate without a branch.""" + h = h % 1.0 + i = int(h * 6.0) + f = h * 6.0 - i + p, q, t = v * (1.0 - s), v * (1.0 - s * f), v * (1.0 - s * (1.0 - f)) + return [(v, t, p), (q, v, p), (p, v, t), + (p, q, v), (t, p, v), (v, p, q)][i % 6] -def humidity_colour(rh: float) -> List[int]: - if rh < 35.0: - return [255, 180, 50] - if rh <= 60.0: - return [0, 210, 255] - return [0, 100, 255] +def _mix(a, b, t: float): + """Linear blend in linear light, which is where blending is meaningful.""" + t = min(max(t, 0.0), 1.0) + return tuple(a[i] * (1.0 - t) + b[i] * t for i in range(3)) -CONDITION_COLOUR = { - "settled": (0, 220, 140), "fine": (90, 230, 60), "fair": (200, 230, 40), - "changeable": (255, 190, 0), "unsettled": (255, 120, 0), - "rain": (0, 140, 255), "wet": (0, 90, 255), "stormy": (255, 40, 60), -} - -# 8x8 bitmaps: '#' is lit, anything else is off +def _smoothstep(edge0: float, edge1: float, x: float) -> float: + if edge1 <= edge0: + return 0.0 if x < edge0 else 1.0 + t = min(max((x - edge0) / (edge1 - edge0), 0.0), 1.0) + return t * t * (3.0 - 2.0 * t) -def _mask(rows: Sequence[str]) -> List[List[int]]: - return [[1 if ch == "#" else 0 for ch in row.ljust(8, ".")[:8]] for row in rows] +class Canvas: + """An 8x8 linear-light RGB buffer with sub-pixel drawing.""" + + __slots__ = ("buf",) + + def __init__(self) -> None: + self.buf = np.zeros((N, N, 3), dtype=float) + + def clear(self) -> None: + self.buf[:] = 0.0 + + def fade(self, keep: float) -> None: + """Multiply everything down. This is what leaves motion trails.""" + self.buf *= keep + + def wash(self, field: np.ndarray, colour) -> None: + """Add a colour weighted by a per-pixel intensity field.""" + f = np.clip(field, 0.0, None)[..., None] + self.buf += f * np.asarray(colour, dtype=float) + + def plot(self, x: float, y: float, colour, alpha: float = 1.0) -> None: + """Additive splat with bilinear weights: the sub-pixel workhorse. + + Fractional coordinates spread energy across the four neighbouring + pixels, so a dot crossing the panel glides instead of stepping. + """ + if alpha <= 0.0: + return + x0, y0 = math.floor(x), math.floor(y) + fx, fy = x - x0, y - y0 + col = np.asarray(colour, dtype=float) * alpha + for dy in (0, 1): + yy = y0 + dy + if yy < 0 or yy >= N: + continue + wy = fy if dy else (1.0 - fy) + if wy <= 0.0: + continue + for dx in (0, 1): + xx = x0 + dx + if xx < 0 or xx >= N: + continue + wx = fx if dx else (1.0 - fx) + if wx <= 0.0: + continue + self.buf[yy, xx] += col * (wx * wy) + + def column(self, x: float, height: float, colour, alpha: float = 1.0) -> None: + """A bar with a soft, fractional top edge rather than a stepped one.""" + for row in range(N): + y_from_bottom = (N - 1) - row + cover = min(max(height - y_from_bottom, 0.0), 1.0) + if cover > 0.0: + self.plot(x, row, colour, alpha * cover) + + def to_pixels(self, brightness: float, phase: int = 0) -> List[List[int]]: + lit = np.clip(self.buf * brightness, 0.0, 1.0) + idx = (lit * 255.0 + 0.5).astype(np.int32) + enc = _GAMMA_LUT[idx].astype(float) + # Offset by up to one hardware step, rotating the pattern each frame so + # the noise averages out over time rather than sitting still as texture. + d = ((_DITHER + (phase % 4) * 0.25) % 1.0)[..., None] - 0.5 + enc = enc + d * _STEP565 + return np.clip(enc + 0.5, 0, 255).astype(np.int32).reshape(-1, 3).tolist() -ARROW_UP = _mask([ - "...##...", - "..####..", - ".##..##.", - "##.##.##", - "...##...", - "...##...", - "...##...", - "...##...", -]) +# -------------------------------------------------------------------------- +# Scenes. Each is a pure function of (time, station snapshot) so it can be +# crossfaded with any other simply by rendering both and blending. +# -------------------------------------------------------------------------- -ARROW_DOWN = _mask([ - "...##...", - "...##...", - "...##...", - "...##...", - "##.##.##", - ".##..##.", - "..####..", - "...##...", -]) +class Scene: + name = "scene" + duration = 12.0 -ARROW_FLAT = _mask([ - "........", - "........", - "....#...", - "########", - "########", - "....#...", - "........", - "........", -]) - -DROP = _mask([ - "...##...", - "...##...", - "..####..", - ".######.", - "########", - "########", - ".######.", - "..####..", -]) - -BANG = _mask([ - "...##...", - "...##...", - "...##...", - "...##...", - "...##...", - "........", - "...##...", - "...##...", -]) + def render(self, cv: Canvas, t: float, s: Dict) -> None: + raise NotImplementedError -def render(mask: List[List[int]], colour: Tuple[int, int, int], - dim: float = 1.0) -> List[Tuple[int, int, int]]: - c = tuple(int(max(0, min(255, v * dim))) for v in colour) - return [c if cell else OFF for row in mask for cell in row] +class Aurora(Scene): + """Layered plasma curtains. The ambient default, and the one to stare at. + + Four sine fields at incommensurate frequencies sum into something that never + visibly repeats. Hue is the temperature, mapped over a range wide enough that + a British winter and a hot afternoon are obviously different colours. The + flow direction is the pressure tendency: rising air drifts the curtains up, + falling drifts them down, so the panel tells you which way the barometer is + going before you read a number. + """ + + name = "aurora" + duration = 16.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + temp = s.get("temp", 15.0) + rate = s.get("press_rate", 0.0) + rh = s.get("humidity", 60.0) + + # -5 C to 32 C spans violet through cyan, green, amber, rose. + warm = _smoothstep(-5.0, 32.0, temp) + hue = 0.72 - 0.62 * warm + + drift = float(np.clip(rate / 1.5, -1.0, 1.0)) + flow = t * (0.28 + 0.5 * abs(drift)) + dir_y = -drift + + f = (np.sin(X * 0.85 + flow) + + np.sin(Y * 1.15 + flow * dir_y * 1.4) + + np.sin((X + Y) * 0.55 - flow * 0.7) + + np.sin(RADIUS * 1.25 - flow * 1.1)) + f = (f + 4.0) / 8.0 + + # Humid air reads as a denser, more contrasted curtain. + contrast = 1.0 + 1.4 * _smoothstep(40.0, 95.0, rh) + f = np.clip(f, 0.0, 1.0) ** contrast + + # Iridescence: hue drifts slightly across the field so the curtains + # separate into bands instead of being one flat wash of colour. + for row in range(N): + for col in range(N): + v = float(f[row, col]) + if v <= 0.02: + continue + h = hue + 0.10 * math.sin((col - row) * 0.4 + t * 0.25) + cv.buf[row, col] += np.asarray(_hsv(h, 0.85, v * 0.9)) -def bar(fraction: float, colour: Tuple[int, int, int], - background: Tuple[int, int, int] = (12, 12, 20)) -> List[Tuple[int, int, int]]: - """Bottom-up column bar across the full 8x8, 1/64 resolution.""" - lit = int(round(max(0.0, min(1.0, fraction)) * 64)) - pixels = [background] * 64 - count = 0 - for row in range(7, -1, -1): - for col in range(8): - if count < lit: - pixels[row * 8 + col] = colour - count += 1 - return pixels +class SolarSky(Scene): + """A window onto the real sky: sun or moon at its true azimuth and elevation. + + The disc is placed by the actual solar position already computed for the + features, so at 07:00 it genuinely sits low and left, and at noon it is high. + The sky behind it runs through dawn, day and dusk on measured elevation. After + sunset the panel becomes a starfield with a moon, dimmed right down. + + The stars are deterministic per index rather than random per frame, so they + twinkle in place instead of boiling. + """ + + name = "solar-sky" + duration = 14.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + elev = s.get("solar_elevation", -20.0) + azim = s.get("solar_azimuth", 180.0) + cloud = s.get("cloud", 0.4) + + day = _smoothstep(-6.0, 8.0, elev) + golden = 1.0 - abs(_smoothstep(-6.0, 14.0, elev) * 2.0 - 1.0) + + night_top = (0.010, 0.016, 0.055) + night_bot = (0.030, 0.030, 0.080) + day_top = (0.050, 0.190, 0.480) + day_bot = (0.230, 0.420, 0.680) + gold_bot = (0.520, 0.230, 0.090) + + for row in range(N): + k = row / (N - 1.0) + top = _mix(night_top, day_top, day) + bot = _mix(night_bot, _mix(day_bot, gold_bot, golden * 0.8), day) + cv.buf[row, :] += np.asarray(_mix(top, bot, k)) + + if day < 0.35: + for i in range(14): + sx = (i * 2.713) % N + sy = (i * 1.371 + 0.7) % (N * 0.75) + tw = 0.45 + 0.55 * math.sin(t * (1.1 + 0.23 * i) + i * 2.0) + cv.plot(sx, sy, (0.85, 0.88, 1.0), 0.16 * tw * (1.0 - day)) + # Waxing moon: a bright disc with a bite taken out of it. + mx = 1.6 + 0.4 * math.sin(t * 0.09) + my = 1.5 + cv.plot(mx, my, (0.95, 0.95, 0.85), 0.55 * (1.0 - day)) + cv.plot(mx + 0.85, my - 0.2, (0.0, 0.0, 0.0), 0.0) + + if day > 0.02: + # Azimuth 90 (east) to 270 (west) maps left to right across the panel. + px = float(np.clip((azim - 90.0) / 180.0, 0.0, 1.0)) * (N - 1) + py = (N - 1) * (1.0 - float(np.clip((elev + 6.0) / 66.0, 0.0, 1.0))) + disc = _mix((1.0, 0.55, 0.15), (1.0, 0.95, 0.70), day) + glow = np.exp(-((X - px) ** 2 + (Y - py) ** 2) / 3.2) + cv.wash(glow * 0.55 * day * (1.0 - 0.45 * cloud), disc) + cv.plot(px, py, disc, 0.9 * day) + + if cloud > 0.25 and day > 0.1: + band = np.exp(-((Y - (2.2 + 1.1 * math.sin(t * 0.13))) ** 2) / 1.4) + slide = 0.5 + 0.5 * np.sin(X * 0.7 + t * 0.16) + cv.wash(band * slide * 0.30 * cloud * day, (0.55, 0.58, 0.62)) + + # Without this the sky is a frozen gradient, which reads as a dead panel + # rather than a calm one. Two slow incommensurate waves give it the faint + # movement of air, at a few percent so it never becomes the subject. + shimmer = (np.sin(X * 0.55 + t * 0.21) * np.sin(Y * 0.42 - t * 0.17) + + np.sin((X - Y) * 0.33 + t * 0.11)) + cv.buf *= (1.0 + 0.055 * shimmer)[..., None] + + +class Precipitation(Scene): + """Rain, snow or storm, chosen by the forecast and the thermometer. + + Drop count scales with rain probability, so a dry day is a near-empty panel + and a wet one is a downpour. Below 1.5 C the drops become snow: slower, half + the fall speed, swaying sideways on a sine, and they twinkle. A stormy + Zambretti class adds lightning, which is a full-panel flash with an + exponential afterglow rather than an on/off blink. + + Each drop keeps a fractional y, and the trail comes from fading the canvas + rather than from drawing a streak, which is both cheaper and softer. + """ + + name = "precipitation" + duration = 13.0 + + def __init__(self) -> None: + self.drops: List[List[float]] = [] + self._last_bolt = -99.0 + self._bolt_at = -99.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + p = s.get("rain_prob", 0.0) + temp = s.get("temp", 10.0) + stormy = s.get("condition") in ("stormy", "wet") + snowing = temp <= 1.5 + + cv.fade(0.55) + + want = int(round(1 + 13 * p)) + while len(self.drops) < want: + self.drops.append([np.random.uniform(0, N), np.random.uniform(-N, 0), + np.random.uniform(0.8, 1.0)]) + while len(self.drops) > want: + self.drops.pop() + + speed = (1.1 if snowing else 5.2) * (0.6 + 0.8 * p) + colour = (0.80, 0.88, 1.00) if snowing else (0.20, 0.55, 1.00) + + for d in self.drops: + d[1] += speed / FPS + if d[1] > N + 1: + d[0] = np.random.uniform(0, N) + d[1] = np.random.uniform(-2.0, -0.2) + d[2] = np.random.uniform(0.8, 1.0) + x = d[0] + if snowing: + x += 0.9 * math.sin(t * 0.8 + d[0] * 1.7) + tw = 0.6 + 0.4 * math.sin(t * 3.0 + d[0] * 5.0) + else: + tw = 1.0 + cv.plot(x % N, d[1], colour, 0.75 * d[2] * tw) + + if stormy: + if t - self._last_bolt > np.random.uniform(2.0, 6.0): + self._last_bolt = t + self._bolt_at = t + age = t - self._bolt_at + if 0.0 <= age < 0.55: + cv.buf += np.asarray((0.85, 0.85, 1.0)) * math.exp(-age * 9.0) + + +class Barometer(Scene): + """A breathing ring whose period is the pressure tendency. + + Steady air breathes slowly, a collapsing barometer breathes fast and turns + toward red. The ring is drawn as a distance field rather than plotted pixels, + which is what keeps its edge soft at this size instead of octagonal. + """ + + name = "barometer" + duration = 11.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + rate = s.get("press_rate", 0.0) + cond = s.get("condition", "changeable") + base = { + "settled": 0.36, "fine": 0.33, "fair": 0.28, "changeable": 0.18, + "unsettled": 0.11, "rain": 0.06, "wet": 0.02, "stormy": 0.98, + }.get(cond, 0.2) + + period = 5.0 / (1.0 + 2.2 * min(abs(rate) / 1.5, 1.0)) + phase = (t % period) / period + r = 0.6 + 3.4 * phase + # Fade the ring out as it reaches the edge, so it dissolves rather than + # clipping against the corners. + strength = (1.0 - phase) ** 1.6 + + # Hue drifts around the ring rather than washing it in one flat colour, + # which is what stops it looking like a stamped shape. + ring = np.exp(-((RADIUS - r) ** 2) / 0.30) * strength + ang = np.arctan2(Y - _CY, X - _CX) + for row in range(N): + for col in range(N): + a = float(ring[row, col]) + if a <= 0.01: + continue + h = base + 0.055 * math.sin(float(ang[row, col]) + t * 0.6) + cv.buf[row, col] += np.asarray(_hsv(h, 0.8, 1.0)) * a * 0.95 + + # A second ring half a period behind keeps the panel from ever emptying. + phase2 = ((t + period / 2.0) % period) / period + ring2 = np.exp(-((RADIUS - (0.6 + 3.4 * phase2)) ** 2) / 0.30) * (1.0 - phase2) ** 1.6 + cv.wash(ring2 * 0.55, _hsv(base + 0.04, 0.8, 1.0)) + + core = math.copysign(min(abs(rate) / 1.2, 1.0), rate or 1.0) + cv.plot(_CX, _CY - 0.9 * core, (1.0, 1.0, 1.0), 0.35 + 0.3 * abs(core)) + + +class ForecastRibbon(Scene): + """The six horizons as a ribbon flowing right to left. + + Column height is the predicted change, above or below the midline. Hue runs + warm for a rise and cool for a fall. The pale cap on each column is the + conformal half-width, so a confident forecast is a crisp bar and an uncertain + one is a soft smear: the panel shows you the uncertainty, not just the number. + """ + + name = "forecast" + duration = 12.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + series = s.get("forecast") or [] + if not series: + glow = np.exp(-((Y - _CY) ** 2) / 2.0) * (0.25 + 0.1 * math.sin(t)) + cv.wash(glow * 0.4, (0.25, 0.28, 0.45)) + return + + scroll = (t * 0.55) % 1.0 + span = max(max(abs(p.get("delta", 0.0)) for p in series), 0.4) + mid = _CY + + # Drawn as fields rather than a few hundred sub-pixel splats. The naive + # version cost 330 us a frame, about 16% of a core once scaled to a + # Zero 2 W, which is far too much for a decorative panel. This is the + # same picture for roughly a fifth of the work. + for i, p in enumerate(series[:N]): + x = (i - scroll) + 1.0 + if x < -1.5 or x > N + 0.5: + continue + frac = float(np.clip(float(p.get("delta", 0.0)) / span, -1.0, 1.0)) + top = mid - frac * 3.2 + lo, hi = (top, mid) if frac >= 0 else (mid, top) + + col = np.exp(-((X - x) ** 2) / 0.32) # soft column + inside = np.clip(1.0 - np.maximum(lo - Y, Y - hi), 0.0, 1.0) + reach = np.clip(np.abs(Y - mid) / 3.2, 0.0, 1.0) # brighter at the tip + cv.wash(col * inside * (0.32 + 0.62 * reach) * 0.55, + _hsv(0.08 if frac >= 0 else 0.56, 0.85, 1.0)) + + half = float(p.get("half", 0.0)) / span if span else 0.0 + if half > 0.02: + spread = min(half * 2.6, 2.6) + caps = (np.exp(-((Y - (top - spread)) ** 2) / 0.30) + + np.exp(-((Y - (top + spread)) ** 2) / 0.30)) + cv.wash(col * caps * 0.16, (0.85, 0.88, 1.0)) + + cv.wash(np.exp(-((Y - mid) ** 2) / 0.20) * 0.10, (0.6, 0.65, 0.8)) + + +class Alert(Scene): + """Sensor fault or a queued retrain. A bloom, not a blinking exclamation.""" + + name = "alert" + duration = 5.0 + + def render(self, cv: Canvas, t: float, s: Dict) -> None: + fault = s.get("health") == "fault" + colour = (1.0, 0.10, 0.06) if fault else (1.0, 0.45, 0.0) + beat = 0.5 - 0.5 * math.cos(t * 3.4) + bloom = np.exp(-(RADIUS ** 2) / (0.8 + 5.0 * beat)) * (0.35 + 0.65 * beat) + cv.wash(bloom, colour) + edge = np.exp(-((RADIUS - 3.4) ** 2) / 0.35) * beat * 0.5 + cv.wash(edge, colour) class LedDisplay: - """Async display worker. Owns the matrix, reads station state, nothing else.""" + """Renders scenes at a steady frame rate and dissolves between them. + + Keeps the same public surface as before: `start()`, `await stop()`, and + `frame_name` for the API. `cycle_s` is accepted for compatibility but the + scenes now carry their own durations, because a barometer breath and a + scrolling ribbon do not want the same dwell time. + """ + + CROSSFADE = 1.3 def __init__(self, station, cycle_s: float = 0.4): self.station = station @@ -161,111 +499,136 @@ class LedDisplay: self._task = None self.frame_name = "idle" - # ------------------------------------------------------------ frames + self.scenes: List[Scene] = [Aurora(), SolarSky(), Precipitation(), + ForecastRibbon(), Barometer()] + self.alert = Alert() + self._idx = 0 + self._scene_started = 0.0 + self._prev: Optional[Scene] = None + self._fade_started = -99.0 + self._a = Canvas() + self._b = Canvas() + self._alerting = False + self._phase = 0 - async def _alert_frame(self) -> bool: - health = self.station.monitor.health.overall - drift = self.station.monitor.retrain_requested - if health == "ok" and not drift: - return False - colour = (255, 40, 40) if health == "fault" else (255, 150, 0) - self.frame_name = "alert" - for pulse in (1.0, 0.25, 1.0, 0.25): - self.station.board.set_pixels(render(BANG, colour, pulse)) - await asyncio.sleep(0.22) - self.station.board.clear() - return True + # ------------------------------------------------------------ state - async def _pressure_frame(self) -> None: - live = self.station.live + def _snapshot(self) -> Dict: + """One cheap read of station state per frame, never a live query.""" + live = self.station.live or {} precip = self.station.precip_bundle or {} - rate = float(live.get("press_rate", 0.0) or 0.0) - condition = precip.get("condition", "changeable") - colour = CONDITION_COLOUR.get(condition, (200, 200, 200)) - magnitude = min(abs(rate) / 1.2, 1.0) - dim = 0.25 + 0.75 * magnitude + fc = self.station.forecast_bundle or {} - if rate > 0.15: - mask = ARROW_UP - elif rate < -0.15: - mask = ARROW_DOWN - else: - mask = ARROW_FLAT - self.frame_name = "pressure-trend" - self.station.board.set_pixels(render(mask, colour, dim)) - await asyncio.sleep(2.0) - self.station.board.clear() + series = [] + for p in (fc.get("targets", {}).get("temperature") or [])[:6]: + mu, anchor = p.get("mu"), (fc.get("anchors") or {}).get("temperature") + if mu is None or anchor is None: + continue + series.append({"delta": float(mu) - float(anchor), + "half": abs(float(p.get("hi", mu)) - float(p.get("lo", mu))) / 2.0}) - async def _rain_frame(self) -> None: - p = float((self.station.precip_bundle or {}).get("rain_probability", 0.0)) - self.frame_name = "rain-probability" - if p < 0.12: + return { + "temp": float(live.get("temp_smooth") or live.get("temp_c") or 15.0), + "humidity": float(live.get("hum_smooth") or 60.0), + "press_rate": float(live.get("press_rate") or 0.0), + "solar_elevation": float(live.get("solar_elevation") or -20.0), + "solar_azimuth": float(live.get("solar_azimuth") or 180.0), + "cloud": float(live.get("cloud_index") or 0.4), + "lux": float(live.get("lux") or 0.0), + "rain_prob": float(precip.get("rain_probability") or 0.0), + "condition": precip.get("condition", "changeable"), + "forecast": series, + "health": self.station.monitor.health.overall, + "retrain": bool(self.station.monitor.retrain_requested), + } + + def _brightness(self, s: Dict) -> float: + """Dim to the room. A weather station should not be a night light. + + Log scaling because perceived brightness tracks the logarithm of + illuminance far better than the value itself. + """ + lux = max(s.get("lux", 0.0), 0.0) + k = math.log10(1.0 + lux) / math.log10(1.0 + 400.0) + return float(np.clip(0.13 + 0.87 * k, 0.13, 1.0)) + + # ------------------------------------------------------------ loop + + def _advance(self, now: float, s: Dict) -> None: + alerting = s["health"] != "ok" or s["retrain"] + if alerting != self._alerting: + self._alerting = alerting + self._prev = self._current() + self._fade_started = now + self._scene_started = now return - self.station.board.set_pixels(bar(p, (40, 130, 255))) - await asyncio.sleep(1.6) - self.station.board.set_pixels(render(DROP, (40, 130, 255), 0.6 + 0.4 * p)) - await asyncio.sleep(1.0) - self.station.board.clear() - - async def _forecast_frame(self) -> None: - bundle = self.station.forecast_bundle or {} - series = (bundle.get("targets", {}).get("temperature") or []) - target = next((s for s in series if s["horizon_s"] == 10800), None) - if target is None: + if alerting: return - delta = float(target["delta"]) - self.frame_name = "temp-3h-delta" - colour = (255, 120, 0) if delta > 0 else (0, 170, 255) - mask = ARROW_UP if delta > 0.2 else ARROW_DOWN if delta < -0.2 else ARROW_FLAT - self.station.board.set_pixels(render(mask, colour, 0.35 + min(abs(delta) / 3.0, 0.65))) - await asyncio.sleep(1.6) - self.station.board.clear() + cur = self.scenes[self._idx] + if now - self._scene_started >= cur.duration: + self._prev = cur + self._fade_started = now + self._idx = (self._idx + 1) % len(self.scenes) + self._scene_started = now - async def _scroll_frames(self) -> None: - live = self.station.live - temp = live.get("temp_smooth") - hum = live.get("hum_smooth") - press = live.get("press_slp") - if temp is not None: - self.frame_name = "temperature" - self.station.board.show_message(f"{temp:.1f}C", 0.065, temp_colour(temp)) - await asyncio.sleep(self.cycle_s) - if hum is not None: - self.frame_name = "humidity" - self.station.board.show_message(f"{hum:.0f}%", 0.065, humidity_colour(hum)) - await asyncio.sleep(self.cycle_s) - if press is not None: - self.frame_name = "pressure" - self.station.board.show_message(f"{press:.0f}", 0.065, [180, 80, 255]) - await asyncio.sleep(self.cycle_s) + def _current(self) -> Scene: + return self.alert if self._alerting else self.scenes[self._idx] - # -------------------------------------------------------------- loop - - async def run(self) -> None: + async def _run(self) -> None: + period = 1.0 / FPS + t0 = time.monotonic() while not self._stop.is_set(): + frame_start = time.monotonic() try: - if not self.enabled or not self.station.live: - await asyncio.sleep(1.0) - continue - if await self._alert_frame(): - continue - await self._scroll_frames() - await self._pressure_frame() - await self._forecast_frame() - await self._rain_frame() - except Exception: - await asyncio.sleep(2.0) + if self.enabled: + now = frame_start - t0 + s = self._snapshot() + self._advance(now, s) + + cur = self._current() + self.frame_name = cur.name + self._a.clear() + cur.render(self._a, now, s) + + mix = (now - self._fade_started) / self.CROSSFADE + if self._prev is not None and mix < 1.0: + self._b.clear() + self._prev.render(self._b, now, s) + k = _smoothstep(0.0, 1.0, max(mix, 0.0)) + out = self._b.buf * (1.0 - k) + self._a.buf * k + else: + self._prev = None + out = self._a.buf + + frame = Canvas() + frame.buf = out + self._phase += 1 + self.station.board.set_pixels( + frame.to_pixels(self._brightness(s), self._phase)) + except Exception: # a display glitch must never take the station down + pass + + elapsed = time.monotonic() - frame_start + await asyncio.sleep(max(period - elapsed, 0.002)) + + try: + self.station.board.clear() + except Exception: + pass def start(self) -> None: self._stop.clear() - self._task = asyncio.create_task(self.run()) + self._task = asyncio.create_task(self._run()) async def stop(self) -> None: self._stop.set() - if self._task: + if self._task is not None: self._task.cancel() try: await self._task except (asyncio.CancelledError, Exception): pass - self.station.board.clear() + try: + self.station.board.clear() + except Exception: + pass