ezmsg.simbiophys#

ezmsg-simbiophys: Signal simulation and synthesis for ezmsg.

class Clock(*args, settings=None, **kwargs)[source]#

Bases: BaseProducerUnit[ClockSettings, LinearAxis, ClockProducer]

Clock unit that produces ticks at a specified rate.

Output is a AxisArray.LinearAxis with: - gain: 1/dispatch_rate (seconds per tick) - offset: Wall clock timestamp

Parameters:

settings (Settings | None)

SETTINGS#

alias of ClockSettings

class ClockProducer(*args, **kwargs)[source]#

Bases: BaseStatefulProducer[ClockSettings, LinearAxis, ClockState]

Produces clock ticks at a specified rate.

Each tick outputs a AxisArray.LinearAxis containing: - gain: 1/dispatch_rate (seconds per tick), or 0.0 if dispatch_rate is infinite - offset: Wall clock timestamp (time.monotonic)

This output type allows downstream components (like Counter) to know both the timing of the tick and the nominal dispatch rate.

class ClockSettings(dispatch_rate=inf)[source]#

Bases: Settings

Settings for ClockProducer.

Parameters:

dispatch_rate (float)

__init__(dispatch_rate=inf)#
Parameters:

dispatch_rate (float)

Return type:

None

dispatch_rate: float = inf#

Dispatch rate in Hz. - Finite value (e.g., 100.0): Dispatch 100 times per second - math.inf: Dispatch as fast as possible (no sleep)

class ClockState[source]#

Bases: object

State for ClockProducer.

n_dispatch: int = 0#

Number of dispatches since reset.

t_0: float#

Start time (monotonic).

class Counter(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[CounterSettings, CounterTransformer]

Transforms clock ticks into monotonically increasing counter values as AxisArray.

Receives timing from INPUT_CLOCK (LinearAxis from Clock) and outputs AxisArray.

Parameters:

settings (Settings | None)

SETTINGS#

alias of CounterSettings

class CounterSettings(fs, n_time=None, mod=None)[source]#

Bases: ClockDrivenSettings

Settings for Counter and CounterTransformer.

Parameters:
__init__(fs, n_time=None, mod=None)#
Parameters:
Return type:

None

mod: int | None = None#

If set, counter values rollover at this modulus.

class CounterTransformer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[CounterSettings, CounterTransformerState]

Transforms clock ticks (LinearAxis) into AxisArray counter values.

Each clock tick produces a block of counter values. The block size is either fixed (n_time setting) or derived from the clock’s gain (fs * gain).

class CounterTransformerState[source]#

Bases: ClockDrivenState

State for CounterTransformer.

template: AxisArray | None = None#
class SinGenerator(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[SinGeneratorSettings, SinProducer]

Generates sinusoidal waveforms synchronized to clock ticks.

Receives timing from INPUT_CLOCK (LinearAxis from Clock) and outputs sinusoidal AxisArray on OUTPUT_SIGNAL.

Parameters:

settings (Settings | None)

SETTINGS#

alias of SinGeneratorSettings

class SinGeneratorSettings(fs, n_time=None, n_ch=1, freq=1.0, amp=1.0, phase=0.0)[source]#

Bases: ClockDrivenSettings

Settings for SinGenerator.

Parameters:
__init__(fs, n_time=None, n_ch=1, freq=1.0, amp=1.0, phase=0.0)#
Parameters:
Return type:

None

amp: Union[float, Buffer, _SupportsArray[dtype[Any]], _NestedSequence[_SupportsArray[dtype[Any]]], complex, bytes, str, _NestedSequence[complex | bytes | str]] = 1.0#

The amplitude of the sinusoid. Scalar or per-channel array.

freq: Union[float, Buffer, _SupportsArray[dtype[Any]], _NestedSequence[_SupportsArray[dtype[Any]]], complex, bytes, str, _NestedSequence[complex | bytes | str]] = 1.0#

The frequency of the sinusoid, in Hz. Scalar or per-channel array.

n_ch: int = 1#

Number of channels to output.

phase: Union[float, Buffer, _SupportsArray[dtype[Any]], _NestedSequence[_SupportsArray[dtype[Any]]], complex, bytes, str, _NestedSequence[complex | bytes | str]] = 0.0#

The initial phase of the sinusoid, in radians. Scalar or per-channel array.

class SinGeneratorState[source]#

Bases: ClockDrivenState

State for SinGenerator.

amp: ndarray | None = None#
ang_freq: ndarray | None = None#
phase: ndarray | None = None#
template: AxisArray | None = None#
class SinProducer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[SinGeneratorSettings, SinGeneratorState]

Generates sinusoidal waveforms synchronized to clock ticks.

Each clock tick produces a block of sinusoidal data based on the sample rate (fs) and chunk size (n_time) settings.

class SpiralGenerator(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[SpiralGeneratorSettings, SpiralProducer]

Generates 2D spiral motion synchronized to clock ticks.

Receives timing from INPUT_CLOCK (LinearAxis from Clock) and outputs 2D position AxisArray (x, y) on OUTPUT_SIGNAL.

The spiral pattern has both radius and angle varying over time: - Radius oscillates sinusoidally (breathing in/out) - Angle increases linearly (rotation)

Parameters:

settings (Settings | None)

SETTINGS#

alias of SpiralGeneratorSettings

class SpiralGeneratorSettings(fs, n_time=None, r_mean=150.0, r_amp=50.0, radial_freq=0.1, radial_phase=0.0, angular_freq=0.25, angular_phase=0.0)[source]#

Bases: ClockDrivenSettings

Settings for SpiralGenerator.

Generates 2D position (x, y) following a spiral pattern where both the radius and angle change over time.

The parametric equations are:

r(t) = r_mean + r_amp * sin(2*π*radial_freq*t + radial_phase) θ(t) = 2*π*angular_freq*t + angular_phase x(t) = r(t) * cos(θ(t)) y(t) = r(t) * sin(θ(t))

Parameters:
__init__(fs, n_time=None, r_mean=150.0, r_amp=50.0, radial_freq=0.1, radial_phase=0.0, angular_freq=0.25, angular_phase=0.0)#
Parameters:
Return type:

None

angular_freq: float = 0.25#

Frequency of the angular rotation in Hz.

angular_phase: float = 0.0#

Initial angular phase in radians.

r_amp: float = 50.0#

Amplitude of the radial oscillation.

r_mean: float = 150.0#

Mean radius of the spiral.

radial_freq: float = 0.1#

Frequency of the radial oscillation in Hz.

radial_phase: float = 0.0#

Initial phase of the radial oscillation in radians.

class SpiralGeneratorState[source]#

Bases: ClockDrivenState

State for SpiralGenerator.

template: AxisArray | None = None#
class SpiralProducer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[SpiralGeneratorSettings, SpiralGeneratorState]

Generates spiral motion synchronized to clock ticks.

Each clock tick produces a block of 2D position data (x, y) following a spiral pattern where both radius and angle change over time.

class PinkNoise(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[PinkNoiseSettings, PinkNoiseProducer]

Generates pink (1/f) noise synchronized to clock ticks.

Receives timing from INPUT_CLOCK (LinearAxis from Clock) and outputs pink noise AxisArray on OUTPUT_SIGNAL.

Parameters:

settings (Settings | None)

SETTINGS#

alias of PinkNoiseSettings

class PinkNoiseProducer(*args, **kwargs)[source]#

Bases: CompositeProcessor[PinkNoiseSettings, LinearAxis, AxisArray]

Generates pink (1/f) noise synchronized to clock ticks.

Pink noise is generated by filtering white noise with a first-order lowpass Butterworth filter.

class PinkNoiseSettings(fs, n_time=None, n_ch=1, loc=0.0, scale=1.0, cutoff=300.0)[source]#

Bases: WhiteNoiseSettings

Settings for pink noise generator.

Parameters:
__init__(fs, n_time=None, n_ch=1, loc=0.0, scale=1.0, cutoff=300.0)#
Parameters:
Return type:

None

cutoff: float = 300.0#

Lowpass cutoff frequency (Hz) for the first-order Butterworth filter that creates the 1/f characteristic.

class WhiteNoise(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[WhiteNoiseSettings, WhiteNoiseProducer]

Generates white noise synchronized to clock ticks.

Receives timing from INPUT_CLOCK (LinearAxis from Clock) and outputs white noise AxisArray on OUTPUT_SIGNAL.

Parameters:

settings (Settings | None)

SETTINGS#

alias of WhiteNoiseSettings

class WhiteNoiseProducer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[WhiteNoiseSettings, WhiteNoiseState]

Generates white noise synchronized to clock ticks.

Each clock tick produces a block of Gaussian white noise based on the sample rate (fs) and chunk size (n_time) settings.

class WhiteNoiseSettings(fs, n_time=None, n_ch=1, loc=0.0, scale=1.0)[source]#

Bases: ClockDrivenSettings

Settings for white noise generators.

Parameters:
__init__(fs, n_time=None, n_ch=1, loc=0.0, scale=1.0)#
Parameters:
Return type:

None

loc: float = 0.0#

DC offset (mean of the distribution).

n_ch: int = 1#

Number of channels to output.

scale: float = 1.0#

Scale (standard deviation of the distribution).

class WhiteNoiseState[source]#

Bases: ClockDrivenState

State for WhiteNoiseProducer.

template: AxisArray | None = None#
class EEGSynth(*args, settings=None, **kwargs)[source]#

Bases: Collection

A Collection that generates synthetic EEG signals.

Combines white noise with alpha oscillations using a diamond flow: Clock -> {Noise, Oscillator} -> Add -> Output

Network flow:

Clock -> {Noise, Oscillator} Noise -> Add.A Oscillator -> Add.B Add -> OUTPUT

Parameters:

settings (Settings | None)

ADD = <ezmsg.sigproc.math.add.Add object>#
CLOCK = <ezmsg.baseproc.clock.Clock object>#
NOISE = <ezmsg.simbiophys.noise.WhiteNoise object>#
OSC = <ezmsg.simbiophys.oscillator.SinGenerator object>#
OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#
SETTINGS#

alias of EEGSynthSettings

configure()[source]#

A lifecycle hook that runs when the Collection is instantiated. This is the best place to call Unit.apply_settings() on each member Unit of the Collection.

Return type:

None

network()[source]#

Override this method and have the definition return a NetworkDefinition which defines how InputStream and OutputStream from member Unit s will be connected.

Return type:

Iterable[Tuple[Union[Stream, str], Union[Stream, str]]]

class EEGSynthSettings(fs=500.0, n_time=100, alpha_freq=10.5, n_ch=8)[source]#

Bases: Settings

Settings for EEG synthesizer.

Parameters:
__init__(fs=500.0, n_time=100, alpha_freq=10.5, n_ch=8)#
Parameters:
Return type:

None

alpha_freq: float = 10.5#

Alpha frequency in Hz.

fs: float = 500.0#

Sample rate in Hz.

n_ch: int = 8#

Number of channels.

n_time: int = 100#

Number of samples per block.

class CosineEncoderSettings(model_file=None, output_ch=10, baseline=0.0, modulation=1.0, speed_modulation=0.0, seed=None)[source]#

Bases: Settings

Settings for CosineEncoder.

Either model_file OR the random generation parameters should be specified. If model_file is provided, parameters are loaded from file. Otherwise, parameters are randomly generated.

Parameters:

  • model_file (str | None)

  • output_ch (int)

  • baseline (float)

  • modulation (float)

  • speed_modulation (float)

  • seed (int | None)

__init__(model_file=None, output_ch=10, baseline=0.0, modulation=1.0, speed_modulation=0.0, seed=None)#
Parameters:

  • model_file (str | None)

  • output_ch (int)

  • baseline (float)

  • modulation (float)

  • speed_modulation (float)

  • seed (int | None)

Return type:

None

baseline: float = 0.0#

Baseline output value for all channels (used if model_file is None).

model_file: str | None = None#

Path to .npz file with encoder parameters (baseline, modulation, pd, speed_modulation). Also supports legacy neural tuning files with keys (b0, m, pd, bs).

modulation: float = 1.0#

Directional modulation depth for all channels (used if model_file is None).

output_ch: int = 10#

Number of output channels (used if model_file is None).

seed: int | None = None#

Random seed for reproducibility of preferred directions (used if model_file is None).

speed_modulation: float = 0.0#

Speed modulation (non-directional) for all channels (used if model_file is None).

class CosineEncoderState[source]#

Bases: object

State for cosine encoder transformer.

Holds the per-channel encoding parameters. All arrays have shape (1, output_ch) for efficient broadcasting during processing.

baseline#

Baseline output value for each channel.

modulation#

Directional modulation depth for each channel.

pd#

Preferred direction (radians) for each channel.

speed_modulation#

Speed modulation (non-directional) for each channel.

ch_axis#

Pre-built channel axis for output messages.

baseline: ndarray[tuple[Any, ...], dtype[floating]] | None = None#
ch_axis: CoordinateAxis | None = None#
init_random(output_ch, baseline=0.0, modulation=1.0, speed_modulation=0.0, seed=None)[source]#

Initialize encoder parameters with random preferred directions.

Parameters:

  • output_ch (int) – Number of output channels.

  • baseline (float) – Baseline value for all channels.

  • modulation (float) – Directional modulation depth for all channels.

  • speed_modulation (float) – Speed modulation (non-directional) for all channels.

  • seed (int | None) – Random seed for reproducibility.

Return type:

None

load_from_file(filepath, output_ch=None)[source]#

Load parameters from a .npz file.

The file should contain arrays with keys matching the parameter names. For backwards compatibility with neural tuning files, the following key mappings are supported:

  • ‘b0’ -> baseline

  • ‘m’ -> modulation

  • ‘pd’ -> pd (preferred direction)

  • ‘bs’ -> speed_modulation

Parameters:

  • filepath (str | Path) – Path to .npz file containing parameter arrays.

  • output_ch (int | None) – Number of channels to use. If None, uses all in file.

Return type:

None

modulation: ndarray[tuple[Any, ...], dtype[floating]] | None = None#
property output_ch: int#

Number of output channels.

pd: ndarray[tuple[Any, ...], dtype[floating]] | None = None#
speed_modulation: ndarray[tuple[Any, ...], dtype[floating]] | None = None#
validate()[source]#

Validate that all parameters have consistent shapes.

Return type:

None

class CosineEncoderTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[CosineEncoderSettings, AxisArray, AxisArray, CosineEncoderState]

Transform polar coordinates to multi-channel encoded output.

Input: AxisArray with shape (n_samples, 2) containing polar coordinates

(magnitude, angle) where magnitude is speed and angle is direction.

Output: AxisArray with shape (n_samples, output_ch) containing encoded values.

The encoding formula is:
output = baseline + modulation * magnitude * cos(angle - pd)

  • speed_modulation * magnitude

This is a generic encoder suitable for various applications including: - Neural firing rate encoding (baseline=10Hz, modulation=20Hz) - LFP spectral parameter modulation (baseline=1.0, modulation=0.5) - Any other cosine-tuning based encoding

class CosineEncoderUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[CosineEncoderSettings, AxisArray, AxisArray, CosineEncoderTransformer]

Unit wrapper for CosineEncoderTransformer.

Parameters:

settings (Settings | None)

SETTINGS#

alias of CosineEncoderSettings

class ColoredNoiseFilterState(delay_line, coeffs)[source]#

Bases: object

State for a single channel’s colored noise filter.

Parameters:
__init__(delay_line, coeffs)#
Parameters:
Return type:

None

delay_line: ndarray[tuple[Any, ...], dtype[float64]]#

Previous output samples (filter memory).

coeffs: ndarray[tuple[Any, ...], dtype[float64]]#

Current filter coefficients (exponentially smoothed).

class DynamicColoredNoiseSettings(output_fs=None, n_poles=5, smoothing_tau=0.01, initial_beta=1.0, scale=1.0, seed=None)[source]#

Bases: Settings

Parameters:
__init__(output_fs=None, n_poles=5, smoothing_tau=0.01, initial_beta=1.0, scale=1.0, seed=None)#
Parameters:
Return type:

None

initial_beta: float = 1.0#

Initial spectral exponent before any input is received.

n_poles: int = 5#

Number of IIR filter poles. More poles extend accuracy to lower frequencies. Default 5 provides good balance.

output_fs: float | None = None#

Output sampling rate in Hz. If None, output rate matches input rate.

scale: float = 1.0#

Output amplitude scaling factor.

seed: int | None = None#

Random seed for reproducibility. If None, uses system entropy.

smoothing_tau: float = 0.01#

Time constant (in seconds) for exponential smoothing of coefficient changes. Coefficients reach ~63% of target in τ seconds, ~95% in 3τ seconds. Set to 0 for instantaneous changes (no smoothing). Default 0.01 (10ms) provides smooth transitions without sluggishness.

class DynamicColoredNoiseState[source]#

Bases: object

alpha: float = 1.0#

Exponential smoothing factor for coefficient updates.

output_gain: float = 0.0#

Time step between output samples (1/output_fs).

rng: Generator | None = None#

Random number generator.

sample_remainder: float = 0.0#

Fractional sample accumulator for resampling.

samples_per_bin: float = 1.0#

Number of output samples per input sample.

filter_states: list[ColoredNoiseFilterState]#

Per-channel filter states.

class DynamicColoredNoiseTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[DynamicColoredNoiseSettings, AxisArray, AxisArray, DynamicColoredNoiseState]

Transform spectral exponent (β) input into colored noise output.

Input: AxisArray with β values. Shape can be:

  • (n_samples,) for single-channel output

  • (n_samples, n_channels) for multi-channel output

Output: AxisArray with colored noise having spectral density ~ 1/f^β.

If output_fs is set, output will be resampled to that rate.

The transformer maintains filter state across chunks to ensure continuity (no discontinuities at chunk boundaries). When β changes, coefficients are exponentially smoothed with time constant smoothing_tau to avoid transients.

Each input β sample is treated as a “bin” - all output samples generated within that bin use that β value as the target. Coefficient smoothing handles transitions between bins.

Example

>>> settings = DynamicColoredNoiseSettings(
...     output_fs=30000.0,  # 30 kHz output
...     n_poles=5,
...     smoothing_tau=0.01,  # 10ms time constant
...     initial_beta=1.0
... )
>>> transformer = DynamicColoredNoiseTransformer(settings)
>>> # Input: β values at 100 Hz
>>> beta_input = AxisArray(np.full((100, 2), 1.5), dims=["time", "ch"], ...)
>>> noise_output = transformer(beta_input)  # Output at 30 kHz
class DynamicColoredNoiseUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[DynamicColoredNoiseSettings, AxisArray, AxisArray, DynamicColoredNoiseTransformer]

Unit wrapper for DynamicColoredNoiseTransformer.

Parameters:

settings (Settings | None)

SETTINGS#

alias of DynamicColoredNoiseSettings

compute_kasdin_coefficients(beta, n_poles)[source]#

Compute IIR filter coefficients for 1/f^β noise using Kasdin’s method.

The coefficients are computed using the recurrence relation:

a₀ = 1 aₖ = (k - 1 - β/2) · aₖ₋₁ / k for k = 1, 2, …, n_poles

Parameters:

  • beta (float) – Spectral exponent. Common values: - 0: white noise - 1: pink noise (1/f) - 2: brown/red noise (1/f²) TODO: jit / vectorize over multiple values of beta

  • n_poles (int) – Number of IIR filter poles. More poles extend accuracy to lower frequencies but increase computation. 5 is a reasonable default.

Return type:

ndarray[tuple[Any, ...], dtype[float64]]

Returns:

Array of shape (n_poles,) containing filter coefficients a₁ through aₙ.

class DNSSLFPProducer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[DNSSLFPSettings, DNSSLFPState]

Produces DNSS LFP signal synchronized to clock ticks.

Each clock tick produces a block of LFP data based on the sample rate (fs) and chunk size (n_time) settings. All channels receive identical LFP values.

class DNSSLFPSettings(fs=30000, n_time=None, n_ch=256, pattern='spike', mode='hdmi')[source]#

Bases: ClockDrivenSettings

Settings for DNSS LFP producer.

Parameters:
__init__(fs=30000, n_time=None, n_ch=256, pattern='spike', mode='hdmi')#
Parameters:
Return type:

None

fs: float = 30000#

Sample rate in Hz. DNSS is fixed at 30kHz.

mode: str = 'hdmi'#

“hdmi” for digital output, “pedestal_norm” or “pedestal_wide” for analog.

Type:

Mode

n_ch: int = 256#

Number of channels.

pattern: str = 'spike'#

“spike” for normal neural signal mode, “other” for other mode.

Type:

LFP pattern

class DNSSLFPUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[DNSSLFPSettings, DNSSLFPProducer]

Unit for generating DNSS LFP from clock input.

Parameters:

settings (Settings | None)

SETTINGS#

alias of DNSSLFPSettings

class DNSSSpikeProducer(*args, **kwargs)[source]#

Bases: BaseClockDrivenProducer[DNSSSpikeSettings, DNSSSpikeState]

Produces DNSS spike signal synchronized to clock ticks.

Each clock tick produces a block of spike data as sparse COO arrays based on the sample rate (fs) and chunk size (n_time) settings.

class DNSSSpikeSettings(fs=30000, n_time=None, n_ch=256, mode='hdmi')[source]#

Bases: ClockDrivenSettings

Settings for DNSS spike producer.

Parameters:
__init__(fs=30000, n_time=None, n_ch=256, mode='hdmi')#
Parameters:
Return type:

None

fs: float = 30000#

Sample rate in Hz. DNSS is fixed at 30kHz.

mode: str = 'hdmi'#

“hdmi” reproduces HDMI bugs, “ideal” for ideal pattern.

Type:

Mode

n_ch: int = 256#

Number of channels.

class DNSSSpikeUnit(*args, settings=None, **kwargs)[source]#

Bases: BaseClockDrivenUnit[DNSSSpikeSettings, DNSSSpikeProducer]

Unit for generating DNSS spikes from clock input.

Parameters:

settings (Settings | None)

SETTINGS#

alias of DNSSSpikeSettings

Modules

cosine_encoder

Generic cosine-tuning encoder for polar coordinates.

dnss

DNSS (Digital Neural Signal Simulator) signal generation.

dynamic_colored_noise

Dynamic colored noise generation with time-varying spectral exponent.

eeg

EEG signal synthesis.

noise

Noise signal generators.

oscillator

Oscillator/sinusoidal signal generators.

system

util

On this page

This Page