kemal/ashvale-station
1
0
Fork 0
mirror of https://github.com/lynchaos/ashvale-station.git synced 2026-09-12 12:47:49 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
e05667d75b2abde63945d7a7aff8cd44a26634a5
ashvale-station/ashvale/led.py
T
kemal e05667d75b Make the matrix frame rate configurable 
24 fps costs about 11% of one core on a Zero 2 W, measured on the board. That
is a reasonable default for something you look at, but it is a decorative load
sharing a 512 MB machine with the forecaster, so it should be the owner's
choice. server.led_fps is clamped to 4..30.

Particle fall speed now divides by the configured rate rather than the module
constant, so rain falls at the same real-world speed whatever the frame rate,
instead of slowing down when you turn the frame rate down.
2026-08-15 23:04:05 +01:00

637 lines
25 KiB
Python
Raw Blame History

# Copyright 2026 Kemal Yaylali
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""The 8x8 matrix as an instrument you actually want to look at.
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:
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.
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
import math
import time
from typing import Dict, List, Optional, Tuple
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 _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 _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))
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)
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()
# --------------------------------------------------------------------------
# Scenes. Each is a pure function of (time, station snapshot) so it can be
# crossfaded with any other simply by rendering both and blending.
# --------------------------------------------------------------------------
class Scene:
name = "scene"
duration = 12.0
def render(self, cv: Canvas, t: float, s: Dict) -> None:
raise NotImplementedError
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))
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 / max(s.get('_fps', FPS), 1.0)
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:
"""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, fps: float = FPS):
self.station = station
self.cycle_s = float(cycle_s)
self.fps = float(np.clip(fps, 4.0, 30.0))
self.enabled = True
self._stop = asyncio.Event()
self._task = None
self.frame_name = "idle"
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
# ------------------------------------------------------------ state
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 {}
fc = self.station.forecast_bundle or {}
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})
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),
"_fps": self.fps,
}
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
if alerting:
return
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
def _current(self) -> Scene:
return self.alert if self._alerting else self.scenes[self._idx]
async def _run(self) -> None:
period = 1.0 / self.fps
t0 = time.monotonic()
while not self._stop.is_set():
frame_start = time.monotonic()
try:
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())
async def stop(self) -> None:
self._stop.set()
if self._task is not None:
self._task.cancel()
try:
await self._task
except (asyncio.CancelledError, Exception):
pass
try:
self.station.board.clear()
except Exception:
pass
Reference in New Issue View Git Blame Copy Permalink