Messages on the trigger line
On the topic of multi-electrode array recordings and 2-photon recordings, something that I have wanted to do for a while but was never pressing enough to tackle is to have descriptive messages encoded onto the trigger channel of a recording.
Claude Opus 4.8 pretty much one-shotted a solution. Our Arduino device for sending triggers can now send utf-8 encoded messages along with a CRC check. Now I don’t need to worry about keeping track of what stimulus I played, as it’s all saved on the trigger line.
Apart from reducing the tedious and error prone process of manually noting down the sequence of played stimuli, this feature also allows Opus to one-shot useful visualization of the recording, as enough information is encoded for each stimulus that the stimulus source files can be matched to each stimulus block in the recording.
The code, by Opus 4.8, used to encode and decode is below.
Encoding on the Arduino:
/*
* Arduino trigger sender.
*
* pin_LED drives the wire into the MEA's trigger input, which the recorder
* stores as channel index 1. Serial commands:
*
* 'T' -> a 1 ms HIGH pulse (the stimulus trigger)
* 'M<text>\n' -> transmit <text> as a framed, CRC-checked message,
* then reply "MOK" (or "MERR" if the payload overflowed)
* anything else -> pin LOW
*
*
* MESSAGE ENCODING -- inverted UART 8N1 at BIT_US per bit: idle LOW, start bit
* HIGH, 8 data bits LSB-first, stop bit LOW.
*
* frame := 0x55 x4 (preamble) | 0x7E (sync) | len | payload | crc16 (lsb first)
*/
int pin_LED = 13;
const unsigned long BIT_US = 1000; // 1 ms/bit -> ~18 samples/bit at 18 kHz
const uint8_t PREAMBLE_BYTE = 0x55;
const uint8_t PREAMBLE_COUNT = 4;
const uint8_t SYNC_BYTE = 0x7E;
const uint8_t MAX_PAYLOAD = 255;
const unsigned long GUARD_BITS = 4; // idle LOW before the preamble
static uint8_t msg_buf[MAX_PAYLOAD];
static uint16_t msg_len = 0;
static bool is_sending = false;
static bool overflowed = false;
static unsigned long next_edge = 0;
void setup() {
Serial.begin(250000);
pinMode(pin_LED, OUTPUT);
digitalWrite(pin_LED, false);
/* On boot, print "READY" once. This is announcing readiness to the host
machine which would otherwise have to hard-code a boot delay waiting for
the Arduino to boot. */
Serial.println("READY");
}
/* Hold the LED pin at either HIGH or LOW for a 1 bit period. */
static void send_bit(bool level) {
digitalWrite(pin_LED, level);
next_edge += BIT_US;
// Signed difference so this still terminates across the ~70 min micros() wrap.
while ((long)(micros() - next_edge) < 0) {
// busy-wait.
}
}
/* Send 8-bits on the trigger line, one at a time, sarting from the least significant. */
static void send_byte(uint8_t v) {
send_bit(true); // Start
for (uint8_t i = 0; i < 8; i++) {
send_bit((v >> i) & 1);
}
send_bit(false); // Stop
}
static uint16_t crc16_update(uint16_t crc, uint8_t b) {
crc ^= (uint16_t)b << 8;
for (uint8_t i = 0; i < 8; i++) {
crc = (crc & 0x8000) ? (uint16_t)((crc << 1) ^ 0x1021) : (uint16_t)(crc << 1);
}
return crc;
}
static void send_message(const uint8_t *payload, uint8_t len) {
uint16_t crc = 0xFFFF; // CCITT-FALSE, over len || payload
crc = crc16_update(crc, len);
for (uint8_t i = 0; i < len; i++) {
crc = crc16_update(crc, payload[i]);
}
digitalWrite(pin_LED, false);
delayMicroseconds(GUARD_BITS * BIT_US);
next_edge = micros();
for (uint8_t i = 0; i < PREAMBLE_COUNT; i++) {
send_byte(PREAMBLE_BYTE);
}
send_byte(SYNC_BYTE);
send_byte(len);
for (uint8_t i = 0; i < len; i++) {
send_byte(payload[i]);
}
send_byte((uint8_t)(crc & 0xFF));
send_byte((uint8_t)(crc >> 8));
digitalWrite(pin_LED, false);
}
void serialEvent()
{
char ch;
while (Serial.available()) {
ch = (char)Serial.read();
if (is_sending) {
if (ch == '\n') {
is_sending = false;
if (overflowed) {
Serial.println("MERR");
} else {
send_message(msg_buf, (uint8_t)msg_len);
Serial.println("MOK"); // the host waits for this before playing
}
}
else if (msg_len < MAX_PAYLOAD) {
msg_buf[msg_len++] = (uint8_t)ch;
}
else {
// Drop the frame rather than transmit a silently truncated one.
overflowed = true;
}
continue;
}
// Standard trigger on-off
if (ch == 'T') {
digitalWrite(pin_LED, true);
delayMicroseconds(1000);
digitalWrite(pin_LED, false);
}
// Send a message.
else if (ch == 'M') {
is_sending = true;
msg_len = 0;
overflowed = false;
}
else
digitalWrite(pin_LED, false);
}
}
Decoding in Python, in this case from 3Brain brw files.
"""Encode and recover string messages carried on the 3Brain MEA trigger wire.
The Arduino puts framed, CRC-checked messages on the trigger line as inverted
UART 8N1 at 1 ms/bit -- idle LOW, start bit HIGH, 8 data bits LSB-first, stop
bit LOW::
frame := 0x55 x4 (preamble) | 0x7E (sync) | len | payload | crc16 (lsb first)
The recorder is the receiver, so decoding happens offline against the *raw*
trigger channel (channel index 1 of a BioCam recording).
Two clocks are involved and neither is exact: the BioCam samples at ~17855 Hz
rather than 18 kHz, and the Arduino has its own oscillator tolerance. So the
decoder never free-runs a bit clock. It re-locks on the rising edge of every
byte's start bit, which keeps error from accumulating across a frame, and it
calibrates the true bit period against the preamble, whose length in bits is
known. Together those tolerate the two clocks disagreeing by a few percent.
Typical use::
trace, fs = load_trigger_channel("Phase_01.brw")
for msg in decode_trace(trace, fs):
print(msg.sync_sample / fs, msg.text)
"""
from __future__ import annotations
import dataclasses
import numpy as np
import spikeinterface.full as si
BIT_US = 1000.0
PREAMBLE_BYTE = 0x55
PREAMBLE_COUNT = 4
SYNC_BYTE = 0x7E
MAX_PAYLOAD = 255
BITS_PER_BYTE = 10 # start + 8 data + stop
# Channel 0 is the reference electrode, 1 is the trigger/TTL line. See
# claudebooks/projects/spikesort/share/spikesort/src/spikesort/pipeline.py.
TRIGGER_CHANNEL_INDEX = 1
# A frame start is preceded by GUARD_BITS of idle LOW in the firmware. Requiring
# most of that both finds frame starts cheaply in a long recording and rules out
# edges inside a frame, where the longest LOW run is 9 bits within a 0x00 byte
# but only 2 bits inside the preamble.
_GUARD_BITS_MIN = 2.5
@dataclasses.dataclass(frozen=True)
class DecodedMessage:
"""One successfully CRC-checked message recovered from a trace."""
text: str
sync_sample: int
"""Index of the rising edge of the sync byte's start bit."""
bit_period_samples: float
"""Bit period measured against the preamble, in samples."""
def crc16_ccitt(data: bytes) -> int:
"""CRC-16/CCITT-FALSE. Mirrors crc16_update() in the firmware."""
crc = 0xFFFF
for b in data:
crc ^= b << 8
for _ in range(8):
if crc & 0x8000:
crc = ((crc << 1) ^ 0x1021) & 0xFFFF
else:
crc = (crc << 1) & 0xFFFF
return crc
def frame_bytes(text: str) -> bytes:
"""The byte sequence the firmware transmits for `text`, framing included."""
payload = text.encode("utf-8")
if len(payload) > MAX_PAYLOAD:
raise ValueError(
f"payload is {len(payload)} bytes (UTF-8), max is {MAX_PAYLOAD}"
)
body = bytes([len(payload)]) + payload
crc = crc16_ccitt(body)
return (
bytes([PREAMBLE_BYTE]) * PREAMBLE_COUNT
+ bytes([SYNC_BYTE])
+ body
+ bytes([crc & 0xFF, crc >> 8])
)
def encode_bits(text: str) -> np.ndarray:
"""The wire bit sequence for `text`: each byte as start, 8 data LSB-first, stop."""
bits = []
for b in frame_bytes(text):
bits.append(1)
bits.extend((b >> i) & 1 for i in range(8))
bits.append(0)
return np.array(bits, dtype=np.uint8)
def simulate_trace(
text: str,
fs: float,
bit_us: float = BIT_US,
clock_error: float = 0.0,
noise: float = 0.0,
levels: tuple[float, float] = (2048.0, 4095.0),
idle_s: float = 0.05,
rng: np.random.Generator | None = None,
) -> np.ndarray:
"""Render the waveform the firmware emits for `text`, as the recorder sees it.
`clock_error` models the Arduino's oscillator against the recorder's:
0.01 means its bit period runs 1% long. `levels` is (LOW, HIGH) in raw ADC
counts; a TTL high saturates the channel, and the LOW level depends on where
the input sits at rest, so only their separation matters. `noise` is the
gaussian sigma added in the same counts.
"""
bits = encode_bits(text)
spb = fs * bit_us * (1.0 + clock_error) / 1e6
n_idle = int(round(idle_s * fs))
n_msg = int(np.ceil(len(bits) * spb))
low, high = levels
trace = np.full(n_idle * 2 + n_msg, low, dtype=np.float64)
# Sample a piecewise-constant waveform: which bit each sample lands in.
k = np.floor(np.arange(n_msg) / spb).astype(int)
trace[n_idle : n_idle + n_msg] = np.where(bits[np.clip(k, 0, len(bits) - 1)], high, low)
if noise:
rng = np.random.default_rng() if rng is None else rng
trace += rng.normal(0.0, noise, size=trace.shape)
return trace
def _binarize(trace: np.ndarray, threshold: float | None) -> np.ndarray:
if threshold is None:
# The wire rails, so the midpoint separates the two levels regardless of
# where LOW sits. On a trace with no message this is meaningless, but
# then no candidate survives the idle-guard test below anyway.
threshold = 0.5 * (float(np.min(trace)) + float(np.max(trace)))
return trace > threshold
def _bit_at(hi: np.ndarray, centre: float, spb: float) -> int | None:
"""Majority vote over the middle half of the bit at `centre`."""
half = max(1, int(round(spb / 4)))
a = int(round(centre)) - half
b = int(round(centre)) + half + 1
if a < 0 or b > len(hi):
return None
return 1 if hi[a:b].mean() >= 0.5 else 0
def _read_byte(hi: np.ndarray, edge: int, spb: float) -> int | None:
"""Read one inverted-UART byte whose start bit rises at `edge`."""
if _bit_at(hi, edge + 0.5 * spb, spb) != 1:
return None # start bit must hold, so a 1-sample spike is not a byte
value = 0
for i in range(8):
bit = _bit_at(hi, edge + (1.5 + i) * spb, spb)
if bit is None:
return None
value |= bit << i
if _bit_at(hi, edge + 9.5 * spb, spb) != 0:
return None # stop bit
return value
def _next_edge(rising: np.ndarray, expected: float, tol: float) -> int | None:
i = int(np.searchsorted(rising, expected - tol))
if i < len(rising) and rising[i] <= expected + tol:
return int(rising[i])
return None
def _read_frame(
hi: np.ndarray, rising: np.ndarray, first_edge: int, spb: float
) -> DecodedMessage | None:
"""Try to read a whole frame whose first preamble byte starts at `first_edge`."""
edge: int | None = first_edge
for _ in range(PREAMBLE_COUNT):
if _read_byte(hi, edge, spb) != PREAMBLE_BYTE:
return None
edge = _next_edge(rising, edge + BITS_PER_BYTE * spb, 0.5 * spb)
if edge is None:
return None
if _read_byte(hi, edge, spb) != SYNC_BYTE:
return None
sync_edge = edge
# The preamble spans a known number of bit periods, so measuring it
# calibrates the Arduino's clock against the recorder's -- the only clock
# error that matters here. Use the measured period for the rest of the frame.
measured = (sync_edge - first_edge) / (PREAMBLE_COUNT * BITS_PER_BYTE)
if not 0.8 * spb <= measured <= 1.25 * spb:
return None
spb = measured
edge = _next_edge(rising, edge + BITS_PER_BYTE * spb, 0.5 * spb)
if edge is None:
return None
length = _read_byte(hi, edge, spb)
if length is None:
return None
body: list[int] = [length]
for _ in range(length + 2): # payload + crc16
edge = _next_edge(rising, edge + BITS_PER_BYTE * spb, 0.5 * spb)
if edge is None:
return None
value = _read_byte(hi, edge, spb)
if value is None:
return None
body.append(value)
crc_rx = body[1 + length] | (body[2 + length] << 8)
if crc16_ccitt(bytes(body[: 1 + length])) != crc_rx:
return None
try:
text = bytes(body[1 : 1 + length]).decode("utf-8")
except UnicodeDecodeError:
return None
return DecodedMessage(
text=text, sync_sample=int(sync_edge), bit_period_samples=spb
)
def decode_trace(
trace: np.ndarray,
fs: float,
bit_us: float = BIT_US,
threshold: float | None = None,
) -> list[DecodedMessage]:
"""Recover every CRC-valid message in `trace`, in order of occurrence.
`fs` should come from the recording itself (BioCam is ~17855 Hz, not 18000);
passing the nominal rate still works, but passing the real one is free.
`threshold` defaults to the midpoint of the trace's min and max, which is
right for a railing TTL; pass it explicitly for an unusual channel.
Only frames whose CRC-16 matches are returned, so stray 'T' trigger pulses
are rejected rather than decoded as text.
"""
spb = fs * bit_us / 1e6
if spb < 4:
raise ValueError(
f"{spb:.1f} samples/bit is too few to decode reliably; "
f"raise bit_us or check fs"
)
hi = _binarize(trace, threshold)
d = np.diff(hi.astype(np.int8))
rising = np.flatnonzero(d == 1) + 1
falling = np.flatnonzero(d == -1) + 1
if len(rising) == 0:
return []
# A frame starts after the firmware's idle-LOW guard. Filtering on that first
# keeps this cheap on a 20-minute recording: only trigger pulses and real
# frame starts survive, and a noise-only trace yields no candidates at all.
prev_fall = np.searchsorted(falling, rising) - 1
low_run = np.where(
prev_fall >= 0, rising - falling[np.clip(prev_fall, 0, None)], rising
)
candidates = rising[low_run >= _GUARD_BITS_MIN * spb]
out = []
for edge in candidates:
msg = _read_frame(hi, rising, int(edge), spb)
if msg is not None:
out.append(msg)
return out
def load_trigger_channel(
brw_path,
channel_index: int = TRIGGER_CHANNEL_INDEX,
start_frame: int | None = None,
end_frame: int | None = None,
) -> tuple[np.ndarray, float]:
"""Return (raw trace, fs) for the trigger channel of a BioCam .brw file.
The channel must be read raw: the spikesort preprocessing drops channel 1 and
bandpasses 300-3000 Hz, either of which destroys the message. Recordings are
large (100+ GB), so pass start_frame/end_frame to read only the stretch that
contains the message.
"""
rec = si.read_biocam(str(brw_path))
trace = rec.get_traces(
channel_ids=[rec.get_channel_ids()[channel_index]],
start_frame=start_frame,
end_frame=end_frame,
return_scaled=False,
)[:, 0]
return trace, float(rec.get_sampling_frequency())