ezmsg.sigproc.synth#

Module Attributes

clock(dispatch_rate)

Alias for aclock expected by synchronous methods.

Functions

aclock(dispatch_rate)[source]#

Construct an async generator that yields events at a specified rate.

Returns:: A ClockProducer object.
Parameters:: dispatch_rate (float | None)
Return type:: ClockProducer

acounter(n_time, fs, n_ch=1, dispatch_rate=None, mod=None)[source]#

Construct an asynchronous generator to generate AxisArray objects at a specified rate and with the specified sampling rate.

NOTE: This module uses asyncio.sleep to delay appropriately in realtime mode. This method of sleeping/yielding execution priority has quirky behavior with sub-millisecond sleep periods which may result in unexpected behavior (e.g. fs = 2000, n_time = 1, realtime = True – may result in ~1400 msgs/sec)

Returns:

An asynchronous generator.

Parameters:

n_time (int)
fs (float | None)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

Return type:

CounterProducer

clock(dispatch_rate)#

Alias for aclock expected by synchronous methods. ClockProducer can be used in sync or async.

Parameters:: dispatch_rate (float | None)
Return type:: ClockProducer

sin(axis='time', freq=1.0, amp=1.0, phase=0.0)[source]#

Construct a generator of sinusoidal waveforms in AxisArray objects.

Returns:

A primed generator that expects .send(axis_array) of sample counts and yields an AxisArray of sinusoids.

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

Return type:

SinTransformer

Classes

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

Bases: Unit

Add two signals together. Assumes compatible/similar axes/dimensions.

Parameters:: settings (Settings | None)

INPUT_SIGNAL_A = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#

INPUT_SIGNAL_B = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#

async initialize()[source]#

Runs when the Unit is instantiated. This is called from within the same process this unit will live. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

Return type:: None

async on_a(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

async on_b(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

async output()[source]#

Return type:: AsyncGenerator

class AddProcessor[source]#

Bases: object

__init__()[source]#

property state: AddState#

push_a(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

push_b(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

class AddState(queue_a: 'asyncio.Queue[AxisArray]' = <factory>, queue_b: 'asyncio.Queue[AxisArray]' = <factory>)[source]#

Bases: object

Parameters:

queue_a (Queue[AxisArray])
queue_b (Queue[AxisArray])

queue_a: Queue[AxisArray]#

queue_b: Queue[AxisArray]#

__init__(queue_a=<factory>, queue_b=<factory>)#

Parameters:

queue_a (Queue[AxisArray])
queue_b (Queue[AxisArray])

Return type:

None

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

Bases: BaseProducerUnit[SettingsType, MessageOutType, ProducerType], Generic[SettingsType, MessageInType, MessageOutType, ProducerType]

Base class for units whose primary processor is a composite producer with a CounterProducer as the first processor (producer) in the chain.

Parameters:: settings (Settings | None)

INPUT_SIGNAL = InputStream:unlocated[~MessageInType]()#

create_producer()[source]#: Create the producer instance from settings.

async on_signal(_)[source]#

Parameters:: _ (Flag)

async produce()[source]#

Return type:: AsyncGenerator

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

Bases: BaseProducerUnit[ClockSettings, Flag, ClockProducer]

Parameters:: settings (Settings | None)

SETTINGS#: alias of ClockSettings

async produce()[source]#

Return type:: AsyncGenerator

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

Bases: BaseStatefulProducer[ClockSettings, Flag, ClockState]

Produces clock ticks at specified rate. Can be used to drive periodic operations.

class ClockSettings(dispatch_rate=None)[source]#

Bases: Settings

Settings for clock generator.

Parameters:: dispatch_rate (float | str | None)

dispatch_rate: float | str | None = None#: Dispatch rate in Hz, ‘realtime’, or None for external clock

__init__(dispatch_rate=None)#

Parameters:: dispatch_rate (float | str | None)
Return type:: None

class ClockState[source]#

Bases: object

State for clock generator.

t_0: float#

n_dispatch: int = 0#

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

Bases: BaseProducerUnit[CounterSettings, AxisArray, CounterProducer]

Generates monotonically increasing counter. Unit for CounterProducer.

Parameters:: settings (Settings | None)

SETTINGS#: alias of CounterSettings

INPUT_CLOCK = InputStream:unlocated[<class 'ezmsg.core.message.Flag'>]()#

async on_clock(_)[source]#

Parameters:: _ (Flag)

async produce()[source]#

Generate counter output. This is an infinite loop, but we will likely only enter the loop once if we are self-timed, and twice if we are using an external clock.

When using an internal clock, we enter the loop, and wait for the event which should have been reset upon initialization then we immediately clear, then go to the internal loop that will async call __acall__ to let the internal timer determine when to produce an output.

When using an external clock, we enter the loop, and wait for the event which should have been reset upon initialization then we immediately clear, then we hit continue to loop back around and wait for the event to be set again – potentially forever. In this case, it is expected that on_clock will be called to produce the output.

Return type:: AsyncGenerator

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

Bases: BaseStatefulProducer[CounterSettings, AxisArray, CounterState]

Produces incrementing integer blocks as AxisArray.

classmethod get_message_type(dir)[source]#

Parameters:: dir (str)
Return type:: type[AxisArray] | None

__init__(*args, **kwargs)[source]#

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

Bases: Settings

Settings for Counter. See acounter for a description of the parameters.

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

n_time: int#: Number of samples to output per block.

fs: float#: Sampling rate of signal output in Hz

n_ch: int = 1#: Number of channels to synthesize

dispatch_rate: float | str | None = None#: Message dispatch rate (Hz), ‘realtime’, ‘ext_clock’, or None (fast as possible) Note: if dispatch_rate is a float then time offsets will be synthetic and the system will run faster or slower than wall clock time.

mod: int | None = None#: If set to an integer, counter will rollover

__init__(n_time, fs, n_ch=1, dispatch_rate=None, mod=None)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

Return type:

None

class CounterState[source]#

Bases: object

State for counter generator.

counter_start: int = 0#: next sample’s first value

n_sent: int = 0#: number of samples sent

clock_zero: float | None = None#: time of first sample

timer_type: str = 'unspecified'#: “realtime” | “ext_clock” | “manual” | “unspecified”

new_generator: Event | None = None#: Event to signal the counter has been reset.

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

Bases: Collection

A Collection that chains a Clock to both PinkNoise and Oscillator, then Add s the result.

Unlike the Oscillator, WhiteNoise, and PinkNoise composite processors which have linear flows, this class has a diamond flow, with clock branching to both PinkNoise and Oscillator, which then are combined in Add.

Optional: Refactor as a ProducerUnit, similar to Clock, but we manually add all the other: transformers.

Parameters:: settings (Settings | None)

SETTINGS#: alias of EEGSynthSettings

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#

CLOCK = <ezmsg.sigproc.synth.Clock object>#

NOISE = <ezmsg.sigproc.synth.PinkNoise object>#

OSC = <ezmsg.sigproc.synth.Oscillator object>#

ADD = <ezmsg.sigproc.synth.Add object>#

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[Stream | str, Stream | str]]

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

Bases: Settings

See OscillatorSettings.

Parameters:

fs (float)
n_time (int)
alpha_freq (float)
n_ch (int)

__init__(fs=500.0, n_time=100, alpha_freq=10.5, n_ch=8)#

Parameters:

fs (float)
n_time (int)
alpha_freq (float)
n_ch (int)

Return type:

None

fs: float = 500.0#

n_time: int = 100#

alpha_freq: float = 10.5#

n_ch: int = 8#

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

Bases: Settings

See CounterSettings and RandomGeneratorSettings.

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
loc (float)
scale (float)

n_time: int#

fs: float#

n_ch: int = 1#

__init__(n_time, fs, n_ch=1, dispatch_rate=None, loc=0.0, scale=1.0)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
loc (float)
scale (float)

Return type:

None

dispatch_rate: float | str | None = None#: (Hz), ‘realtime’, or ‘ext_clock’

loc: float = 0.0#

scale: float = 1.0#

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

Bases: BaseCounterFirstProducerUnit[OscillatorSettings, AxisArray, AxisArray, OscillatorProducer]

Generates sinusoidal waveforms using a counter and sine transformer.

Parameters:: settings (Settings | None)

SETTINGS#: alias of OscillatorSettings

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

Bases: CompositeProducer[OscillatorSettings, AxisArray]

class OscillatorSettings(n_time, fs, n_ch=1, dispatch_rate=None, freq=1.0, amp=1.0, phase=0.0, sync=False)[source]#

Bases: Settings

Settings for Oscillator

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
freq (float)
amp (float)
phase (float)
sync (bool)

n_time: int#: Number of samples to output per block.

fs: float#: Sampling rate of signal output in Hz

n_ch: int = 1#: Number of channels to output per block

dispatch_rate: float | str | None = None#: (Hz) | ‘realtime’ | ‘ext_clock’

freq: float = 1.0#: Oscillation frequency in Hz

amp: float = 1.0#: Amplitude

phase: float = 0.0#: Phase offset (in radians)

sync: bool = False#: Adjust freq to sync with sampling rate

__init__(n_time, fs, n_ch=1, dispatch_rate=None, freq=1.0, amp=1.0, phase=0.0, sync=False)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
freq (float)
amp (float)
phase (float)
sync (bool)

Return type:

None

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

Bases: BaseCounterFirstProducerUnit[NoiseSettings, AxisArray, AxisArray, PinkNoiseProducer]

chains WhiteNoise and ButterworthFilter.

Parameters:: settings (Settings | None)

SETTINGS#: alias of NoiseSettings

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

Bases: CompositeProducer[NoiseSettings, AxisArray]

PinkNoiseSettings#: alias of NoiseSettings

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

Bases: BaseTransformerUnit[RandomGeneratorSettings, AxisArray, AxisArray, RandomTransformer]

Parameters:: settings (Settings | None)

SETTINGS#: alias of RandomGeneratorSettings

class RandomGeneratorSettings(loc: float = 0.0, scale: float = 1.0)[source]#

Bases: Settings

Parameters:

loc (float)
scale (float)

loc: float = 0.0#: loc argument for numpy.random.normal

scale: float = 1.0#: scale argument for numpy.random.normal

__init__(loc=0.0, scale=1.0)#

Parameters:

loc (float)
scale (float)

Return type:

None

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

Bases: BaseTransformer[RandomGeneratorSettings, AxisArray, AxisArray]

Replaces input data with random data and returns the result.

Parameters:: settings (SettingsType)

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

Parameters:: settings (RandomGeneratorSettings | None)

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

Bases: BaseTransformerUnit[SinGeneratorSettings, AxisArray, AxisArray, SinTransformer]

Unit for generating sinusoidal waveforms.

Parameters:: settings (Settings | None)

SETTINGS#: alias of SinGeneratorSettings

class SinGeneratorSettings(axis='time', freq=1.0, amp=1.0, phase=0.0)[source]#

Bases: Settings

Settings for SinGenerator. See sin for parameter descriptions.

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

axis: str | None = 'time'#: The name of the axis over which the sinusoid passes. Note: The axis must exist in the msg.axes and be of type AxisArray.LinearAxis.

freq: float = 1.0#: The frequency of the sinusoid, in Hz.

amp: float = 1.0#: The amplitude of the sinusoid.

phase: float = 0.0#: The initial phase of the sinusoid, in radians.

__init__(axis='time', freq=1.0, amp=1.0, phase=0.0)#

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

Return type:

None

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

Bases: BaseTransformer[SinGeneratorSettings, AxisArray, AxisArray]

Transforms counter values into sinusoidal waveforms.

Parameters:: settings (SettingsType)

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

Bases: BaseCounterFirstProducerUnit[NoiseSettings, AxisArray, AxisArray, WhiteNoiseProducer]

chains a Counter and RandomGenerator.

Parameters:: settings (Settings | None)

SETTINGS#: alias of NoiseSettings

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

Bases: CompositeProducer[NoiseSettings, AxisArray]

WhiteNoiseSettings#: alias of NoiseSettings

class AddState(queue_a: 'asyncio.Queue[AxisArray]' = <factory>, queue_b: 'asyncio.Queue[AxisArray]' = <factory>)[source]#

Bases: object

Parameters:

queue_a (Queue[AxisArray])
queue_b (Queue[AxisArray])

queue_a: Queue[AxisArray]#

queue_b: Queue[AxisArray]#

__init__(queue_a=<factory>, queue_b=<factory>)#

Parameters:

queue_a (Queue[AxisArray])
queue_b (Queue[AxisArray])

Return type:

None

class AddProcessor[source]#

Bases: object

__init__()[source]#

property state: AddState#

push_a(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

push_b(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

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

Bases: Unit

Add two signals together. Assumes compatible/similar axes/dimensions.

Parameters:: settings (Settings | None)

INPUT_SIGNAL_A = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#

INPUT_SIGNAL_B = InputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>]()#

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#

async initialize()[source]#

Return type:: None

async on_a(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

async on_b(msg)[source]#

Parameters:: msg (AxisArray)
Return type:: None

async output()[source]#

Return type:: AsyncGenerator

class ClockSettings(dispatch_rate=None)[source]#

Bases: Settings

Settings for clock generator.

Parameters:: dispatch_rate (float | str | None)

dispatch_rate: float | str | None = None#: Dispatch rate in Hz, ‘realtime’, or None for external clock

__init__(dispatch_rate=None)#

Parameters:: dispatch_rate (float | str | None)
Return type:: None

class ClockState[source]#

Bases: object

State for clock generator.

t_0: float#

n_dispatch: int = 0#

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

Bases: BaseStatefulProducer[ClockSettings, Flag, ClockState]

Produces clock ticks at specified rate. Can be used to drive periodic operations.

aclock(dispatch_rate)[source]#

Construct an async generator that yields events at a specified rate.

Returns:: A ClockProducer object.
Parameters:: dispatch_rate (float | None)
Return type:: ClockProducer

clock(dispatch_rate)#

Alias for aclock expected by synchronous methods. ClockProducer can be used in sync or async.

Parameters:: dispatch_rate (float | None)
Return type:: ClockProducer

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

Bases: BaseProducerUnit[ClockSettings, Flag, ClockProducer]

Parameters:: settings (Settings | None)

SETTINGS#: alias of ClockSettings

async produce()[source]#

Return type:: AsyncGenerator

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

Bases: Settings

Settings for Counter. See acounter for a description of the parameters.

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

n_time: int#: Number of samples to output per block.

fs: float#: Sampling rate of signal output in Hz

n_ch: int = 1#: Number of channels to synthesize

dispatch_rate: float | str | None = None#: Message dispatch rate (Hz), ‘realtime’, ‘ext_clock’, or None (fast as possible) Note: if dispatch_rate is a float then time offsets will be synthetic and the system will run faster or slower than wall clock time.

mod: int | None = None#: If set to an integer, counter will rollover

__init__(n_time, fs, n_ch=1, dispatch_rate=None, mod=None)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

Return type:

None

class CounterState[source]#

Bases: object

State for counter generator.

counter_start: int = 0#: next sample’s first value

n_sent: int = 0#: number of samples sent

clock_zero: float | None = None#: time of first sample

timer_type: str = 'unspecified'#: “realtime” | “ext_clock” | “manual” | “unspecified”

new_generator: Event | None = None#: Event to signal the counter has been reset.

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

Bases: BaseStatefulProducer[CounterSettings, AxisArray, CounterState]

Produces incrementing integer blocks as AxisArray.

classmethod get_message_type(dir)[source]#

Parameters:: dir (str)
Return type:: type[AxisArray] | None

__init__(*args, **kwargs)[source]#

acounter(n_time, fs, n_ch=1, dispatch_rate=None, mod=None)[source]#

Construct an asynchronous generator to generate AxisArray objects at a specified rate and with the specified sampling rate.

Returns:

An asynchronous generator.

Parameters:

n_time (int)
fs (float | None)
n_ch (int)
dispatch_rate (float | str | None)
mod (int | None)

Return type:

CounterProducer

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

Bases: BaseProducerUnit[CounterSettings, AxisArray, CounterProducer]

Generates monotonically increasing counter. Unit for CounterProducer.

Parameters:: settings (Settings | None)

SETTINGS#: alias of CounterSettings

INPUT_CLOCK = InputStream:unlocated[<class 'ezmsg.core.message.Flag'>]()#

async on_clock(_)[source]#

Parameters:: _ (Flag)

async produce()[source]#

Generate counter output. This is an infinite loop, but we will likely only enter the loop once if we are self-timed, and twice if we are using an external clock.

Return type:: AsyncGenerator

class SinGeneratorSettings(axis='time', freq=1.0, amp=1.0, phase=0.0)[source]#

Bases: Settings

Settings for SinGenerator. See sin for parameter descriptions.

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

axis: str | None = 'time'#: The name of the axis over which the sinusoid passes. Note: The axis must exist in the msg.axes and be of type AxisArray.LinearAxis.

freq: float = 1.0#: The frequency of the sinusoid, in Hz.

amp: float = 1.0#: The amplitude of the sinusoid.

phase: float = 0.0#: The initial phase of the sinusoid, in radians.

__init__(axis='time', freq=1.0, amp=1.0, phase=0.0)#

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

Return type:

None

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

Bases: BaseTransformer[SinGeneratorSettings, AxisArray, AxisArray]

Transforms counter values into sinusoidal waveforms.

Parameters:: settings (SettingsType)

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

Bases: BaseTransformerUnit[SinGeneratorSettings, AxisArray, AxisArray, SinTransformer]

Unit for generating sinusoidal waveforms.

Parameters:: settings (Settings | None)

SETTINGS#: alias of SinGeneratorSettings

sin(axis='time', freq=1.0, amp=1.0, phase=0.0)[source]#

Construct a generator of sinusoidal waveforms in AxisArray objects.

Returns:

A primed generator that expects .send(axis_array) of sample counts and yields an AxisArray of sinusoids.

Parameters:

axis (str | None)
freq (float)
amp (float)
phase (float)

Return type:

SinTransformer

class RandomGeneratorSettings(loc: float = 0.0, scale: float = 1.0)[source]#

Bases: Settings

Parameters:

loc (float)
scale (float)

loc: float = 0.0#: loc argument for numpy.random.normal

scale: float = 1.0#: scale argument for numpy.random.normal

__init__(loc=0.0, scale=1.0)#

Parameters:

loc (float)
scale (float)

Return type:

None

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

Bases: BaseTransformer[RandomGeneratorSettings, AxisArray, AxisArray]

Replaces input data with random data and returns the result.

Parameters:: settings (SettingsType)

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

Parameters:: settings (RandomGeneratorSettings | None)

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

Bases: BaseTransformerUnit[RandomGeneratorSettings, AxisArray, AxisArray, RandomTransformer]

Parameters:: settings (Settings | None)

SETTINGS#: alias of RandomGeneratorSettings

class OscillatorSettings(n_time, fs, n_ch=1, dispatch_rate=None, freq=1.0, amp=1.0, phase=0.0, sync=False)[source]#

Bases: Settings

Settings for Oscillator

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
freq (float)
amp (float)
phase (float)
sync (bool)

n_time: int#: Number of samples to output per block.

fs: float#: Sampling rate of signal output in Hz

n_ch: int = 1#: Number of channels to output per block

dispatch_rate: float | str | None = None#: (Hz) | ‘realtime’ | ‘ext_clock’

freq: float = 1.0#: Oscillation frequency in Hz

amp: float = 1.0#: Amplitude

phase: float = 0.0#: Phase offset (in radians)

sync: bool = False#: Adjust freq to sync with sampling rate

__init__(n_time, fs, n_ch=1, dispatch_rate=None, freq=1.0, amp=1.0, phase=0.0, sync=False)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
freq (float)
amp (float)
phase (float)
sync (bool)

Return type:

None

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

Bases: CompositeProducer[OscillatorSettings, AxisArray]

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

Bases: BaseProducerUnit[SettingsType, MessageOutType, ProducerType], Generic[SettingsType, MessageInType, MessageOutType, ProducerType]

Base class for units whose primary processor is a composite producer with a CounterProducer as the first processor (producer) in the chain.

Parameters:: settings (Settings | None)

INPUT_SIGNAL = InputStream:unlocated[~MessageInType]()#

create_producer()[source]#: Create the producer instance from settings.

async on_signal(_)[source]#

Parameters:: _ (Flag)

async produce()[source]#

Return type:: AsyncGenerator

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

Bases: BaseCounterFirstProducerUnit[OscillatorSettings, AxisArray, AxisArray, OscillatorProducer]

Generates sinusoidal waveforms using a counter and sine transformer.

Parameters:: settings (Settings | None)

SETTINGS#: alias of OscillatorSettings

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

Bases: Settings

See CounterSettings and RandomGeneratorSettings.

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
loc (float)
scale (float)

n_time: int#

fs: float#

n_ch: int = 1#

__init__(n_time, fs, n_ch=1, dispatch_rate=None, loc=0.0, scale=1.0)#

Parameters:

n_time (int)
fs (float)
n_ch (int)
dispatch_rate (float | str | None)
loc (float)
scale (float)

Return type:

None

dispatch_rate: float | str | None = None#: (Hz), ‘realtime’, or ‘ext_clock’

loc: float = 0.0#

scale: float = 1.0#

WhiteNoiseSettings#: alias of NoiseSettings

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

Bases: CompositeProducer[NoiseSettings, AxisArray]

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

Bases: BaseCounterFirstProducerUnit[NoiseSettings, AxisArray, AxisArray, WhiteNoiseProducer]

chains a Counter and RandomGenerator.

Parameters:: settings (Settings | None)

SETTINGS#: alias of NoiseSettings

PinkNoiseSettings#: alias of NoiseSettings

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

Bases: CompositeProducer[NoiseSettings, AxisArray]

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

Bases: BaseCounterFirstProducerUnit[NoiseSettings, AxisArray, AxisArray, PinkNoiseProducer]

chains WhiteNoise and ButterworthFilter.

Parameters:: settings (Settings | None)

SETTINGS#: alias of NoiseSettings

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

Bases: Settings

See OscillatorSettings.

Parameters:

fs (float)
n_time (int)
alpha_freq (float)
n_ch (int)

__init__(fs=500.0, n_time=100, alpha_freq=10.5, n_ch=8)#

Parameters:

fs (float)
n_time (int)
alpha_freq (float)
n_ch (int)

Return type:

None

fs: float = 500.0#

n_time: int = 100#

alpha_freq: float = 10.5#

n_ch: int = 8#

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

Bases: Collection

A Collection that chains a Clock to both PinkNoise and Oscillator, then Add s the result.

Optional: Refactor as a ProducerUnit, similar to Clock, but we manually add all the other: transformers.

Parameters:: settings (Settings | None)

SETTINGS#: alias of EEGSynthSettings

OUTPUT_SIGNAL = OutputStream:unlocated[<class 'ezmsg.util.messages.axisarray.AxisArray'>](self.num_buffers=32, self.force_tcp=False)#

CLOCK = <ezmsg.sigproc.synth.Clock object>#

NOISE = <ezmsg.sigproc.synth.PinkNoise object>#

OSC = <ezmsg.sigproc.synth.Oscillator object>#

ADD = <ezmsg.sigproc.synth.Add object>#

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[Stream | str, Stream | str]]

ezmsg.sigproc.synth#

This Page