Free SKILL.md scraped from GitHub. Clone the repo or copy the file directly into your Claude Code skills directory.
npx versuz@latest install hiyenwong-ai-collection-collection-skills-async-delta-modulator-bmigit clone https://github.com/hiyenwong/ai_collection.gitcp ai_collection/SKILL.MD ~/.claude/skills/hiyenwong-ai-collection-collection-skills-async-delta-modulator-bmi/SKILL.md---
name: async-delta-modulator-bmi
version: v1.0.0
last_updated: 2026-04-19
description: >
Asynchronous Delta Modulation (ADM) for Brain-Machine Interface (BMI) applications.
Converts continuous neural signals into event-driven spike trains using adaptive
delta modulation, enabling ultra-low-power, low-latency neural encoding for implantable
and wearable BCI systems. Covers adaptive threshold mechanism, event-driven architecture,
implementation patterns, and common pitfalls.
keywords:
- asynchronous delta modulation
- brain-machine interface
- BMI
- neural spike encoding
- event-driven processing
- adaptive threshold
- low-power neural coding
- 异步增量调制
- 脑机接口
- 脉冲编码
- 事件驱动
---
# Asynchronous Delta Modulator for Brain-Machine Interface (BMI)
## Overview
Asynchronous Delta Modulation (ADM) is a signal encoding technique that converts continuous
neural signals (EEG, LFP, spike waveforms) into asynchronous event-based spike trains.
Unlike synchronous sampling-based approaches, ADM generates output events **only when the
input signal changes by more than an adaptive threshold**, making it inherently sparse,
energy-efficient, and well-suited for neural encoding in BMI systems.
### Why ADM for BMI?
| Aspect | Traditional ADC | ADM Encoder |
|--------|----------------|-------------|
| Sampling | Fixed clock rate | Event-driven, no clock |
| Data rate | Constant (high) | Signal-dependent (sparse) |
| Power | Clock + conversion | Near-zero at rest |
| Latency | 1/f_s sampling delay | Instantaneous on change |
| Output | Quantized samples | Spike events (time, polarity) |
### Key Advantages
- **Ultra-low power**: No clock domain; activity scales with signal dynamics
- **Bandwidth compression**: Only transmits meaningful changes
- **Natural spike compatibility**: Output matches SNN input format
- **Low latency**: No sampling delay — event fires immediately on threshold crossing
---
## Asynchronous Delta Modulation Principle
### Core Equation
The ADM encoder maintains an internal estimate `x̂(t)` of the input signal `x(t)`:
```
δ(t) = x(t) - x̂(t) # instantaneous error
spike(t) = |δ(t)| ≥ Δ(t) # fire when error exceeds threshold
x̂(t+) = x̂(t-) ± Δ(t) # update estimate on spike
```
Where:
- `x(t)`: continuous neural input (e.g., EEG voltage at electrode)
- `x̂(t)`: internal stepwise reconstruction
- `Δ(t)`: adaptive quantization step (threshold)
- `spike(t)`: binary event with polarity (UP = +Δ, DOWN = -Δ)
### Operation Cycle
```
Time → ─────────────────────────────────────────────►
Input: ━━━/‾‾‾\___/‾‾‾‾‾‾\____/‾‾\______________
↑ ↑ ↑ ↑
Estimate: ━┻━━━━━━┻━━━━━━━━━┻━━━━┻━━━━━━━━━━━━━━━━
│ │ │ │
Spike: ↑ ↓ ↑ ↓ (event times)
+Δ -Δ +Δ -Δ (encoded output)
```
### Algorithm Pseudocode
```python
class AsynchronousDeltaModulator:
def __init__(self, delta_init=1.0, delta_min=0.1, delta_max=10.0):
self.x_hat = 0.0 # internal estimate
self.delta = delta_init # adaptive threshold
self.delta_min = delta_min
self.delta_max = delta_max
self.last_spike_time = 0.0
def process(self, x, t):
"""Process one input sample; returns spike event or None."""
error = x - self.x_hat
if abs(error) >= self.delta:
polarity = 1 if error > 0 else -1
self.x_hat += polarity * self.delta
self.adapt_threshold(polarity, t)
self.last_spike_time = t
return {"time": t, "polarity": polarity, "estimate": self.x_hat}
return None # no spike — silence is meaningful
```
---
## Adaptive Threshold Mechanism for Spike Encoding
The adaptive threshold `Δ(t)` is the heart of ADM. A static threshold either:
- **Too high**: misses small but meaningful neural features
- **Too low**: floods the system with noise-triggered spikes
### Exponential Adaptation Rule
```python
def adapt_threshold(self, polarity, dt):
"""Adjust delta based on recent spike activity."""
# Inter-spike interval (ISI) based adaptation
isi = dt - self.last_spike_time if self.last_spike_time > 0 else float('inf')
if isi < self.isi_target:
# Spiking too fast → increase threshold (reduce sensitivity)
self.delta *= self.alpha_up # e.g., alpha_up = 1.1
elif isi > self.isi_target * 2:
# Spiking too slow → decrease threshold (increase sensitivity)
self.delta *= self.alpha_down # e.g., alpha_down = 0.95
# Clamp to bounds
self.delta = max(self.delta_min, min(self.delta_max, self.delta))
```
### Multi-Timescale Adaptation
For neural signals with multiple frequency components (e.g., EEG + spike band):
```python
def multi_scale_adaptation(self, error, t):
"""Fast + slow adaptation tracks both spikes and slow drifts."""
# Fast component: tracks rapid changes (spike band 300-3000 Hz)
self.delta_fast *= (1 + self.k_fast * abs(error))
self.delta_fast = np.clip(self.delta_fast, self.delta_fast_min, self.delta_fast_max)
# Slow component: tracks baseline drift (delta/theta 1-8 Hz)
self.delta_slow += self.k_slow * (abs(error) - self.delta_slow)
# Combined threshold
self.delta = self.delta_fast + self.delta_slow
```
### Threshold Adaptation Parameters
| Parameter | Typical Range | Description |
|-----------|--------------|-------------|
| `Δ_init` | 0.5–2.0 × σ_signal | Initial threshold relative to signal std |
| `α_up` | 1.02–1.15 | Threshold increase factor on rapid spiking |
| `α_down` | 0.90–0.98 | Threshold decrease factor during silence |
| `Δ_min` | 0.01–0.1 × σ_signal | Floor (prevents runaway noise spikes) |
| `Δ_max` | 5–20 × σ_signal | Ceiling (ensures tracking of large transients) |
| `ISI_target` | 1–10 ms | Target inter-spike interval for neural signals |
---
## Event-Driven BMI Architecture
### Full Pipeline
```
┌─────────────┐ ┌──────────────────┐ ┌────────────────┐ ┌──────────────┐
│ Neural │ │ Analog Front │ │ Asynchronous │ │ Event │
│ Source │──►│ End (Amp + │──►│ Delta │──►│ Processor │
│ (EEG/ECoG/ │ │ Filter + │ │ Modulator │ │ (SNN / │
│ spike) │ │ Anti-alias) │ │ Array │ │ Decoder) │
└─────────────┘ └──────────────────┘ └────────────────┘ └──────────────┘
↑ │ │
│ ┌──────────────────┐ │ │
└──────────│ Adaptive │◄───────────┘ │
│ Threshold │ │
│ Controller │ │
└──────────────────┘ │
▼
┌──────────────────┐
│ Motor / Cursor / │
│ Stimulator / │
│ Feedback Output │
└──────────────────┘
```
### Multi-Channel Architecture
```python
class ADM_BMI_System:
def __init__(self, n_channels=64, shared_adaptation=True):
# Per-channel ADM encoders
self.channels = [
AsynchronousDeltaModulator(
delta_init=self._estimate_noise_level(ch),
delta_min=0.05, delta_max=15.0
)
for ch in range(n_channels)
]
# Shared adaptation controller (optional)
if shared_adaptation:
self.global_controller = GlobalThresholdController(
n_channels=n_channels,
target_spike_rate=50.0 # Hz per channel
)
def encode_frame(self, neural_data, timestamps):
"""Encode multi-channel neural data into spike events."""
all_events = []
for ch_idx, (channel, data, ts) in enumerate(zip(
self.channels, neural_data.T, timestamps.T
)):
for x, t in zip(data, ts):
event = channel.process(x, t)
if event:
event['channel'] = ch_idx
all_events.append(event)
# Global adaptation (prevents channel starvation/dominance)
if hasattr(self, 'global_controller'):
self.global_controller.update(self.channels)
return sorted(all_events, key=lambda e: e['time'])
```
### Event Stream Format
```python
# Each spike event is a minimal tuple:
SpikeEvent = namedtuple('SpikeEvent', ['timestamp_us', 'channel_id', 'polarity'])
# - timestamp_us: microsecond-precision event time
# - channel_id: electrode/channel index
# - polarity: +1 (UP) or -1 (DOWN)
# Address-Event Representation (AER) compatible:
# [31:24] = channel_id (up to 256 channels)
# [23:1] = relative timestamp
# [0] = polarity bit
```
---
## Implementation Patterns
### Pattern 1: Pure Python Reference Implementation
```python
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class SpikeEvent:
time: float # seconds
polarity: int # +1 or -1
channel: int = 0
delta: float = 0.0 # threshold at firing time
class ADM_Encoder:
"""
Asynchronous Delta Modulator for neural signal encoding.
Suitable for EEG, LFP, and single-unit spike waveforms.
"""
def __init__(
self,
delta_init: float = 1.0,
delta_min: float = 0.1,
delta_max: float = 10.0,
alpha_up: float = 1.05,
alpha_down: float = 0.95,
isi_target: float = 0.005, # 5 ms target ISI
leak: float = 0.0, # optional leaky estimate
):
self.x_hat = 0.0
self.delta = delta_init
self.delta_min = delta_min
self.delta_max = delta_max
self.alpha_up = alpha_up
self.alpha_down = alpha_down
self.isi_target = isi_target
self.leak = leak
self.last_spike_time = -1.0
self.spike_count = 0
def encode(self, signal: np.ndarray, times: np.ndarray,
channel: int = 0) -> List[SpikeEvent]:
"""Encode a continuous signal into spike events."""
events = []
for x, t in zip(signal, times):
event = self.step(x, t, channel)
if event is not None:
events.append(event)
return events
def step(self, x: float, t: float, channel: int = 0) -> Optional[SpikeEvent]:
"""Process one sample."""
# Optional leak on estimate
if self.leak > 0 and self.last_spike_time >= 0:
dt = t - self.last_spike_time
self.x_hat *= np.exp(-self.leak * dt)
error = x - self.x_hat
if abs(error) >= self.delta:
polarity = 1 if error > 0 else -1
self.x_hat += polarity * self.delta
self._adapt(t)
event = SpikeEvent(
time=t, polarity=polarity, channel=channel, delta=self.delta
)
self.last_spike_time = t
self.spike_count += 1
return event
return None
def _adapt(self, t: float):
"""Adaptive threshold adjustment."""
if self.last_spike_time < 0:
return
isi = t - self.last_spike_time
if isi < self.isi_target:
self.delta *= self.alpha_up
else:
self.delta *= self.alpha_down
self.delta = np.clip(self.delta, self.delta_min, self.delta_max)
def get_stats(self) -> dict:
return {
"spike_count": self.spike_count,
"final_delta": self.delta,
"final_estimate": self.x_hat,
}
```
### Pattern 2: NumPy Vectorized Batch Encoding
```python
def encode_batch_vectorized(
signals: np.ndarray, # shape: (n_channels, n_samples)
times: np.ndarray, # shape: (n_samples,)
deltas_init: np.ndarray, # shape: (n_channels,)
delta_min: float = 0.1,
delta_max: float = 10.0,
alpha_up: float = 1.05,
alpha_down: float = 0.95,
isi_target: float = 0.005,
) -> List[SpikeEvent]:
"""
Vectorized batch encoding for offline analysis / training.
Still sequential in time (ADM is inherently temporal).
"""
n_ch, n_samples = signals.shape
x_hat = np.zeros(n_ch)
deltas = deltas_init.copy()
last_spike_t = np.full(n_ch, -np.inf)
all_events = []
for i in range(n_samples):
t = times[i]
errors = signals[:, i] - x_hat
fired = np.abs(errors) >= deltas
if np.any(fired):
polarity = np.sign(errors[fired])
x_hat[fired] += polarity * deltas[fired]
# Adaptation for fired channels
isi = t - last_spike_t[fired]
deltas[fired] *= np.where(
isi < isi_target, alpha_up, alpha_down
)
deltas[fired] = np.clip(deltas[fired], delta_min, delta_max)
last_spike_t[fired] = t
for ch_idx, pol, d in zip(
np.where(fired)[0], polarity, deltas[fired]
):
all_events.append(
SpikeEvent(time=t, polarity=int(pol), channel=int(ch_idx), delta=d)
)
return all_events
```
### Pattern 3: Real-Time Streaming with Ring Buffer
```python
from collections import deque
import threading
class Streaming_ADM_BMI:
"""
Real-time ADM encoder for streaming neural data.
Suitable for implantable/wearable BMI hardware.
"""
def __init__(self, n_channels=64, buffer_size=1024):
self.n_channels = n_channels
self.encoders = [ADM_Encoder() for _ in range(n_channels)]
self.event_queue = deque(maxlen=buffer_size)
self.lock = threading.Lock()
self._running = False
def feed_sample(self, sample: np.ndarray, timestamp: float):
"""Ingest one sample from ADC/hardware."""
for ch in range(self.n_channels):
event = self.encoders[ch].step(sample[ch], timestamp, channel=ch)
if event is not None:
with self.lock:
self.event_queue.append(event)
def get_events(self, max_events: int = 256) -> List[SpikeEvent]:
"""Drain pending events for downstream processing."""
with self.lock:
events = []
for _ in range(min(max_events, len(self.event_queue))):
events.append(self.event_queue.popleft())
return events
def get_spike_rates(self, window_ms: float = 1000.0) -> np.ndarray:
"""Estimate per-channel firing rates from recent history."""
now = self.event_queue[-1].time if self.event_queue else 0
rates = np.zeros(self.n_channels)
for event in self.event_queue:
if now - event.time < window_ms / 1000.0:
rates[event.channel] += 1
return rates * (1000.0 / window_ms) # convert to Hz
```
### Pattern 4: Reconstruction / Decoding
```python
def reconstruct_from_events(
events: List[SpikeEvent],
n_channels: int,
delta_fixed: Optional[float] = None,
t_max: Optional[float] = None,
n_samples: int = 1000,
) -> np.ndarray:
"""
Reconstruct approximate continuous signal from ADM spike events.
Useful for validation, visualization, and offline analysis.
"""
if t_max is None:
t_max = events[-1].time if events else 1.0
times = np.linspace(0, t_max, n_samples)
reconstruction = np.zeros((n_channels, n_samples))
for ch in range(n_channels):
x_hat = 0.0
idx = 0
ch_events = [e for e in events if e.channel == ch]
event_idx = 0
for t_idx, t in enumerate(times):
# Process events up to current time
while event_idx < len(ch_events) and ch_events[event_idx].time <= t:
d = delta_fixed if delta_fixed else ch_events[event_idx].delta
x_hat += ch_events[event_idx].polarity * d
event_idx += 1
reconstruction[ch, t_idx] = x_hat
return reconstruction
```
---
## Activation Keywords
### English
- asynchronous delta modulation
- delta modulator BMI
- neural spike encoding
- event-driven neural processing
- adaptive threshold encoding
- BMI signal compression
- low-power neural interface
- asynchronous neural ADC
- spike-based neural coding
- address-event representation neural
- continuous-to-spike conversion
- delta modulation BCI
- neural signal encoding
- event-based brain machine interface
### Chinese (中文)
- 异步增量调制
- 脑机接口脉冲编码
- 事件驱动神经处理
- 自适应阈值编码
- 增量调制器
- 低功耗神经接口
- 异步神经模数转换
- 脉冲编码
- 脑信号压缩编码
- 事件驱动脑机接口
- 连续信号转脉冲
- 神经信号增量调制
---
## Pitfalls & Mitigation Strategies
### 1. Threshold Drift (阈值漂移)
**Problem**: The adaptive threshold can drift away from optimal values due to:
- Prolonged signal silence (Δ decays to minimum)
- Sustained high-frequency activity (Δ saturates at maximum)
- DC offset in neural signals
**Symptoms**:
- Sudden burst of spikes after long silence (threshold too low)
- Missed features during active periods (threshold too high)
- Reconstruction bias / baseline wander
**Mitigation**:
```python
# Strategy A: Hard bounds with periodic reset
def periodic_reset(self, t, reset_interval=10.0):
if t - self.last_reset > reset_interval:
self.delta = self._signal_std * self.initial_multiplier
self.last_reset = t
# Strategy B: Exponential moving average of recent Δ
self.delta_ema = 0.99 * self.delta_ema + 0.01 * self.delta
# Use delta_ema for decisions when delta hits bounds
# Strategy C: High-pass filter input before ADM
signal_hp = butter_highpass(signal, cutoff=1.0, fs=sample_rate)
# Removes DC offset that causes threshold drift
```
### 2. Noise Sensitivity (噪声敏感性)
**Problem**: High-frequency noise triggers false spikes, wasting bandwidth
and corrupting the spike train.
**Symptoms**:
- Uniformly distributed spikes (no signal structure)
- Spike rate much higher than expected for neural data
- Poor reconstruction quality
**Mitigation**:
```python
# Strategy A: Dead zone / hysteresis
class ADM_With_Hysteresis(ADM_Encoder):
def __init__(self, hysteresis_ratio=0.5, **kwargs):
super().__init__(**kwargs)
self.hysteresis = self.delta * hysteresis_ratio
def step(self, x, t, channel=0):
error = x - self.x_hat
# Require larger change to fire in same direction
threshold = self.delta + (
self.hysteresis if self._last_polarity > 0 else -self.hysteresis
)
if abs(error) >= abs(threshold):
# ... spike logic with self._last_polarity tracking
# Strategy B: Noise-aware minimum threshold
def noise_aware_min_threshold(self, noise_rms):
self.delta_min = max(self.delta_min, 2.0 * noise_rms)
# Strategy C: Temporal filtering (leaky estimate acts as LPF)
encoder = ADM_Encoder(leak=1.0 / (2 * np.pi * 300e-3)) # 300 Hz cutoff
```
### 3. Synchronization Issues (同步问题)
**Problem**: In multi-channel systems, asynchronous events from different
channels can arrive out of temporal order at the decoder, or clock drift
between encoder and decoder causes timing errors.
**Symptoms**:
- Decoded multi-channel signals appear misaligned
- Spike correlation analysis shows artifacts
- Downstream SNN receives temporally scrambled input
**Mitigation**:
```python
# Strategy A: Timestamp synchronization header
class Synced_ADM:
SYNC_INTERVAL = 0.1 # seconds
def __init__(self):
self.last_sync = 0.0
def needs_sync(self, t):
return (t - self.last_sync) >= self.SYNC_INTERVAL
def sync_event(self, t):
self.last_sync = t
return SpikeEvent(time=t, polarity=0, channel=0xFF, delta=0)
# Special channel ID = 0xFF marks sync event
# Strategy B: Temporal ordering buffer at receiver
class TemporalReorderBuffer:
def __init__(self, max_delay_us=1000):
self.buffer = deque()
self.max_delay = max_delay_us / 1e6
def insert(self, event):
self.buffer.append(event)
# Sort by timestamp
self.buffer = deque(sorted(self.buffer, key=lambda e: e.time))
def drain_ordered(self, current_time):
"""Release events that are safely ordered."""
cutoff = current_time - self.max_delay
ordered = []
while self.buffer and self.buffer[0].time <= cutoff:
ordered.append(self.buffer.popleft())
return ordered
# Strategy C: AER bus with hardware arbitration
# Use standard AER protocols (e.g., JTAG-based, SPI-based)
# with hardware-level priority encoding for multi-channel ADM
```
### 4. Additional Considerations
| Issue | Cause | Mitigation |
|-------|-------|------------|
| **Slope overload** | Signal changes faster than Δ can track | Increase Δ_max; use predictive ADM |
| **Granular noise** | Δ too large for quiet signal regions | Lower Δ_min; add dithering |
| **Channel crosstalk** | Shared analog front-end coupling | Per-channel calibration; shielded routing |
| **Memory constraints** | Long event histories for adaptation | Fixed-size ring buffer; exponential decay |
| **Calibration drift** | Hardware parameter changes over time | Periodic recalibration with known stimuli |
---
## Related BCI/SNN Skills
### Directly Related
- **spiking-neural-network-analysis** — SNN pattern extraction and analysis
- **adaptive-spiking-neuron-multimodal** — Adaptive spiking neuron models (ASN)
- **spiking-memristor-multimodal** — Memristive neuron hardware for spike encoding
- **spike-image-decoder** — Spike-based decoding methodologies
- **snn-learning-survey** — SNN learning algorithms and approaches
- **quantized-snn-hardware-optimization** — SNN hardware deployment
- **snn-firing-distribution-quantization** — Spike firing distribution analysis
### BMI & Neural Decoding
- **neural-digital-twins-bci** — Neural digital twins for BCI
- **neural-population-decoding** — Population-level neural decoding
- **neural-encoding-evaluation-meeg** — Neural encoding evaluation for M/EEG
- **bci-rehabilitation-protocols** — BCI rehabilitation applications
- **eeg-ieeg-bridge-bci** — EEG-to-iEEG transfer for BCI
- **copilot-assisted-second-thought-bci** — AI-assisted BCI decoding
- **rl-closed-loop-eeg-tms** — RL for closed-loop EEG-TMS
- **sensorless-gaze-following** — Gaze following without sensors
### Spiking & Event-Driven Computing
- **spiking-compositional-neural-operator** — Spiking neural operators
- **wta-spiking-transformer-language** — Winner-take-all spiking transformers
- **gemst-multidimensional-grouping-snn** — Grouped spiking transformers
- **spiking-reservoir-robustness** — Spiking reservoir computing
- **adaptive-spiking-neurons-asn** — Adaptive spiking neuron models
- **decolle-snn-learning** — DECOLLE local learning for SNNs
- **bio-neuron-snn-learning** — Biologically plausible SNN learning
### Brain Connectivity & Network Analysis
- **brain-connectivity-analysis** — Brain network analysis
- **eeg-brain-connectivity-bci** — EEG-based brain connectivity for BCI
- **thermodynamic-brain-connectivity** — Thermodynamic approaches to connectivity
- **kuramoto-brain-network** — Kuramoto model for brain networks
---
## References & Further Reading
1. **Asynchronous Delta Modulation for Neural Interfaces** — Core methodology for event-based neural encoding
2. **Address-Event Representation (AER)** — Standard protocol for event-based neuromorphic systems
3. **Sigma-Delta Modulation** — Related oversampling technique with noise shaping
4. **Event-Based Vision Sensors (DVS)** — Parallel development in visual domain; similar principles
5. **Neuromorphic Engineering** — General field covering event-driven neural hardware
## Quick Reference Card
```
┌─────────────────────────────────────────────────────┐
│ ADM BMI Quick Reference │
├─────────────────────────────────────────────────────┤
│ Input: Continuous neural signal x(t) │
│ Output: Spike events {(t_i, polarity_i)} │
│ Rule: spike when |x(t) - x̂(t)| ≥ Δ(t) │
│ Update: x̂(t+) = x̂(t-) ± Δ(t) │
│ Adapt: Δ ← Δ × α_up (if ISI < target) │
│ Δ ← Δ × α_down (if ISI ≥ target) │
├─────────────────────────────────────────────────────┤
│ Key Params: │
│ Δ_init ≈ 1.0 × σ_signal │
│ α_up = 1.02–1.15 │
│ α_down = 0.90–0.98 │
│ ISI_target = 1–10 ms │
├─────────────────────────────────────────────────────┤
│ Watch for: │
│ ⚠ Threshold drift → periodic reset │
│ ⚠ Noise sensitivity → hysteresis + HPF │
│ ⚠ Sync issues → temporal reorder buffer │
│ ⚠ Slope overload → increase Δ_max │
└─────────────────────────────────────────────────────┘
```