ezmsg.learn.collection.sample_adapt_regressor#

Classes

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

Bases: Collection

Parameters:

settings (Settings | None)

SETTINGS#

alias of SampleAdaptRegressorSettings

INPUT_LABELS = InputTopicStream:unlocated[AxisArray]()#
INPUT_SIGNAL = InputTopicStream:unlocated[AxisArray]()#
INPUT_TRIGGER = InputTopicStream:unlocated[SampleTriggerMessage]()#
OUTPUT_SIGNAL = OutputTopicStream:unlocated[AxisArray]()#
RESAMPLE = <ezmsg.sigproc.resample.ResampleUnit object>#
SEQSEQSAMPLER = <ezmsg.learn.process.seqseqsampler.SeqSeqSamplerUnit object>#
WINDOW = <ezmsg.sigproc.window.Window object>#
FLATTEN = <ezmsg.learn.process.flatten.Flatten object>#
REGRESSOR = <ezmsg.learn.process.adaptive_linear_regressor.AdaptiveLinearRegressorUnit 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. Override this method to perform collection-specific configuration of child components.

Return type:

None

network()[source]#

Override this method and have the definition return a NetworkDefinition which defines how InputStreams and OutputStreams from member Units will be connected.

The NetworkDefinition specifies the message routing between components by connecting output streams to input streams.

Returns:

Network definition specifying stream connections

Return type:

Iterable[tuple[Stream | str, Stream | str]]

class SampleAdaptRegressorSettings(model_type=AdaptiveLinearRegressor.LINEAR, model_path=None, model_kwargs=<factory>, resample_axis='time', resample_buffer_duration=2.0, sampler_max_buffer_dur=5.0, decode_window_dur=None, decode_window_shift=None)[source]#

Bases: Settings

Parameters:
model_type: AdaptiveLinearRegressor = 'linear'#

Adaptive regressor backend/model.

model_path: str | None = None#

Optional path to a pickled preconfigured model.

model_kwargs: dict#

Extra kwargs passed to the underlying regressor.

resample_axis: str = 'time'#

Axis to resample along.

resample_buffer_duration: float = 2.0#

Duration of the buffer for resampling in seconds.

sampler_max_buffer_dur: float = 5.0#

Maximum buffer duration for the SeqSeqSampler in seconds.

decode_window_dur: float | None = None#

Optional inference-side feature window duration in seconds.

decode_window_shift: float | None = None#

Optional inference-side feature window shift in seconds.

__init__(model_type=AdaptiveLinearRegressor.LINEAR, model_path=None, model_kwargs=<factory>, resample_axis='time', resample_buffer_duration=2.0, sampler_max_buffer_dur=5.0, decode_window_dur=None, decode_window_shift=None)#
Parameters:
Return type:

None