From 2d611d74241ce97c122bc647f7149c85f195f508 Mon Sep 17 00:00:00 2001
From: jl33ai <justinkhlee27@gmail.com>
Date: Fri, 29 May 2026 09:27:34 -0700
Subject: [PATCH] add synthetic data source for running without trodes

lets you run the pipeline end to end without an acquisition rig. useful for smoke testing a fresh install, working on a laptop, and CI.

- realtime_decoder/synthetic.py: SyntheticDataReceiver and SyntheticClient that match the surface of TrodesDataReceiver and TrodesClient. drop in replacements. non-blocking __next__ semantics so the polling loops in encoder/decoder/ripple work unchanged.
- runscript.py picks the data source via config['datasource']. defaults to 'trodes' so every existing config keeps working.
- config/demo_synthetic.yml is a 5 rank demo wired to the synthetic source.
- README adds a 'running without acquisition hardware' section.

spikes are poisson, marks gaussian, position walks a triangle wave along a single segment. not biologically realistic, the point is to exercise the data path and message plumbing.

to try it:
  mpiexec -np 5 python -u runscript.py config/demo_synthetic.yml
---
 README.md                     |  17 ++
 config/demo_synthetic.yml     | 175 +++++++++++++++++++
 realtime_decoder/synthetic.py | 305 ++++++++++++++++++++++++++++++++++
 runscript.py                  |  42 +++--
 4 files changed, 528 insertions(+), 11 deletions(-)
 create mode 100644 config/demo_synthetic.yml
 create mode 100644 realtime_decoder/synthetic.py

diff --git a/README.md b/README.md
index db0fc9a..803ee71 100644
--- a/README.md
+++ b/README.md
@@ -24,6 +24,23 @@ mpiexec -np <num_processes> -bind-to hwthread python -u runscript.py <path/to/co
 
 Note: `-bind-to hwthread` is optional but expected to give the best performance if enough threads are available.
 
+### Running without acquisition hardware (synthetic data source)
+
+You can run the full MPI pipeline against an in-process synthetic data
+generator — no Trodes, SpikeGLX, or other acquisition rig required. This
+is useful for smoke-testing an install, developing on a laptop, and CI.
+
+```
+mpiexec -np 5 python -u runscript.py config/demo_synthetic.yml
+```
+
+Selection is driven by the top-level `datasource` config key
+(`trodes` by default; `synthetic` to use the generator). The synthetic
+source produces Poisson spikes with Gaussian mark vectors, walks the
+synthetic animal back and forth along a single linear segment, and
+auto-terminates after `synthetic.run_duration_s` seconds. See
+`realtime_decoder/synthetic.py` for the full set of tunables.
+
 # Configuration
 
 Please see the example configuration file in the `example_config` folder. Options are described in more detail below.
diff --git a/config/demo_synthetic.yml b/config/demo_synthetic.yml
new file mode 100644
index 0000000..666a7a3
--- /dev/null
+++ b/config/demo_synthetic.yml
@@ -0,0 +1,175 @@
+---
+# Demo config: runs the full MPI pipeline against the in-process
+# synthetic data source (realtime_decoder.synthetic). No Trodes, no
+# acquisition hardware, no driver setup required.
+#
+# Run:
+#   mpiexec -np 5 python -u runscript.py config/demo_synthetic.yml
+#
+# Ranks: 0=supervisor, 1=decoder, 2=gui, 3=ripples, 4=encoder
+rank:
+  supervisor: [0]
+  ripples: [3]
+  decoders: [1]
+  encoders: [4]
+  gui: [2]
+rank_settings:
+  enable_rec: [0,1,3,4]
+trode_selection:
+  ripples: [1]
+  decoding: [1]
+decoder_assignment:
+  1: [1]
+algorithm: "clusterless_decoder"
+datasource: "synthetic"
+num_setup_messages: 100
+preloaded_model: false
+frozen_model: false
+files:
+  output_dir: '/tmp/realtime_decoder_demo'
+  prefix: 'demo'
+  rec_postfix: 'bin_rec'
+  timing_postfix: 'timing'
+# --- synthetic-source parameters (all optional, defaults shown) -----------
+synthetic:
+  spike_rate_hz: 30
+  mark_dim: 4
+  mark_amplitude_uv: 120     # well above encoder.spk_amp below
+  track_length_cm: 40        # fits into the 0..41 bins
+  walk_speed_cm_s: 20
+  startup_delay_s: 1.0       # supervisor waits this long, then fires play
+  run_duration_s: 30         # auto-terminate after this many seconds
+  voltage_scaling_factor: 0.195
+sampling_rate:
+  spikes: 30000
+  lfp: 1500
+  position: 30
+ripples:
+  max_ripple_samples: 450
+  vel_thresh: 10
+  freeze_stats: false
+  timings_bufsize: 100000
+  filter:
+    type: 'iir'
+    order: 2
+    crit_freqs: [150, 250]
+    kwargs:
+      btype: 'bandpass'
+      ftype: 'butter'
+  smoothing_filter:
+    num_taps: 15
+    band_edges: [50, 55]
+    desired: [1, 0]
+  threshold:
+    standard: 3.5
+    conditioning: 3.75
+    content: 4
+    end: 0
+encoder:
+  spk_amp: 60
+  use_channel_dist_from_max_amp: 2
+  mark_dim: 4
+  bufsize: 5000
+  timings_bufsize: 5000
+  vel_thresh: 5
+  num_pos_points: 30
+  position:
+    lower: 0
+    upper: 41
+    num_bins: 41
+    arm_ids: [0]
+    arm_coords: [[0,40]]
+  mark_kernel:
+    mean: 0
+    std: 20
+    use_filter: false
+    n_std: 1
+    n_marks_min: 10
+decoder:
+  decoder_to_message: 1
+  bufsize: 2000
+  timings_bufsize: 10000
+  cred_int_bufsize: 10
+  starting_arm1_bin: 10
+  starting_arm2_bin: 30
+  num_pos_points: 30
+  time_bin:
+    samples: 180
+    delay_samples: 180
+clusterless_decoder:
+  state_labels: ['state']
+  transmat_bias: 1
+gui:
+  colormap: 'rocket'
+  send_interval: 0
+  refresh_rate: 25
+  trace_length: 2
+  state_colors: ['#4c72b0','#dd8452', '#55a868']
+  num_xticks: 5
+mua:
+  threshold:
+    trigger: 4
+    end: 0
+  freeze_stats: false
+  moving_avg_window: 5
+stimulation:
+  instructive: false
+  shortcut_msg_on: false       # no ECU in demo mode
+  automatic_threshold_update: false
+  num_each_arm_per_minute: 1.1
+  num_pos_points: 30
+  center_well_loc: [100, 100]
+  max_center_well_dist: 50
+  replay:
+    enabled: false
+    method: "posterior"
+    target_arm: 0
+    event_lockout: 0.2
+    sliding_window: 5
+    primary_arm_threshold: 0.4
+    secondary_arm_threshold: 0.4
+    other_arm_threshold: 0.3
+    max_arm_repeats: 8
+    instr_max_repeats: 3
+    min_unique_trodes: 1
+  ripples:
+    enabled: false
+    type: "standard"
+    method: "multichannel"
+    event_lockout: 0
+    num_above_thresh: 1
+    suprathreshold_period: 0.1
+  head_direction:
+    enabled: false
+    rotate_180: false
+    event_lockout: 10
+    min_duration: 2
+    well_angle_range: 6
+    within_angle_range: 6
+    well_loc: [[100, 100], [200, 200]]
+kinematics:
+  smooth_x: true
+  smooth_y: true
+  smooth_speed: false
+  smoothing_filter: [0.31, 0.29, 0.25, 0.15]
+  scale_factor: 0.2644
+cred_interval:
+  val: 0.5
+  max_num: 5
+display:
+  stim_decider:
+    position: 150
+    decoding_bins: 200
+  ripples:
+    lfp: 10000
+  encoder:
+    encoding_spikes: 500
+    total_spikes: 5000
+    occupancy: 500
+    position: 500
+  decoder:
+    total_spikes: 5000
+    occupancy: 100
+process_monitor:
+  interval: 15
+  timeout: 3
diff --git a/realtime_decoder/synthetic.py b/realtime_decoder/synthetic.py
new file mode 100644
index 0000000..f3c8e30
--- /dev/null
+++ b/realtime_decoder/synthetic.py
@@ -0,0 +1,305 @@
+"""Synthetic data source for the realtime_decoder.
+
+Lets you install the package and run the full MPI pipeline end-to-end
+without any acquisition hardware (Trodes, SpikeGLX, Open Ephys, etc.).
+
+This is intended for:
+  * smoke-testing a fresh install
+  * developing/debugging the decoder loop on a laptop
+  * regression testing in CI
+
+It is NOT intended to be biologically realistic. The synthetic spikes are
+Poisson with a fixed rate, marks are gaussian, and the synthetic position
+walks back and forth on a simple linear track. The goal is to exercise the
+data path and message plumbing, not to validate decoding accuracy.
+
+Wiring is parallel to ``trodesnet.py``: a ``SyntheticDataReceiver`` that
+mirrors ``TrodesDataReceiver`` and a ``SyntheticClient`` that mirrors
+``TrodesClient``. The dispatch happens in ``runscript.py`` based on
+``config['datasource']``.
+
+Config block expected under the top-level ``synthetic`` key (all optional):
+
+    synthetic:
+      spike_rate_hz: 20         # per ntrode, Poisson
+      mark_dim: 8               # must match encoder.mark_dim
+      mark_amplitude_uv: 80     # mean spike amplitude
+      track_length_cm: 200      # walk distance
+      walk_speed_cm_s: 15       # synthetic animal speed
+      startup_delay_s: 1.0      # delay before firing the startup callback
+      run_duration_s: 60        # auto-terminate after this long
+      voltage_scaling_factor: 0.195
+"""
+
+import time
+
+import numpy as np
+
+from realtime_decoder import utils
+from realtime_decoder.base import DataSourceReceiver
+from realtime_decoder.datatypes import (
+    Datatypes,
+    LFPPoint,
+    SpikePoint,
+    CameraModulePoint,
+)
+
+
+_DEFAULTS = {
+    'spike_rate_hz': 20.0,
+    'mark_dim': 4,
+    'mark_amplitude_uv': 80.0,
+    'track_length_cm': 200.0,
+    'walk_speed_cm_s': 15.0,
+    'startup_delay_s': 1.0,
+    'run_duration_s': 60.0,
+    'voltage_scaling_factor': 1.0,
+}
+
+
+def _params(config):
+    """Read the ``synthetic`` config block, applying defaults."""
+    p = dict(_DEFAULTS)
+    p.update(config.get('synthetic') or {})
+    return p
+
+
+class SyntheticDataReceiver(DataSourceReceiver):
+    """Drop-in synthetic replacement for ``trodesnet.TrodesDataReceiver``.
+
+    Generates LFP / spike / position samples on demand at clock-driven
+    rates. ``__next__`` returns None when no sample is due yet, matching
+    the non-blocking semantics of the Trodes receiver — the polling main
+    loops do not need to know they are reading synthetic data.
+    """
+
+    def __init__(self, comm, rank, config, datatype):
+        if datatype not in (
+            Datatypes.LFP,
+            Datatypes.SPIKES,
+            Datatypes.LINEAR_POSITION,
+        ):
+            raise TypeError(f"Invalid datatype {datatype}")
+        super().__init__(comm, rank, config, datatype)
+
+        self._p = _params(config)
+        self._started = False
+        self._stopped = False
+
+        self.ntrode_ids = []
+
+        # Per-stream pacing: we advance a deterministic virtual clock
+        # (sample index) from t=0 at activate(), and emit samples as
+        # wall-clock catches up. This gives roughly the same delivery
+        # cadence as a live acquisition system at the configured rates.
+        self._t0_wall = None
+        self._next_sample_idx = 0
+        if datatype == Datatypes.LFP:
+            self._fs = config['sampling_rate']['lfp']
+        elif datatype == Datatypes.SPIKES:
+            self._fs = config['sampling_rate']['spikes']
+        else:  # LINEAR_POSITION
+            self._fs = config['sampling_rate']['position']
+
+        self._spike_clock = config['sampling_rate']['spikes']
+
+        # Spike-stream specific: independent Poisson process per ntrode.
+        # ``_next_spike_sample[ntid]`` stores the spike-clock sample index
+        # at which that ntrode's next spike will fire.
+        self._rng = np.random.default_rng(seed=rank * 1009 + int(datatype))
+        self._next_spike_sample = {}
+        self._mark_dim = self._p['mark_dim']
+        self._amp = self._p['mark_amplitude_uv']
+
+    # ------------------------------------------------------------------
+    # DataSourceReceiver contract
+    # ------------------------------------------------------------------
+
+    def register_datatype_channel(self, channel):
+        ntrode_id = int(channel)
+        if self.datatype in (Datatypes.LFP, Datatypes.SPIKES):
+            if ntrode_id not in self.ntrode_ids:
+                self.ntrode_ids.append(ntrode_id)
+        # position has no channels
+
+    def activate(self):
+        self._t0_wall = time.time()
+        self._next_sample_idx = 0
+        if self.datatype == Datatypes.SPIKES:
+            for ntid in self.ntrode_ids:
+                self._schedule_next_spike(ntid, sample_now=0)
+        self._started = True
+        self.class_log.debug(
+            f"Synthetic {self.datatype.name} datastream activated "
+            f"({len(self.ntrode_ids)} ntrodes)"
+        )
+
+    def deactivate(self):
+        self._started = False
+
+    def stop_iterator(self):
+        raise StopIteration()
+
+    def __next__(self):
+        if not self._started:
+            return None
+
+        elapsed = time.time() - self._t0_wall
+        if (
+            self._p['run_duration_s'] > 0
+            and elapsed > self._p['run_duration_s']
+            and not self._stopped
+        ):
+            # one-time log; the supervisor's termination is wired
+            # through SyntheticClient.receive() below.
+            self._stopped = True
+
+        if self.datatype == Datatypes.LFP:
+            return self._next_lfp(elapsed)
+        elif self.datatype == Datatypes.SPIKES:
+            return self._next_spike(elapsed)
+        else:
+            return self._next_position(elapsed)
+
+    # ------------------------------------------------------------------
+    # Per-datatype generators
+    # ------------------------------------------------------------------
+
+    def _next_lfp(self, elapsed):
+        target_idx = int(elapsed * self._fs)
+        if target_idx < self._next_sample_idx:
+            return None
+        idx = self._next_sample_idx
+        self._next_sample_idx += 1
+        # white-ish noise sized to (num_channels,), scaled the same way
+        # TrodesDataReceiver does (raw * voltage_scaling_factor)
+        n = max(1, len(self.ntrode_ids))
+        raw = self._rng.standard_normal(n) * 200.0  # ~uV range pre-scale
+        data = raw * self._p['voltage_scaling_factor']
+        local_ts = idx  # LFP uses spike-clock timestamps in real Trodes;
+        # at fs_lfp=1500, fs_spike=30000 the ratio is 20, but downstream
+        # only cares about monotonicity within a stream, so use idx.
+        system_ts = time.time_ns()
+        return LFPPoint(
+            local_ts,
+            list(self.ntrode_ids),
+            data,
+            system_ts,
+            time.time_ns(),
+        )
+
+    def _next_spike(self, elapsed):
+        if not self.ntrode_ids:
+            return None
+        spike_sample_now = int(elapsed * self._spike_clock)
+        # Find any ntrode whose next-spike sample has arrived.
+        for ntid in self.ntrode_ids:
+            if self._next_spike_sample[ntid] <= spike_sample_now:
+                ts = self._next_spike_sample[ntid]
+                self._schedule_next_spike(ntid, sample_now=spike_sample_now)
+                # mark vector: gaussian around _amp, all channels positive
+                samples = (
+                    self._rng.standard_normal(self._mark_dim) * 8.0 + self._amp
+                ) / self._p['voltage_scaling_factor']
+                # SpikePoint.data is later multiplied by voltage_scaling_factor
+                # in real Trodes; the encoder reads `max(mark_vec)` so we just
+                # need the post-scaling magnitudes to clear `encoder.spk_amp`.
+                return SpikePoint(
+                    ts,
+                    ntid,
+                    samples * self._p['voltage_scaling_factor'],
+                    time.time_ns(),
+                    time.time_ns(),
+                )
+        return None
+
+    def _next_position(self, elapsed):
+        target_idx = int(elapsed * self._fs)
+        if target_idx < self._next_sample_idx:
+            return None
+        idx = self._next_sample_idx
+        self._next_sample_idx += 1
+
+        # triangle-wave walk along a single linear segment between 0 and
+        # track_length_cm
+        L = self._p['track_length_cm']
+        v = self._p['walk_speed_cm_s']
+        t = elapsed
+        period = 2.0 * L / max(v, 1e-6)
+        phase = (t % period) / period  # 0..1
+        pos_cm = L * (1.0 - abs(2.0 * phase - 1.0))
+        # x/y/x2/y2 in "pixel" units — kinematics.scale_factor converts back
+        sf = self.config['kinematics']['scale_factor']
+        x = pos_cm / sf
+        y = 100.0  # constant
+        x2 = x + 5.0
+        y2 = y
+        return CameraModulePoint(
+            idx,
+            segment=0,
+            position=pos_cm,
+            x=x,
+            y=y,
+            x2=x2,
+            y2=y2,
+            t_recv_data=time.time_ns(),
+        )
+
+    # ------------------------------------------------------------------
+    # internals
+    # ------------------------------------------------------------------
+
+    def _schedule_next_spike(self, ntid, *, sample_now):
+        rate = max(self._p['spike_rate_hz'], 1e-6)
+        # exponential inter-arrival in seconds → samples
+        gap_s = self._rng.exponential(1.0 / rate)
+        gap_samples = max(1, int(gap_s * self._spike_clock))
+        self._next_spike_sample[ntid] = sample_now + gap_samples
+
+
+class SyntheticClient(object):
+    """Drop-in synthetic replacement for ``trodesnet.TrodesClient``.
+
+    Exposes the same surface used by the supervisor and stim decider:
+      * ``set_startup_callback`` / ``set_termination_callback``
+      * ``receive`` (called from the supervisor main loop)
+      * ``send_statescript_shortcut_message`` (called from stim_decider)
+
+    ``receive`` fires the startup callback once after ``startup_delay_s``
+    of wall clock has elapsed, and fires termination once ``run_duration_s``
+    has elapsed.
+    """
+
+    def __init__(self, config):
+        self._startup_callback = utils.nop
+        self._termination_callback = utils.nop
+        self._p = _params(config)
+        self._t0_wall = time.time()
+        self._started = False
+        self._terminated = False
+        # log-only buffer of "shortcut messages" the stim decider would
+        # have sent to ECU; useful for asserting in tests later.
+        self.sent_shortcuts = []
+
+    def receive(self):
+        elapsed = time.time() - self._t0_wall
+        if not self._started and elapsed >= self._p['startup_delay_s']:
+            self._started = True
+            self._startup_callback()
+        if (
+            self._started
+            and not self._terminated
+            and self._p['run_duration_s'] > 0
+            and elapsed >= self._p['run_duration_s'] + self._p['startup_delay_s']
+        ):
+            self._terminated = True
+            self._termination_callback()
+
+    def send_statescript_shortcut_message(self, val):
+        self.sent_shortcuts.append((time.time_ns(), int(val)))
+
+    def set_startup_callback(self, callback):
+        self._startup_callback = callback
+
+    def set_termination_callback(self, callback):
+        self._termination_callback = callback
diff --git a/runscript.py b/runscript.py
index 3e920c8..5b93548 100644
--- a/runscript.py
+++ b/runscript.py
@@ -11,12 +11,30 @@
 from mpi4py import MPI
 
 from realtime_decoder import (
-    datatypes, position, trodesnet, stimulation,
+    datatypes, position, trodesnet, synthetic, stimulation,
     main_process, ripple_process, encoder_process,
     decoder_process, gui_process, base, messages,
     merge_rec
 )
 
+
+def _data_source_factory(config):
+    """Pick the (receiver_class, client_class) pair for the configured
+    data source.
+
+    `datasource: trodes` (default) uses the live Trodes streams.
+    `datasource: synthetic` uses the in-process generator from
+    `realtime_decoder.synthetic` — install-and-run with no hardware.
+    """
+    ds = config.get('datasource', 'trodes')
+    if ds == 'trodes':
+        return trodesnet.TrodesDataReceiver, trodesnet.TrodesClient
+    if ds == 'synthetic':
+        return synthetic.SyntheticDataReceiver, synthetic.SyntheticClient
+    raise ValueError(
+        f"Unknown datasource {ds!r}; expected 'trodes' or 'synthetic'"
+    )
+
 # from line_profiler import LineProfiler
 
 class GuiProcessStub(base.RealtimeProcess, base.MessageHandler):
@@ -169,21 +187,23 @@ def setup(config_path, numprocs):
     regloop = True
     #################################################
     
+    DataReceiver, Client = _data_source_factory(config)
+
     if rank in config['rank']['supervisor']:
-        trodes_client = trodesnet.TrodesClient(config)
+        net_client = Client(config)
         stim_decider = stimulation.TwoArmTrodesStimDecider(
-            comm, rank, config, trodes_client
+            comm, rank, config, net_client
         )
         process = main_process.MainProcess(
-            comm, rank, config, stim_decider, trodes_client
+            comm, rank, config, stim_decider, net_client
         )
-        trodes_client.set_startup_callback(process.startup)
-        trodes_client.set_termination_callback(process.trigger_termination)
+        net_client.set_startup_callback(process.startup)
+        net_client.set_termination_callback(process.trigger_termination)
     elif rank in config['rank']['ripples']:
-        lfp_interface = trodesnet.TrodesDataReceiver(
+        lfp_interface = DataReceiver(
             comm, rank, config, datatypes.Datatypes.LFP
         )
-        pos_interface = trodesnet.TrodesDataReceiver(
+        pos_interface = DataReceiver(
             comm, rank, config, datatypes.Datatypes.LINEAR_POSITION
         )
         process = ripple_process.RippleProcess(
@@ -196,10 +216,10 @@ def setup(config_path, numprocs):
         # prof.print_stats()
         # regloop = False
     elif rank in config['rank']['encoders']:
-        spikes_interface = trodesnet.TrodesDataReceiver(
+        spikes_interface = DataReceiver(
             comm, rank, config, datatypes.Datatypes.SPIKES
         )
-        pos_interface = trodesnet.TrodesDataReceiver(
+        pos_interface = DataReceiver(
             comm, rank, config, datatypes.Datatypes.LINEAR_POSITION
         )
         pos_mapper = position.TrodesPositionMapper(
@@ -211,7 +231,7 @@ def setup(config_path, numprocs):
             pos_mapper
         )
     elif rank in config['rank']['decoders']:
-        pos_interface = trodesnet.TrodesDataReceiver(
+        pos_interface = DataReceiver(
             comm, rank, config, datatypes.Datatypes.LINEAR_POSITION
         )
         pos_mapper = position.TrodesPositionMapper(