SubScribe Field OS , A Production‑Ready Local‑First Voice Platform for Construction Paperwork
Construction sites generate a constant stream of forms, check‑lists, and safety logs. A voice‑first interface that runs entirely on‑device removes the need for unreliable cellular links and protects sensitive data. The following masterclass walks through the hardened architecture that keeps the system inside an 8 GB RAM envelope, eliminates hidden race conditions, and guarantees deterministic behavior on low‑cost NUCs or Raspberry Pi 4 units.
1. Memory Discipline Starts at the Edge
The OS itself occupies roughly 600 MB on a stripped‑down Linux image. Adding the Python interpreter and the SQLite shared library pushes the baseline to about 780 MB. A quantised ONNX speech‑to‑text model consumes another 45 MB when loaded in a worker process. That leaves just over 6 GB for the rest of the stack.
A common mistake is to treat the Python deque(maxlen=…) as a free‑lunch memory guard. The container silently drops the oldest element once the limit is reached, but it does not stop the underlying byte buffers from being allocated. When a noisy site‑wide PA system floods the microphone with ten times the expected frame rate, the queue can grow far beyond the intended budget before the maxlen policy kicks in. The hardened pattern below adds an explicit byte counter that aborts the burst before any allocation exceeds the hard cap.
# Bounded audio queue with explicit byte accounting
import collections
import asyncio
MAX_AUDIO_FRAMES = 50 # keep at most 50 frames
CHUNK_BYTES = 4800 # 24 kHz, 16‑bit, mono → 200 ms per buffer
MAX_QUEUE_BYTES = MAX_AUDIO_FRAMES * CHUNK_BYTES # 240 KB hard limit
class AudioPipeline:
def __init__(self):
# Fixed‑size pool avoids repeated allocations
self.buffer_pool = [bytearray(CHUNK_BYTES) for _ in range(8)]
self.pool_idx = 0
self.frame_queue = collections.deque(maxlen=MAX_AUDIO_FRAMES)
self.current_bytes = 0
def _next_buffer(self) -> bytearray:
buf = self.buffer_pool[self.pool_idx]
self.pool_idx = (self.pool_idx + 1) % len(self.buffer_pool)
return buf
async def capture_loop(self, mic):
"""Read microphone frames, enforce byte budget, and keep semantic continuity."""
while True:
buf = self._next_buffer()
n = await mic.read_into(buf) # n bytes filled
if n == 0:
continue
# If the new frame would exceed the cap, drop the whole queue.
if self.current_bytes + n > MAX_QUEUE_BYTES:
self.frame_queue.clear()
self.current_bytes = 0
continue
self.frame_queue.append(memoryview(buf[:n]))
self.current_bytes += n
The key difference is the current_bytes guard. Instead of letting the deque silently discard frames, the pipeline clears the queue when a burst threatens the memory budget, preserving the integrity of the utterance that follows.
2. Back‑Pressure for the SQLite Writer
The original DBWriter.write() pushed commands into an asyncio.Queue(maxsize=100) without handling the QueueFull exception. Under heavy voice command traffic the coroutine would raise asyncio.QueueFull, terminate silently, and lose data. The hardened writer introduces a timeout‑based back‑pressure mechanism and records overflow events for telemetry.
# SQLite writer with timeout‑based back‑pressure
import asyncio
import aiosqlite
class DBWriter:
BACKPRESSURE_TIMEOUT = 2.0 # seconds before we consider the queue saturated
def __init__(self, db_path: str):
self.db_path = db_path
self.command_queue: asyncio.Queue = asyncio.Queue(maxsize=100)
self._conn: aiosqlite.Connection | None = None
self._running = False
self._overflow = 0
async def start(self):
self._conn = await aiosqlite.connect(self.db_path)
self._running = True
asyncio.create_task(self._writer_loop())
async def write(self, sql: str, params: tuple = ()) -> bool:
"""Enqueue a write operation; return False if the queue is saturated."""
try:
await asyncio.wait_for(
self.command_queue.put((sql, params)),
timeout=self.BACKPRESSURE_TIMEOUT,
)
return True
except asyncio.TimeoutError:
self._overflow += 1
return False
async def _writer_loop(self):
while self._running:
sql, params = await self.command_queue.get()
await self._conn.execute(sql, params)
await self._conn.commit()
If the queue remains full for more than two seconds, the caller receives a False result and can decide whether to drop the operation, retry later, or surface a warning to the user interface.
3. Process‑Pool Model That Respects Memory Limits
Loading the ONNX model in the parent process and passing it to run_in_executor creates two copies of the model: one in the main interpreter and another in each worker. Because the model object is not picklable, the executor either fails or silently reloads the model, doubling memory pressure. The fix is to initialise the model inside each worker process using the initializer argument of ProcessPoolExecutor.
# Worker‑local model loading for speech‑to‑text
from concurrent.futures import ProcessPoolExecutor
import asyncio
_worker_model = None # will hold the model inside the worker process only
def _worker_init(model_path: str):
"""Executed once per worker; loads the quantised ONNX model."""
global _worker_model
_worker_model = load_quantised_onnx(model_path) # user‑provided helper
class STTService:
def __init__(self, loop: asyncio.AbstractEventLoop):
self.executor = ProcessPoolExecutor(
max_workers=1,
initializer=_worker_init,
initargs=("stt_arm_small.onnx",),
)
self.loop = loop
async def transcribe(self, frames: list[memoryview]) -> str:
"""Offload the heavy inference to the isolated worker."""
return await self.loop.run_in_executor(
self.executor, self._transcribe_blocking, frames
)
@staticmethod
def _transcribe_blocking(frames: list[memoryview]) -> str:
# The worker process has a global _worker_model already loaded.
return _worker_model.infer(frames) # placeholder for actual inference call
Because the model lives only inside the worker, the main process stays lightweight, and the overall RAM footprint drops by roughly 45 MB.
4. Atomic Batch Flush in the Sync Engine
The original SyncEngine.enqueue_op() set a timer without any synchronisation, while _flush_batch cleared the pending list without checking for concurrent appends. This race could drop operations during a high‑frequency voice command burst. The hardened version protects the critical section with an asyncio.Lock and resets the timer atomically.
# Sync engine with lock‑protected batch flush
import asyncio
import msgpack
import zlib
class SyncEngine:
BATCH_WINDOW_MS = 200 # time window for grouping ops
def __init__(self, ws):
self.ws = ws
self.pending_ops: list[dict] = []
self._batch_timer: asyncio.TimerHandle | None = None
self._lock = asyncio.Lock()
def enqueue_op(self, op: dict):
"""Add an operation and start a timer if none is active."""
self.pending_ops.append(op)
if not self._batch_timer:
loop = asyncio.get_event_loop()
self._batch_timer = loop.call_later(
self.BATCH_WINDOW_MS / 1000, lambda: asyncio.create_task(self._flush_batch())
)
async def _flush_batch(self):
"""Serialize, compress, and send the batch under a lock."""
async with self._lock:
if not self.pending_ops:
self._batch_timer = None
return
payload = msgpack.packb(self.pending_ops)
compressed = zlib.compress(payload, level=6)
self.pending_ops.clear()
self._batch_timer = None
await self.ws.send(compressed)
The lock guarantees that no new operation can be appended while the batch is being packaged and transmitted, eliminating lost updates.
5. End‑to‑End Memory Profile
| Component | Naive (no limits) | Refactored (soft limits) | Hardened (hard caps) |
|---|---|---|---|
| Audio queue | Unbounded → OOM | 240 KB cap | 240 KB + burst guard |
| STT worker | Model in parent | Separate process | Worker‑local init |
| DB writer | Shared connections | Single serial writer | Back‑pressure timeout |
| Sync engine | Per‑op send | 200 ms batch | Lock‑protected flush |
| Peak RAM usage | ~890 MB | ~310 MB | ~287 MB |
On a Raspberry Pi 4 with 4 GB RAM the hardened stack peaks at roughly 145 MB, leaving ample headroom for the Chromium‑based UI and the underlying OS.
6. Deployment Checklist
-
Validate buffer pool size , ensure
BUFFER_POOL_SIZE * CHUNK_BYTESfits comfortably within the device’s cache. -
Run a stress test , simulate a PA system that emits 10× the normal frame rate for 30 seconds; verify that the audio queue never exceeds
MAX_QUEUE_BYTES.
Top comments (0)