kyleandClaude Sonnet 5.5
ac4e09bf35
audio_state taps the microphone and the speaker monitor with parec and publishes levels, a speaking/listening/quiet activity, and trigger-locked waveform windows; audio itself stays out of ROS. The activity detector calibrates its noise floor at start-up and absorbs steady noise, so a room at -43 dBFS does not read as "listening" forever, and it ignores the mic while the robot speaks. The face arbitrates the state it shows: sleeping (agent side only, so room sound never wakes a face that means "no agent connected") > speaking > listening > thinking > idle, and draws the mouth from the real samples while fresh scope data exists, auto-scaled so a quiet voice is still visible. Checked: a tone played to the sink shows as `speaking` with the mic ignored. The microphone-to-mouth path with a live voice is not yet confirmed. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs
440 lines
20 KiB
Python
440 lines
20 KiB
Python
"""Oscilloscope-phosphor face: a numpy port of docs/face-prototype.html (flat face).
|
|
|
|
No ROS or hardware imports, so it runs on any host and in tests. The model is the prototype's:
|
|
a 240x240 float32 energy buffer that decays each frame, a beam traced at constant speed that
|
|
deposits energy with bilinear splatting, two box-blur passes for glow, and a tone map tinted
|
|
with the phosphor colour. Only the flat shape is ported so far (head and lowpoly are not).
|
|
|
|
Beyond the prototype: a `sleeping` state for "no agent connected", and `looking` returns to the
|
|
state it interrupted instead of always going to idle.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass
|
|
|
|
import numpy as np
|
|
|
|
W = H = 240
|
|
N = W * H
|
|
|
|
PHOSPHORS = {
|
|
'P31': (0.30, 1.00, 0.45), # bright green, the usual scope tube
|
|
'P1': (0.45, 1.00, 0.25), # yellower green, longer-glowing
|
|
'P7': (0.62, 0.80, 1.00), # blue-white flash, radar screens
|
|
'P3': (1.00, 0.68, 0.18), # amber
|
|
}
|
|
|
|
STATES = ('sleeping', 'idle', 'listening', 'thinking', 'looking', 'speaking')
|
|
LOOK_HOLD_S = 1.2 # how long `looking` holds before returning to the prior state
|
|
SQUINT_S = 0.18 # squint before the shutter flash
|
|
FLASH_AT_S = 0.16 # the shutter bloom lands this long after entering `looking`
|
|
SLEEP_OPEN = 0.07 # eye openness while asleep: a thin line
|
|
SCOPE_FRESH_S = 0.35 # a waveform window older than this is ignored (the synthetic voice returns)
|
|
GAZE_HOLD_S = 2.5 # a glance toward a detected person lasts this long without a refresh
|
|
GAZE_RANGE = (12.0, 9.0) # eye travel in pixels for a gaze of +-1 (display x, y)
|
|
|
|
|
|
def _approach(cur: float, target: float, rate: float, dt: float) -> float:
|
|
return cur + (target - cur) * (1.0 - math.exp(-rate * dt))
|
|
|
|
|
|
# ---------------------------------------------------------------- synthetic voices
|
|
def _hash(i: float, seed: float) -> float:
|
|
x = math.sin(i * 127.1 + seed * 311.7) * 43758.5453
|
|
return x - math.floor(x)
|
|
|
|
|
|
_VOWELS = ((730, 1090), (270, 2290), (300, 870), (530, 1840), (570, 840), (660, 1720))
|
|
_K = np.arange(1, 10, dtype=np.float64)
|
|
|
|
|
|
@dataclass
|
|
class Voice:
|
|
env: float
|
|
f0: float
|
|
w: np.ndarray
|
|
norm: float
|
|
noise: float
|
|
|
|
|
|
def voice(T: float, seed: float, base_f0: float, rate: float) -> Voice:
|
|
u = T * rate
|
|
i = math.floor(u)
|
|
frac = u - i
|
|
pause = _hash(i, seed) < 0.16
|
|
amp = 0.0 if pause else 0.55 + 0.45 * _hash(i + 7, seed)
|
|
env = amp * math.sin(math.pi * frac) ** 0.7
|
|
v0 = _VOWELS[int(math.floor(_hash(i, seed + 3) * len(_VOWELS)))]
|
|
v1 = _VOWELS[int(math.floor(_hash(i + 1, seed + 3) * len(_VOWELS)))]
|
|
m = frac * frac * (3 - 2 * frac)
|
|
f1 = v0[0] + (v1[0] - v0[0]) * m
|
|
f2 = v0[1] + (v1[1] - v0[1]) * m
|
|
f0 = base_f0 * (1 + 0.12 * math.sin(T * 1.7 + seed) + 0.06 * math.sin(T * 4.3))
|
|
f = _K * f0
|
|
d1 = (f - f1) / 160.0
|
|
d2 = (f - f2) / 260.0
|
|
a = np.exp(-d1 * d1) + 0.65 * np.exp(-d2 * d2) + np.where(_K == 1, 0.35, 0.04)
|
|
noise = (0.14 - frac) * 3.2 if frac < 0.14 and not pause else 0.0
|
|
return Voice(env, f0, a, float(a.sum()), noise)
|
|
|
|
|
|
def voice_sample(v: Voice, t: np.ndarray, phase: float) -> np.ndarray:
|
|
"""The voice's waveform at local times t (seconds) inside a trigger-locked window."""
|
|
arg = 2 * math.pi * _K[:, None] * v.f0 * t[None, :] + phase * _K[:, None]
|
|
return (v.w[:, None] * np.sin(arg)).sum(axis=0) / v.norm
|
|
|
|
|
|
# ---------------------------------------------------------------- blur
|
|
def box_blur(a: np.ndarray, r: int) -> np.ndarray:
|
|
"""Separable box blur, zero beyond the edges, via cumulative sums (conserves total energy)."""
|
|
k = 2 * r + 1
|
|
for axis in (1, 0):
|
|
n = a.shape[axis]
|
|
shape = list(a.shape)
|
|
shape[axis] = n + 2 * r + 1
|
|
p = np.zeros(shape, np.float32)
|
|
body = [slice(None), slice(None)]
|
|
tail = [slice(None), slice(None)]
|
|
last = [slice(None), slice(None)]
|
|
lo = [slice(None), slice(None)]
|
|
hi = [slice(None), slice(None)]
|
|
body[axis] = slice(r + 1, r + 1 + n)
|
|
tail[axis] = slice(r + 1 + n, None)
|
|
last[axis] = slice(r + n, r + n + 1)
|
|
lo[axis] = slice(0, n)
|
|
hi[axis] = slice(k, k + n)
|
|
np.cumsum(a, axis=axis, dtype=np.float32, out=p[tuple(body)])
|
|
p[tuple(tail)] = p[tuple(last)]
|
|
a = (p[tuple(hi)] - p[tuple(lo)]) * np.float32(1.0 / k)
|
|
return a
|
|
|
|
|
|
_LUT_SCALE = 512 # tone-map table entries per unit of energy
|
|
_LUT_SIZE = 8 * _LUT_SCALE # covers energy 0..8, far beyond any overdriven spot
|
|
|
|
|
|
def _rgb565_be(rgb: np.ndarray) -> np.ndarray:
|
|
r, g, b = (rgb[..., i].astype(np.uint16) for i in range(3))
|
|
return (((r & 0xF8) << 8) | ((g & 0xFC) << 3) | (b >> 3)).astype('>u2') # big-endian for SPI
|
|
|
|
|
|
class FaceRenderer:
|
|
def __init__(self, phosphor: str = 'P7', style: str = 'phosphor', persist_ms: float = 70.0,
|
|
intensity: float = 1.0, focus: float = 0.35, jitter: float = 0.3,
|
|
state: str = 'idle', now: float = 0.0, seed: int | None = None, rotation: int = 0):
|
|
if style not in ('phosphor', 'vector'):
|
|
raise ValueError(f'unknown style {style!r}')
|
|
if rotation not in (0, 90, 180, 270):
|
|
raise ValueError(f'rotation must be 0, 90, 180 or 270, not {rotation!r}')
|
|
self.rotation = rotation # output turned this many degrees counter-clockwise
|
|
if phosphor not in PHOSPHORS:
|
|
raise ValueError(f'unknown phosphor {phosphor!r}; known: {", ".join(PHOSPHORS)}')
|
|
self.rgb = PHOSPHORS[phosphor]
|
|
self.style = style
|
|
# Clamped, because these arrive as node parameters: focus sets a blur radius, and
|
|
# persistence divides.
|
|
self.persist_ms = min(5000.0, max(5.0, float(persist_ms)))
|
|
self.intensity = min(4.0, max(0.0, float(intensity)))
|
|
self.focus = min(1.0, max(0.0, float(focus)))
|
|
self.jitter = min(5.0, max(0.0, float(jitter)))
|
|
self.rng = np.random.default_rng(seed)
|
|
|
|
self.buf = np.zeros((H, W), np.float32)
|
|
self._xi: list[np.ndarray] = []
|
|
self._yi: list[np.ndarray] = []
|
|
self._w: list[np.ndarray] = []
|
|
self.beam = (120.0, 120.0)
|
|
self.flash = 0.0
|
|
|
|
self.state = state
|
|
self._state_t0 = now
|
|
self._return_state = 'idle'
|
|
self._flash_pending = state == 'looking'
|
|
self.pan = 0.0 # degrees, from the servos: positive pan = robot's left, tilt = looks up
|
|
self.tilt = 0.0
|
|
self._scope = {'mic': (None, -1e9), 'speaker': (None, -1e9)} # newest real waveform per source
|
|
self._scope_peak = {'mic': 0.05, 'speaker': 0.05}
|
|
self._gaze = (0.0, 0.0) # where a person is, as eye direction: +x viewer's right, +y down
|
|
self._gaze_t = -1e9
|
|
self.a = dict(think=0.0, open=SLEEP_OPEN if state == 'sleeping' else 1.0, eye_scale=1.0,
|
|
mouth_speak=0.0, mouth_listen=0.0, smile=1.0, pan_x=0.0, tilt_y=0.0,
|
|
gaze_x=0.0, gaze_y=0.0,
|
|
sleep=1.0 if state == 'sleeping' else 0.0)
|
|
self._next_blink = now + 2.5
|
|
self._blink_t = -1.0
|
|
self._build_lut()
|
|
|
|
def _build_lut(self) -> None:
|
|
"""Tone curve as a table indexed by energy, so a frame is one gather, not exp/clip/stack."""
|
|
v = np.arange(_LUT_SIZE, dtype=np.float64) / _LUT_SCALE
|
|
pr = np.array(self.rgb)
|
|
if self.style == 'vector':
|
|
c = pr + (1 - pr) * 0.35 # pale the tint so lines read as light, not phosphor
|
|
out = np.minimum(1.0, c[None, :] * v[:, None])
|
|
else:
|
|
base = 1 - np.exp(-v * 1.6)
|
|
hot = np.clip((v - 1.1) * 0.35, 0, 0.75) # overdriven spots whiten
|
|
out = np.minimum(1.0, pr[None, :] * base[:, None] + hot[:, None])
|
|
self._lut_rgb = (out * 255).astype(np.uint8)
|
|
self._lut_565 = _rgb565_be(self._lut_rgb)
|
|
|
|
# ------------------------------------------------------------ inputs
|
|
def set_state(self, s: str, now: float) -> None:
|
|
if s not in STATES:
|
|
raise ValueError(f'unknown state {s!r}')
|
|
if s == self.state and s != 'looking':
|
|
return
|
|
if s == 'looking' and self.state != 'looking':
|
|
self._return_state = self.state
|
|
self.state = s
|
|
self._state_t0 = now
|
|
if s == 'looking':
|
|
self._flash_pending = True
|
|
|
|
def set_pose(self, pan_deg: float, tilt_deg: float) -> None:
|
|
self.pan, self.tilt = pan_deg, tilt_deg
|
|
|
|
def set_scope(self, kind: str, samples, now: float) -> None:
|
|
"""The real waveform for the mouth: a trigger-locked window of samples in [-1, 1] from the
|
|
'mic' or the 'speaker'. While it is fresh and the face is listening/speaking, the mouth draws
|
|
it instead of the synthetic voice, auto-scaled so a quiet voice is still visible."""
|
|
if kind not in self._scope:
|
|
raise ValueError(f'unknown scope {kind!r}')
|
|
x = np.asarray(samples, np.float32)
|
|
if len(x) < 8:
|
|
return
|
|
peak = float(np.abs(x).max())
|
|
self._scope_peak[kind] = max(peak, self._scope_peak[kind] * 0.97, 0.02) # fast attack, slow release
|
|
self._scope[kind] = (x, now)
|
|
|
|
def _real_wave(self, kind: str, T: float):
|
|
x, t = self._scope[kind]
|
|
if x is None or T - t > SCOPE_FRESH_S:
|
|
return None
|
|
return x / self._scope_peak[kind]
|
|
|
|
def set_gaze(self, gx: float, gy: float, now: float) -> None:
|
|
"""Glance toward a person: gx, gy in [-1, 1] as eye direction (+x is the viewer's right, +y
|
|
down). Held for GAZE_HOLD_S, then the eyes drift back to centre unless it is refreshed."""
|
|
self._gaze = (min(1.0, max(-1.0, float(gx))), min(1.0, max(-1.0, float(gy))))
|
|
self._gaze_t = now
|
|
|
|
# ------------------------------------------------------------ beam
|
|
def _queue(self, x, y, e) -> None:
|
|
x = np.atleast_1d(np.asarray(x, np.float64))
|
|
y = np.atleast_1d(np.asarray(y, np.float64))
|
|
if self.jitter > 0 and self.style == 'phosphor':
|
|
x = x + (self.rng.random(x.shape) - 0.5) * 2 * self.jitter
|
|
y = y + (self.rng.random(y.shape) - 0.5) * 2 * self.jitter
|
|
xi = np.floor(x).astype(np.int64)
|
|
yi = np.floor(y).astype(np.int64)
|
|
ok = (xi >= 0) & (yi >= 0) & (xi < W - 1) & (yi < H - 1)
|
|
e = np.broadcast_to(np.asarray(e, np.float64), x.shape)
|
|
self._xi.append(xi[ok])
|
|
self._yi.append(yi[ok])
|
|
fx = (x - xi)[ok]
|
|
fy = (y - yi)[ok]
|
|
self._w.append(np.stack([e[ok] * (1 - fx) * (1 - fy), e[ok] * fx * (1 - fy),
|
|
e[ok] * (1 - fx) * fy, e[ok] * fx * fy]))
|
|
|
|
def _flush(self) -> None:
|
|
if not self._xi:
|
|
return
|
|
xi = np.concatenate(self._xi)
|
|
yi = np.concatenate(self._yi)
|
|
w = np.concatenate(self._w, axis=1)
|
|
idx = yi * W + xi
|
|
acc = np.bincount(np.concatenate([idx, idx + 1, idx + W, idx + W + 1]),
|
|
weights=w.reshape(-1), minlength=N)
|
|
self.buf += acc.reshape(H, W).astype(np.float32)
|
|
self._xi.clear()
|
|
self._yi.clear()
|
|
self._w.clear()
|
|
|
|
def _stroke(self, x: np.ndarray, y: np.ndarray, e: float) -> None:
|
|
if self.style == 'vector' and len(x) > 1: # keep line weight even: fill the gaps
|
|
gaps = np.floor(np.hypot(np.diff(x), np.diff(y)) / 0.7).astype(np.int64)
|
|
reps = gaps + 1
|
|
seg = np.repeat(np.arange(len(reps)), reps)
|
|
within = np.arange(len(seg)) - np.repeat(np.cumsum(reps) - reps, reps)
|
|
t = within / np.repeat(reps, reps)
|
|
x, y = (np.append(x[seg] + (x[seg + 1] - x[seg]) * t, x[-1]),
|
|
np.append(y[seg] + (y[seg + 1] - y[seg]) * t, y[-1]))
|
|
self._queue(x, y, e)
|
|
|
|
def _jump_to(self, x: float, y: float, e: float) -> None:
|
|
if self.style == 'phosphor': # a vector display blanks the beam on retrace
|
|
bx, by = self.beam
|
|
n = max(4, round(math.hypot(x - bx, y - by) / 1.5))
|
|
j = np.arange(1, n + 1) / n
|
|
self._queue(bx + (x - bx) * j, by + (y - by) * j, e * 0.045)
|
|
self.beam = (x, y)
|
|
|
|
def _trace(self, fn, n: int, e: float) -> None:
|
|
x, y = fn(np.linspace(0.0, 1.0, n + 1))
|
|
self._jump_to(float(x[0]), float(y[0]), e)
|
|
self._stroke(x, y, e)
|
|
self.beam = (float(x[-1]), float(y[-1]))
|
|
|
|
# ------------------------------------------------------------ the face
|
|
def _draw(self, T: float, dt: float) -> None:
|
|
a = self.a
|
|
sleep = a['sleep']
|
|
e = 0.45 * (dt * 30) * (1.0 if self.style == 'vector' else
|
|
self.intensity * (70.0 / self.persist_ms) ** 0.6)
|
|
if sleep > 0.001: # asleep: dim, with a slow breathing swell
|
|
breath = 0.5 + 0.5 * math.sin(2 * math.pi * 0.22 * T)
|
|
e *= 1.0 - sleep * (0.55 - 0.18 * breath)
|
|
|
|
ex = a['pan_x'] + a['gaze_x'] * GAZE_RANGE[0]
|
|
ey = a['tilt_y'] + a['gaze_y'] * GAZE_RANGE[1] + 8.0 * sleep
|
|
eyes = [(82 + ex, 98 + ey), (158 + ex, 98 + ey)]
|
|
rx = 21 * a['eye_scale']
|
|
ry = 25 * a['eye_scale'] * a['open']
|
|
listening = self.state == 'listening'
|
|
lv = voice(T, 9, 190, 4.2) if listening or a['mouth_listen'] > 0.01 else None
|
|
listen_pulse = 0.12 * lv.env if listening else 0.0
|
|
|
|
# eyes: an ellipse morphing to a rotating 3:2 Lissajous figure while thinking
|
|
k = a['think']
|
|
for idx, (cx, cy) in enumerate(eyes):
|
|
ph = T * 1.6 + idx * math.pi * 0.5
|
|
|
|
def eye(u, cx=cx, cy=cy, ph=ph):
|
|
th = u * 2 * math.pi
|
|
ox, oy = rx * np.cos(th), ry * np.sin(th)
|
|
lx, ly = rx * 0.95 * np.sin(3 * th + ph), ry * 0.95 * np.sin(2 * th)
|
|
return cx + ox + (lx - ox) * k, cy + oy + (ly - oy) * k
|
|
self._trace(eye, 260 + round(160 * k), e * (1 + listen_pulse))
|
|
|
|
# mouth: a smile at rest, the live waveform when speaking or listening
|
|
mx0, mx1 = 70 + ex * 0.6, 170 + ex * 0.6
|
|
my = 166 + ey * 0.6 + 4.0 * sleep
|
|
sv = voice(T, 1, 118, 4.6) if a['mouth_speak'] > 0.01 else None
|
|
win_s, win_l = (2.4 / sv.f0 if sv else 0.0), (3.2 / lv.f0 if lv else 0.0)
|
|
speak_amp = 26 * a['mouth_speak'] * sv.env if sv else 0.0
|
|
listen_amp = 12 * a['mouth_listen'] * lv.env if lv else 0.0
|
|
smile = 7 * a['smile'] * (1 - a['mouth_speak']) * (1 - k * 0.8)
|
|
mouth_w = (mx1 - mx0) * (1 - 0.45 * k) * (1 - 0.5 * sleep)
|
|
m_start = (mx0 + mx1) / 2 - mouth_w / 2
|
|
|
|
real_s = self._real_wave('speaker', T) if self.state == 'speaking' else None
|
|
real_l = self._real_wave('mic', T) if self.state == 'listening' else None
|
|
|
|
def real(wave, u):
|
|
return np.interp(u * (len(wave) - 1), np.arange(len(wave)), wave)
|
|
|
|
def mouth(u):
|
|
x = m_start + u * mouth_w
|
|
c = 2 * u - 1
|
|
y = my + smile * (1 - c * c) - smile * 0.4
|
|
if real_s is not None: # the actual outgoing audio
|
|
y = y - 26 * a['mouth_speak'] * real(real_s, u)
|
|
elif speak_amp > 0.05:
|
|
y = y - speak_amp * voice_sample(sv, u * win_s, 0.0)
|
|
if sv.noise:
|
|
y = y - sv.noise * 5 * a['mouth_speak'] * (np.sin(u * 97 + T * 53) *
|
|
np.sin(u * 61 - T * 31))
|
|
if real_l is not None: # the actual incoming audio, dimmer and smaller
|
|
y = y - 12 * a['mouth_listen'] * real(real_l, u)
|
|
elif listen_amp > 0.05:
|
|
y = y - listen_amp * voice_sample(lv, u * win_l, 0.4)
|
|
return x, y
|
|
self._trace(mouth, 320, e * 0.65 * (0.95 - 0.3 * a['mouth_listen']))
|
|
self._jump_to(eyes[0][0] - rx, eyes[0][1], e)
|
|
|
|
# ------------------------------------------------------------ per-frame update
|
|
def _update(self, T: float, dt: float) -> None:
|
|
a = self.a
|
|
if self.state == 'looking' and T - self._state_t0 > LOOK_HOLD_S:
|
|
self.set_state(self._return_state, T)
|
|
st = self.state
|
|
in_look = st == 'looking'
|
|
tl = T - self._state_t0
|
|
|
|
a['sleep'] = _approach(a['sleep'], 1.0 if st == 'sleeping' else 0.0, 2.5, dt)
|
|
a['think'] = _approach(a['think'], 1.0 if st == 'thinking' else 0.0, 6, dt)
|
|
a['mouth_speak'] = _approach(a['mouth_speak'], 1.0 if st == 'speaking' else 0.0, 10, dt)
|
|
a['mouth_listen'] = _approach(a['mouth_listen'], 1.0 if st == 'listening' else 0.0, 8, dt)
|
|
a['smile'] = _approach(a['smile'], {'idle': 1.0, 'speaking': 0.6, 'sleeping': 0.0}.get(st, 0.25),
|
|
4, dt)
|
|
scale_t = 1.08 if st == 'listening' else 1.0
|
|
if in_look:
|
|
scale_t = 0.92 if tl < SQUINT_S else 1.14
|
|
a['eye_scale'] = _approach(a['eye_scale'], scale_t, 12, dt)
|
|
a['pan_x'] = _approach(a['pan_x'], self.pan * 0.22, 7, dt) # +pan = robot's left = viewer's right
|
|
a['tilt_y'] = _approach(a['tilt_y'], -self.tilt * 0.3, 7, dt) # +tilt = looks up = display up
|
|
glancing = st != 'sleeping' and T - self._gaze_t < GAZE_HOLD_S
|
|
gx, gy = self._gaze if glancing else (0.0, 0.0)
|
|
rate = 6.0 if glancing else 1.5 # follow briskly, drift back slowly
|
|
a['gaze_x'] = _approach(a['gaze_x'], gx, rate, dt)
|
|
a['gaze_y'] = _approach(a['gaze_y'], gy, rate, dt)
|
|
|
|
# blink, and the squint before a photo; none while thinking or asleep
|
|
open_t = SLEEP_OPEN if st == 'sleeping' else 1.0
|
|
if in_look and tl < SQUINT_S:
|
|
open_t = 0.35
|
|
if st not in ('thinking', 'sleeping') and not in_look:
|
|
if self._blink_t < 0 and T >= self._next_blink:
|
|
self._blink_t = T
|
|
if self._blink_t >= 0:
|
|
b = (T - self._blink_t) / 0.17
|
|
if b >= 1:
|
|
self._blink_t = -1.0
|
|
self._next_blink = T + 2.2 + float(self.rng.random()) * 4
|
|
else:
|
|
open_t = max(0.06, abs(1 - 2 * b))
|
|
if self._blink_t >= 0:
|
|
a['open'] = open_t
|
|
else:
|
|
rate = 3.0 if st == 'sleeping' else (6.0 if a['sleep'] > 0.1 else 18.0) # drift shut, wake gently
|
|
a['open'] = _approach(a['open'], open_t, rate, dt)
|
|
|
|
if self._flash_pending and tl > FLASH_AT_S: # shutter: bloom the whole tube once
|
|
if self.style == 'phosphor':
|
|
self.buf += 0.55
|
|
else:
|
|
self.flash = 0.6
|
|
self._flash_pending = False
|
|
self.flash = max(0.0, self.flash - dt * 2.5)
|
|
|
|
# ------------------------------------------------------------ tone map
|
|
def _glow(self, r: int) -> np.ndarray:
|
|
"""Two blur passes, at half resolution (the glow is smooth) and repeated back up."""
|
|
b = self.buf
|
|
half = (b[0::2, 0::2] + b[1::2, 0::2] + b[0::2, 1::2] + b[1::2, 1::2]) * np.float32(0.25)
|
|
rh = max(1, int(r / 2 + 0.5))
|
|
g = box_blur(box_blur(half, rh), rh)
|
|
return np.repeat(np.repeat(g, 2, axis=0), 2, axis=1)
|
|
|
|
def _energy_index(self) -> np.ndarray:
|
|
if self.style == 'vector':
|
|
k = np.float32(2.6 * self.intensity)
|
|
glow = box_blur(self.buf, 1)
|
|
v = (np.minimum(1, self.buf * k) * np.float32(0.9) + np.minimum(1, glow * k) * np.float32(0.25)
|
|
+ np.float32(self.flash * 0.35))
|
|
else:
|
|
r = 1 + round(self.focus * 5)
|
|
v = self.buf * np.float32(1.15 - self.focus * 0.55) + self._glow(r) * np.float32(0.6 + self.focus * 2.2)
|
|
v *= np.float32(_LUT_SCALE)
|
|
np.minimum(v, _LUT_SIZE - 1, out=v)
|
|
idx = v.astype(np.int32)
|
|
return np.rot90(idx, self.rotation // 90) if self.rotation else idx # free: the gather re-packs it
|
|
|
|
def frame(self, T: float, dt: float, fmt: str = 'rgb'):
|
|
"""Advance by dt seconds (clock at T). Returns an (H, W, 3) uint8 array for fmt='rgb', or
|
|
the frame as big-endian RGB565 bytes (what the ST7789 wants) for fmt='rgb565'."""
|
|
dt = min(dt, 0.1)
|
|
if self.style == 'vector':
|
|
self.buf.fill(0)
|
|
else:
|
|
self.buf *= np.float32(math.exp(-(dt * 1000) / self.persist_ms))
|
|
self._update(T, dt)
|
|
self._draw(T, dt)
|
|
self._flush()
|
|
idx = self._energy_index()
|
|
if fmt == 'rgb565':
|
|
return self._lut_565[idx].tobytes()
|
|
return self._lut_rgb[idx]
|