Source code for ezmsg.simbiophys.system.velocity2ecephys

"""Convert 2D cursor velocity to simulated extracellular electrophysiology.

This module provides a complete system that encodes cursor velocity into
realistic ecephys signals containing both spike waveforms and LFP-like
background activity.

Pipeline::

    velocity (x,y) -> CART2POL --+--> Velocity2Spike --> spikes --|
                                 |                                +--> Add --> ecephys
                                 +--> Velocity2LFP ----> lfp -----|

The coordinate transformation from Cartesian to polar is done once at the
input, then shared by both spike and LFP encoding branches.

This is the top-level system for velocity-encoded neural simulation. Use this
when you need full ecephys-like output suitable for testing BCI decoders.

See Also:
    :mod:`ezmsg.simbiophys.system.velocity2spike`: Spike-only encoding.
    :mod:`ezmsg.simbiophys.system.velocity2lfp`: LFP-only encoding.
"""

import ezmsg.core as ez
from ezmsg.sigproc.coordinatespaces import CoordinateMode, CoordinateSpaces, CoordinateSpacesSettings
from ezmsg.sigproc.math.add import Add
from ezmsg.util.messages.axisarray import AxisArray

from ..line_noise import LineNoiseSettings, LineNoiseUnit
from .velocity2lfp import Velocity2LFP, Velocity2LFPSettings
from .velocity2spike import Velocity2Spike, Velocity2SpikeSettings


[docs] class VelocityEncoderSettings(ez.Settings): """Settings for :obj:`VelocityEncoder`.""" output_fs: float = 30_000.0 """Output sampling rate in Hz.""" output_ch: int = 256 """Number of output channels (simulated electrodes).""" seed: int = 6767 """Random seed for reproducible spike and LFP generation.""" # Spike branch tuning. baseline_rate: float = 10 """Baseline firing rate in Hz""" modulation_depth: float = 20 / 314 """Directional modulation depth in Hz per (pixel/second)""" # LFP branch tuning. max_velocity: float = 315 """Velocity (px/s) at which LFP beta reaches full modulation.""" n_sources: int = 8 """Number of cosine-encoded LFP sources.""" drift_scale: float = 4.0 """Amplitude of always-on slow 1/f drift added to each LFP source before mixing (a shared low-frequency field drift across channels). Velocity- independent, so baseline wander is present even at rest. Set to 0 to disable.""" line_noise_freq: float | None = None """Mains line-noise frequency in Hz (50.0 or 60.0). None (default) disables line noise (pass-through). Added to every channel of the final signal.""" line_noise_amp: float = 10.0 """Amplitude of the line-noise sinusoid (same units as the output signal).""" line_noise_drift_rate: float = 0.002 """Line-noise frequency drift rate in Hz per second (recording-clock drift).""" line_noise_drift_bound: float = 1.5 """Line-noise frequency is clamped to ``line_noise_freq +/- this`` Hz."""
[docs] class VelocityEncoder(ez.Collection): """Encode cursor velocity into simulated extracellular electrophysiology. This system combines spike and LFP encoding to produce realistic ecephys signals. It runs two parallel pipelines: 1. **Spike branch** (:obj:`Velocity2Spike`): Generates cosine-tuned spike waveforms based on velocity direction and magnitude. 2. **LFP branch** (:obj:`Velocity2LFP`): Generates velocity-modulated colored noise representing local field potentials. The outputs are summed to produce the final ecephys signal. Input: AxisArray with shape (N, 2) containing cursor velocity in pixels/second. Dimension 0 is time, dimension 1 is [vx, vy]. Output: AxisArray with shape (M, output_ch) containing combined spike and LFP signals at output_fs sampling rate. Example: >>> encoder = VelocityEncoder(VelocityEncoderSettings( ... output_fs=30_000.0, ... output_ch=256, ... seed=42, ... )) """ SETTINGS = VelocityEncoderSettings # Velocity inputs (via mouse / gamepad system, or via task parsing system) INPUT_SIGNAL = ez.InputTopic(AxisArray) COORDS = CoordinateSpaces() # Cartesian to polar (done once, shared by both branches) SPIKES = Velocity2Spike() LFP = Velocity2LFP() ADD = Add() # Add colored noise and waveforms LINE_NOISE = LineNoiseUnit() # Optional mains pickup on the combined signal OUTPUT_SIGNAL = ez.OutputTopic(AxisArray)
[docs] def configure(self) -> None: self.COORDS.apply_settings(CoordinateSpacesSettings(mode=CoordinateMode.CART2POL, axis="ch")) self.SPIKES.apply_settings( Velocity2SpikeSettings( output_fs=self.SETTINGS.output_fs, output_ch=self.SETTINGS.output_ch, baseline_rate=self.SETTINGS.baseline_rate, modulation_depth=self.SETTINGS.modulation_depth, seed=self.SETTINGS.seed, ) ) self.LFP.apply_settings( Velocity2LFPSettings( output_fs=self.SETTINGS.output_fs, output_ch=self.SETTINGS.output_ch, n_lfp_sources=self.SETTINGS.n_sources, max_velocity=self.SETTINGS.max_velocity, drift_scale=self.SETTINGS.drift_scale, seed=self.SETTINGS.seed, ) ) self.LINE_NOISE.apply_settings( LineNoiseSettings( freq=self.SETTINGS.line_noise_freq, amp=self.SETTINGS.line_noise_amp, drift_rate=self.SETTINGS.line_noise_drift_rate, drift_bound=self.SETTINGS.line_noise_drift_bound, seed=self.SETTINGS.seed, ) )
[docs] def network(self) -> ez.NetworkDefinition: return ( (self.INPUT_SIGNAL, self.COORDS.INPUT_SIGNAL), (self.COORDS.OUTPUT_SIGNAL, self.SPIKES.INPUT_SIGNAL), (self.SPIKES.OUTPUT_SIGNAL, self.ADD.INPUT_SIGNAL_A), (self.COORDS.OUTPUT_SIGNAL, self.LFP.INPUT_SIGNAL), (self.LFP.OUTPUT_SIGNAL, self.ADD.INPUT_SIGNAL_B), (self.ADD.OUTPUT_SIGNAL, self.LINE_NOISE.INPUT_SIGNAL), (self.LINE_NOISE.OUTPUT_SIGNAL, self.OUTPUT_SIGNAL), )