🟡 unaiverse.streams.streamlib

What this module does 🟡

Library of synthetic signal stream generators (random, sinusoidal, square, smooth/square frequency-amplitude variants) plus a one-hot label stream for testing and demos.

streamlib

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

AllHotLabelStream

AllHotLabelStream(labels: list[str], device: device = device('cpu'))

Bases: BufferedStream

A static buffered stream that repeatedly yields an all-ones tensor with named labels.

The stream holds a single torch.float32 tensor of shape (1, len(labels)) filled with ones. Because it is constructed with is_static=True, the same tensor is returned on every read without any index-based generation. This is useful as a dummy or constant input stream whose labelling rule ("geq0.5") marks all labels as active.

The tensor is placed on the requested device at construction time and does not move afterwards. The DataProps name is set to AllHotLabelStream and the data description to "dummy stream".

Examples:

>>> import torch
>>> stream = AllHotLabelStream(labels=["cat", "dog", "bird"])
>>> data, cycle_offset = stream[0, None]
>>> data.shape
torch.Size([1, 3])
>>> data
tensor([[1., 1., 1.]])

Initialize an AllHotLabelStream with the given label names.

Constructs the underlying BufferedStream with a DataProps that describes a single-row float tensor whose columns correspond to labels. The tensor is immediately set to all-ones and the stream is restarted so it is ready for reading.

Parameters:

Name	Type	Description	Default
labels	list[str]	List of label strings to associate with the all-ones tensor. The length determines the number of columns in the output tensor.	required
device	device	PyTorch device on which the tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, labels: list[str],
             device: torch.device = torch.device('cpu')) -> None:
    """Initialize an AllHotLabelStream with the given label names.

    Constructs the underlying ``BufferedStream`` with a ``DataProps`` that describes a
    single-row float tensor whose columns correspond to ``labels``. The tensor is immediately
    set to all-ones and the stream is restarted so it is ready for reading.

    Args:
        labels: List of label strings to associate with the all-ones tensor. The length
            determines the number of columns in the output tensor.
        device: PyTorch device on which the tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(props=DataProps(name=AllHotLabelStream.__name__,
                                     data_type="tensor",
                                     data_desc="dummy stream",
                                     tensor_shape=(1, len(labels)),
                                     tensor_labels=labels,
                                     tensor_dtype=str(torch.float),
                                     tensor_labeling_rule="geq0.5"),
                     is_static=True)

    self.set(torch.ones((1, len(labels)), dtype=torch.float32, device=device))
    self.restart()

Random

Random(std: float, shape: tuple[int, ...] | None = (1,), device: device = device('cpu'))

Bases: BufferedStream

A buffered stream that yields random tensors drawn from a scaled uniform distribution.

Each call to __getitem__ generates a fresh tensor by sampling from torch.rand (values in [0, 1)) and multiplying by std. The resulting values are therefore in [0, std), not zero-centred. The stream name in DataProps is "rand" and the data description is "stream of random numbers".

The tensor shape is fixed at construction time and every sample has the same shape. Generation happens on the fly: no buffer is pre-filled. The stream must be restarted (via restart) before the first read, which the constructor does automatically.

Examples:

>>> import torch
>>> stream = Random(std=1.0, shape=(1, 4))
>>> data, cycle_offset = stream[0, None]
>>> data.shape
torch.Size([1, 4])

Initialize a Random stream.

Sets up DataProps for a float tensor of the given shape, stores std and device, and calls restart so the stream is ready immediately.

Parameters:

Name	Type	Description	Default
std	float	Scale factor applied to the uniformly-sampled values. Each generated tensor element is in the range [0, std).	required
shape	tuple[int, ...] | None	Shape of each output tensor. Defaults to (1,).	(1,)
device	device	PyTorch device on which tensors are generated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, std: float, shape: tuple[int, ...] | None = (1,),
             device: torch.device = torch.device('cpu')) -> None:
    """Initialize a Random stream.

    Sets up ``DataProps`` for a float tensor of the given ``shape``, stores ``std`` and
    ``device``, and calls ``restart`` so the stream is ready immediately.

    Args:
        std: Scale factor applied to the uniformly-sampled values. Each generated
            tensor element is in the range ``[0, std)``.
        shape: Shape of each output tensor. Defaults to ``(1,)``.
        device: PyTorch device on which tensors are generated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(props=DataProps(name="rand",
                                     data_type="tensor",
                                     data_desc="stream of random numbers",
                                     tensor_shape=shape,
                                     tensor_dtype=str(torch.float)))
    self.std = std
    self.device = device
    self.restart()

std instance-attribute

std = std

device instance-attribute

device = device

Sin

Sin(freq: float, phase: float, delta: float, device: device = device('cpu'))

Bases: BufferedStream

A buffered stream that generates samples from a sine function.

Each sample at integer index idx is computed as sin(2 * pi * freq * (idx * delta + phase * period)), where period = 1 / freq. The output tensor has shape (1, 1) and dtype torch.float.

The stream name in DataProps is "sin" and the data description is "stream of samples from the sin function". The stream calls restart during construction so it is ready for reading immediately.

Examples:

>>> import torch
>>> stream = Sin(freq=1.0, phase=0.0, delta=0.25)
>>> data, _ = stream[0, None]   # sin(0) = 0
>>> data.shape
torch.Size([1, 1])

Initialize a Sin stream.

Stores the frequency, phase, derived period, and time step, then calls restart so the stream is ready for reading immediately.

Parameters:

Name	Type	Description	Default
freq	float	Frequency of the sine wave in Hz.	required
phase	float	Phase offset in units of full periods (0.0 = no offset, 1.0 = one full period offset). The effective time shift is phase / freq seconds.	required
delta	float	Time step between consecutive sample indices in seconds. The sample at index idx corresponds to time idx * delta.	required
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, freq: float, phase: float, delta: float,
             device: torch.device = torch.device('cpu')) -> None:
    """Initialize a Sin stream.

    Stores the frequency, phase, derived period, and time step, then calls
    ``restart`` so the stream is ready for reading immediately.

    Args:
        freq: Frequency of the sine wave in Hz.
        phase: Phase offset in units of full periods (0.0 = no offset, 1.0 = one
            full period offset). The effective time shift is ``phase / freq`` seconds.
        delta: Time step between consecutive sample indices in seconds. The sample at
            index ``idx`` corresponds to time ``idx * delta``.
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(props=DataProps(name="sin",
                                     data_type="tensor",
                                     data_desc="stream of samples from the sin function",
                                     tensor_shape=(1, 1),
                                     tensor_dtype=str(torch.float)))
    self.freq = freq
    self.phase = phase
    self.period = 1. / self.freq
    self.delta = delta
    self.device = device
    self.restart()

freq instance-attribute

freq = freq

phase instance-attribute

phase = phase

period instance-attribute

period = 1.0 / freq

delta instance-attribute

delta = delta

device instance-attribute

device = device

Square

Square(freq: float, ampl: float, phase: float, delta: float, device: device = device('cpu'))

Bases: BufferedStream

A buffered stream that generates samples from a square wave function.

Each sample at integer index idx is computed as ampl * (-1) ** floor(2 * freq * t) where t = idx * delta + phase * period and period = 1 / freq. The result alternates between +ampl and -ampl at the given frequency. The output tensor has shape (1, 1) and dtype torch.float.

The stream name in DataProps is "square" and the data description is "stream of samples from the square function". The stream calls restart during construction so it is ready for reading immediately.

Examples:

>>> import torch
>>> stream = Square(freq=1.0, ampl=1.0, phase=0.0, delta=0.1)
>>> data, _ = stream[0, None]   # t=0, value = ampl * (-1)^0 = 1.0
>>> float(data)
1.0

Initialize a Square wave stream.

Stores the frequency, amplitude, phase, derived period, and time step, then calls restart so the stream is ready for reading immediately.

Parameters:

Name	Type	Description	Default
freq	float	Frequency of the square wave in Hz.	required
ampl	float	Peak amplitude of the square wave. The output alternates between +ampl and -ampl.	required
phase	float	Phase offset in units of full periods. The effective time shift is phase / freq seconds.	required
delta	float	Time step between consecutive sample indices in seconds.	required
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, freq: float, ampl: float, phase: float, delta: float,
             device: torch.device = torch.device('cpu')) -> None:
    """Initialize a Square wave stream.

    Stores the frequency, amplitude, phase, derived period, and time step, then calls
    ``restart`` so the stream is ready for reading immediately.

    Args:
        freq: Frequency of the square wave in Hz.
        ampl: Peak amplitude of the square wave. The output alternates between
            ``+ampl`` and ``-ampl``.
        phase: Phase offset in units of full periods. The effective time shift is
            ``phase / freq`` seconds.
        delta: Time step between consecutive sample indices in seconds.
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(props=DataProps(name="square",
                                     data_type="tensor",
                                     data_desc="stream of samples from the square function",
                                     tensor_shape=(1, 1),
                                     tensor_dtype=str(torch.float)))
    self.freq = freq
    self.ampl = ampl
    self.phase = phase
    self.period = 1. / self.freq
    self.delta = delta
    self.device = device
    self.restart()

freq instance-attribute

freq = freq

ampl instance-attribute

ampl = ampl

phase instance-attribute

phase = phase

period instance-attribute

period = 1.0 / freq

delta instance-attribute

delta = delta

device instance-attribute

device = device

CombSin

CombSin(f_cap: float | list, c_cap: float | list, order: int, delta: float, device: device = device('cpu'))

Bases: BufferedStream

A buffered stream that generates samples from a weighted sum of sine functions.

The output at index idx is computed as:

``sum_k( coeffs[k] * sin(2 * pi * freqs[k] * idx * delta + phases[k]) )``

where all phases are initialised to zero. Frequencies and coefficients can be either randomly sampled (when a scalar cap is given) or prescribed explicitly (when a list is provided). The output tensor has shape (1, 1) and dtype torch.float.

The stream name in DataProps is "combsin" and the data description is "stream of samples from combined sin functions". The stream calls restart during construction so it is ready for reading immediately.

This class is the base for the preset streams SmoothHFHA, SmoothHFLA, SmoothLFLA, and SmoothLFHA.

Examples:

>>> import torch
>>> # Prescribed frequencies and coefficients:
>>> stream = CombSin(f_cap=[0.1, 0.2], c_cap=[0.5, 0.5], order=2, delta=0.1)
>>> data, _ = stream[0, None]
>>> data.shape
torch.Size([1, 1])
>>>
>>> # Random frequencies and coefficients:
>>> stream_rand = CombSin(f_cap=1.0, c_cap=1.0, order=4, delta=0.05)

Initialize a CombSin stream.

When f_cap is a float, order frequencies are sampled uniformly from [0, f_cap) using torch.rand. When it is a list, those values are used directly (the order argument is ignored for sizing in this case, but the assertion that len(coeffs) == len(freqs) still applies). The same logic applies to c_cap for coefficients, except that random coefficients are drawn from a symmetric distribution: c_cap * (2 * torch.rand(order) - 1), so they lie in (-c_cap, c_cap).

All phases are initialised to zero. The stream calls restart at the end of construction.

Parameters:

Name	Type	Description	Default
f_cap	float | list	If a float, the upper bound for randomly sampled frequencies. If a list, the explicit frequency values (one per sine component).	required
c_cap	float | list	If a float, the cap for randomly sampled coefficients drawn from a zero-centred uniform distribution in (-c_cap, c_cap). If a list, the explicit coefficient values.	required
order	int	Number of sine components. Used as the sample size when f_cap or c_cap are scalars.	required
delta	float	Time step between consecutive sample indices in seconds.	required
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Raises:

Type	Description
Exception	If f_cap is not a float or list.
Exception	If c_cap is not a float or list.
AssertionError	If the number of coefficients does not equal the number of frequencies after construction.

Source code in unaiverse/streams/streamlib.py

def __init__(self, f_cap: float | list, c_cap: float | list, order: int, delta: float,
             device: torch.device = torch.device('cpu')) -> None:
    """Initialize a CombSin stream.

    When ``f_cap`` is a ``float``, ``order`` frequencies are sampled uniformly from
    ``[0, f_cap)`` using ``torch.rand``. When it is a ``list``, those values are used
    directly (the ``order`` argument is ignored for sizing in this case, but the
    assertion that ``len(coeffs) == len(freqs)`` still applies). The same logic
    applies to ``c_cap`` for coefficients, except that random coefficients are drawn
    from a symmetric distribution: ``c_cap * (2 * torch.rand(order) - 1)``, so they
    lie in ``(-c_cap, c_cap)``.

    All phases are initialised to zero. The stream calls ``restart`` at the end of
    construction.

    Args:
        f_cap: If a ``float``, the upper bound for randomly sampled frequencies.
            If a ``list``, the explicit frequency values (one per sine component).
        c_cap: If a ``float``, the cap for randomly sampled coefficients drawn from
            a zero-centred uniform distribution in ``(-c_cap, c_cap)``. If a ``list``,
            the explicit coefficient values.
        order: Number of sine components. Used as the sample size when ``f_cap`` or
            ``c_cap`` are scalars.
        delta: Time step between consecutive sample indices in seconds.
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.

    Raises:
        Exception: If ``f_cap`` is not a ``float`` or ``list``.
        Exception: If ``c_cap`` is not a ``float`` or ``list``.
        AssertionError: If the number of coefficients does not equal the number of
            frequencies after construction.
    """
    super().__init__(props=DataProps(name="combsin",
                                     data_type="tensor",
                                     data_desc="stream of samples from combined sin functions",
                                     tensor_shape=(1, 1),
                                     tensor_dtype=str(torch.float)))
    if isinstance(f_cap, float):
        self.freqs = f_cap * torch.rand(order)
    elif isinstance(f_cap, list):
        self.freqs = torch.tensor(f_cap)
    else:
        raise Exception(f"expected float or list for f_cap, not {type(f_cap)}")
    self.phases = torch.zeros_like(self.freqs)
    if isinstance(c_cap, float):
        self.coeffs = c_cap * (2 * torch.rand(order) - 1)
    elif isinstance(c_cap, list):
        self.coeffs = torch.tensor(c_cap)
    else:
        raise Exception(f"expected float or list for c_cap, not {type(c_cap)}")

    # Check all the dimensions
    assert len(self.coeffs) == len(self.freqs), \
        (f"specify the same number of coefficients and frequencies (got {len(self.coeffs)} "
         f"and {len(self.freqs)} respectively).")

    self.delta = delta
    self.device = device
    self.restart()

freqs instance-attribute

freqs = f_cap * rand(order)

phases instance-attribute

phases = zeros_like(freqs)

coeffs instance-attribute

coeffs = c_cap * (2 * rand(order) - 1)

delta instance-attribute

delta = delta

device instance-attribute

device = device

SmoothHFHA

SmoothHFHA(device: device = device('cpu'))

Bases: CombSin

Preset CombSin stream: smooth signal with high frequency and high amplitude.

Fixed frequencies [0.11, 0.07, 0.05] Hz and coefficients [0.8, 0.16, 0.16] are used with delta=0.1 s. The highest-frequency component carries the largest weight, producing a high-amplitude, high-frequency smooth waveform. The stream name is set to "smoHfHa" after construction.

The FEATURES class attribute ['3sin', 'hf', 'ha'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SmoothHFHA stream with fixed high-frequency, high-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SmoothHFHA stream with fixed high-frequency, high-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    freqs = [0.11, 0.07, 0.05]
    coeffs = [0.8, 0.16, 0.16]
    super().__init__(f_cap=freqs, c_cap=coeffs, order=3, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("smoHfHa")

FEATURES class-attribute instance-attribute

FEATURES = ['3sin', 'hf', 'ha']

SmoothHFLA

SmoothHFLA(device: device = device('cpu'))

Bases: CombSin

Preset CombSin stream: smooth signal with high frequency and low amplitude.

Fixed frequencies [0.11, 0.07, 0.05] Hz and coefficients [0.4, 0.08, 0.08] are used with delta=0.1 s. The highest-frequency component carries the largest weight, producing a high-frequency but low-amplitude smooth waveform. The stream name is set to "smoHfLa" after construction.

The FEATURES class attribute ['3sin', 'hf', 'la'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SmoothHFLA stream with fixed high-frequency, low-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SmoothHFLA stream with fixed high-frequency, low-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    freqs = [0.11, 0.07, 0.05]
    coeffs = [0.4, 0.08, 0.08]
    super().__init__(f_cap=freqs, c_cap=coeffs, order=3, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("smoHfLa")

FEATURES class-attribute instance-attribute

FEATURES = ['3sin', 'hf', 'la']

SmoothLFLA

SmoothLFLA(device: device = device('cpu'))

Bases: CombSin

Preset CombSin stream: smooth signal with low frequency and low amplitude.

Fixed frequencies [0.11, 0.07, 0.05] Hz and coefficients [0.08, 0.08, 0.4] are used with delta=0.1 s. The lowest-frequency component carries the largest weight, producing a low-frequency, low-amplitude smooth waveform. The stream name is set to "smoLfLa" after construction.

The FEATURES class attribute ['3sin', 'lf', 'la'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SmoothLFLA stream with fixed low-frequency, low-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SmoothLFLA stream with fixed low-frequency, low-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    freqs = [0.11, 0.07, 0.05]
    coeffs = [0.08, 0.08, 0.4]
    super().__init__(f_cap=freqs, c_cap=coeffs, order=3, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("smoLfLa")

FEATURES class-attribute instance-attribute

FEATURES = ['3sin', 'lf', 'la']

SmoothLFHA

SmoothLFHA(device: device = device('cpu'))

Bases: CombSin

Preset CombSin stream: smooth signal with low frequency and high amplitude.

Fixed frequencies [0.11, 0.07, 0.05] Hz and coefficients [0.16, 0.16, 0.8] are used with delta=0.1 s. The lowest-frequency component carries the largest weight, producing a low-frequency, high-amplitude smooth waveform. The stream name is set to "smoLfHa" after construction.

The FEATURES class attribute ['3sin', 'lf', 'ha'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SmoothLFHA stream with fixed low-frequency, high-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SmoothLFHA stream with fixed low-frequency, high-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    freqs = [0.11, 0.07, 0.05]
    coeffs = [0.16, 0.16, 0.8]
    super().__init__(f_cap=freqs, c_cap=coeffs, order=3, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("smoLfHa")

FEATURES class-attribute instance-attribute

FEATURES = ['3sin', 'lf', 'ha']

SquareHFHA

SquareHFHA(device: device = device('cpu'))

Bases: Square

Preset Square wave stream: high frequency and high amplitude.

Uses freq=0.06 Hz, ampl=1.0, phase=0.5, and delta=0.1 s. The stream name is set to "squHfHa" after construction.

The FEATURES class attribute ['square', 'hf', 'ha'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SquareHFHA stream with fixed high-frequency, high-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SquareHFHA stream with fixed high-frequency, high-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(freq=0.06, phase=0.5, ampl=1.0, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("squHfHa")

FEATURES class-attribute instance-attribute

FEATURES = ['square', 'hf', 'ha']

SquareHFLA

SquareHFLA(device: device = device('cpu'))

Bases: Square

Preset Square wave stream: high frequency and low amplitude.

Uses freq=0.06 Hz, ampl=0.5, phase=0.5, and delta=0.1 s. The stream name is set to "squHfLa" after construction.

The FEATURES class attribute ['square', 'hf', 'la'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SquareHFLA stream with fixed high-frequency, low-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SquareHFLA stream with fixed high-frequency, low-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(freq=0.06, phase=0.5, ampl=0.5, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("squHfLa")

FEATURES class-attribute instance-attribute

FEATURES = ['square', 'hf', 'la']

SquareLFHA

SquareLFHA(device: device = device('cpu'))

Bases: Square

Preset Square wave stream: low frequency and high amplitude.

Uses freq=0.03 Hz, ampl=1.0, phase=0.5, and delta=0.1 s. The stream name is set to "squLfHa" after construction.

The FEATURES class attribute ['square', 'lf', 'ha'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SquareLFHA stream with fixed low-frequency, high-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SquareLFHA stream with fixed low-frequency, high-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(freq=0.03, phase=0.5, ampl=1.0, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("squLfHa")

FEATURES class-attribute instance-attribute

FEATURES = ['square', 'lf', 'ha']

SquareLFLA

SquareLFLA(device: device = device('cpu'))

Bases: Square

Preset Square wave stream: low frequency and low amplitude.

Uses freq=0.03 Hz, ampl=0.5, phase=0.5, and delta=0.1 s. The stream name is set to "squLfLa" after construction.

The FEATURES class attribute ['square', 'lf', 'la'] can be used by processor code to discover this stream's signal characteristics.

Initialize a SquareLFLA stream with fixed low-frequency, low-amplitude parameters.

Parameters:

Name	Type	Description	Default
device	device	PyTorch device on which the output tensor is allocated. Defaults to torch.device('cpu').	device('cpu')

Source code in unaiverse/streams/streamlib.py

def __init__(self, device: torch.device = torch.device('cpu')) -> None:
    """Initialize a SquareLFLA stream with fixed low-frequency, low-amplitude parameters.

    Args:
        device: PyTorch device on which the output tensor is allocated. Defaults to
            ``torch.device('cpu')``.
    """
    super().__init__(freq=0.03, phase=0.5, ampl=0.5, delta=0.1, device=device)
    assert self.props is not None
    self.props.set_name("squLfLa")

FEATURES class-attribute instance-attribute

FEATURES = ['square', 'lf', 'la']