sgnts.transforms.correlate

AdaptiveCorrelate dataclass

Bases: Correlate

              flowchart TD
              sgnts.transforms.correlate.AdaptiveCorrelate[AdaptiveCorrelate]
              sgnts.transforms.correlate.Correlate[Correlate]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.transforms.correlate.Correlate --> sgnts.transforms.correlate.AdaptiveCorrelate
                                sgnts.base.base.TSTransform --> sgnts.transforms.correlate.Correlate
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                




              click sgnts.transforms.correlate.AdaptiveCorrelate href "" "sgnts.transforms.correlate.AdaptiveCorrelate"
              click sgnts.transforms.correlate.Correlate href "" "sgnts.transforms.correlate.Correlate"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"
            

Adaptive Correlate filter with Strategy Pattern for transitions.

This element implements the Adaptive Finite Impulse Response (AFIR) theory for streaming data. It manages a queue of filter updates arriving asynchronously and ensures mathematical coherence during transitions via four core design principles:

Principle 1 (Event-Driven Updates): Filters arrive as discrete events with a validity offset. The element maintains a chronologically sorted queue of these states.

Principle 2 (Last-Write-Wins): If multiple updates arrive for the same offset, the most recent reception overwrites the previous ones.

Principle 3 (Clock Coherence): Continuous-time validity offsets are mapped to discrete integer sample boundaries to prevent sub-sample phase artifacts.

Principle 4 (Stationarity Preservation): Stride processing is segmented into intervals of local stationarity (Discrete) or smooth blending (Adiabatic) to avoid unphysical transients.

Notes

Startup behavior (no explicit initial conditions). This element accepts filters=None. On startup, it emits gap buffers (no data) until a filter bank is received on the dedicated filters sink pad (filter_sink_name); the element reconfigures its shape and overlap from the first bank as it becomes active. Subsequent updates are blended over a stride as described below. During this gap startup, filter_dtype declares the output dtype before any filters exist.

Thread safety. Marked thread_safe = True. With Pipeline.run(threaded=N) the pad callbacks for this element are dispatched onto worker threads.

Pad layout: 2 sink pads (data + filter) + 1 source pad. The two sink pads' pull callbacks CAN run concurrently in the same wave; this is the per-pad concurrency to reason about. internal runs alone (single InternalPad).

Where the GIL-releasing work lives: internal() calls scipy.signal.correlate (and scipy.signal.windows.cosine during filter adaptation), both of which release the GIL, so significant speedup is expected when multiple correlation branches run in parallel.

Per-pad concurrency analysis:

• pull on the data sink pad: inherited TimeSeriesMixin.pull only; writes per-pad-keyed inbufs/metadata for the data pad. Does NOT touch self.filter_deque.
• pull on the filter sink pad: overridden; calls super().pull() (per-pad-keyed for the filter pad's own slot), then appends to self.filter_deque. The filter sink pad is the sole writer of self.filter_deque, so concurrent same-element pulls cannot produce a same-deque write race.
• internal: reads self.filter_deque and may popleft from it; runs alone (no concurrent reader or writer in the same wave).

Future editors MUST preserve thread safety:

1. Do NOT add new writers to self.filter_deque from any pull path other than the filter sink pad; that would introduce a same-deque write race across same-wave pulls.
2. Do NOT introduce additional element-level state mutated from pull paths outside of per-pad-keyed containers.
3. self.filters is reassigned during adaptation in internal(); that is fine because internal runs alone, but if you split adaptation logic into pull you must move self.filters to per-call local state.

Parameters:

Name	Type	Description	Default
filter_sink_name	str	Name of the sink pad receiving filter updates. Defaults to "filters".	'filters'
verbose	bool	Enables diagnostic logging of filter scheduling decisions.	False
transition_profile	TransitionProfile	A TransitionProfile instance defining the blending strategy between the old and new filters. Options are: CosSquaredTransition (default) DiscreteTransition ReverseDiscreteTransition LinearTransition PlanckTaperTransition	CosSquaredTransition()
filter_dtype	Optional[Any]	Optional[Any], the filter dtype. Must be specified if the filters differ from the dtype of the incoming data, so that the output dtype can be inferred correctly. If None, it is assumed that the output dtype is equal to the input dtype.	None

Raises:

Type	Description
ValueError	Raises a value error if more than one filter update is passed per stride, or if a filter update would change the output dtype mid-stream

Source code in src/sgnts/transforms/correlate.py

@dataclass(kw_only=True)
class AdaptiveCorrelate(Correlate):
    """
    Adaptive Correlate filter with Strategy Pattern for transitions.

    This element implements the Adaptive Finite Impulse Response (AFIR)
    theory for streaming data. It manages a queue of filter updates
    arriving asynchronously and ensures mathematical coherence during
    transitions via four core design principles:

    Principle 1 (Event-Driven Updates): Filters arrive as discrete events
    with a validity offset. The element maintains a chronologically
    sorted queue of these states.

    Principle 2 (Last-Write-Wins): If multiple updates arrive for the
    same offset, the most recent reception overwrites the previous ones.

    Principle 3 (Clock Coherence): Continuous-time validity offsets are
    mapped to discrete integer sample boundaries to prevent sub-sample
    phase artifacts.

    Principle 4 (Stationarity Preservation): Stride processing is
    segmented into intervals of local stationarity (Discrete) or
    smooth blending (Adiabatic) to avoid unphysical transients.

    Notes:
        **Startup behavior (no explicit initial conditions).** This element
        accepts filters=None. On startup, it emits gap buffers (no data) until a
        filter bank is received on the dedicated filters sink pad
        (filter_sink_name); the element reconfigures its shape and overlap from
        the first bank as it becomes active. Subsequent updates are blended over
        a stride as described below. During this gap startup, ``filter_dtype``
        declares the output dtype before any filters exist.

        **Thread safety.** Marked ``thread_safe = True``. With
        ``Pipeline.run(threaded=N)`` the pad callbacks for this element are
        dispatched onto worker threads.

        Pad layout: 2 sink pads (data + filter) + 1 source pad. The two sink
        pads' ``pull`` callbacks CAN run concurrently in the same wave; this is
        the per-pad concurrency to reason about. ``internal`` runs alone (single
        ``InternalPad``).

        Where the GIL-releasing work lives: ``internal()`` calls
        ``scipy.signal.correlate`` (and ``scipy.signal.windows.cosine`` during
        filter adaptation), both of which release the GIL, so significant
        speedup is expected when multiple correlation branches run in parallel.

        Per-pad concurrency analysis:

        - ``pull`` on the **data sink pad**: inherited ``TimeSeriesMixin.pull``
          only; writes per-pad-keyed ``inbufs``/``metadata`` for the data pad.
          Does NOT touch ``self.filter_deque``.
        - ``pull`` on the **filter sink pad**: overridden; calls
          ``super().pull()`` (per-pad-keyed for the filter pad's own slot), then
          appends to ``self.filter_deque``. The filter sink pad is the sole
          writer of ``self.filter_deque``, so concurrent same-element pulls
          cannot produce a same-deque write race.
        - ``internal``: reads ``self.filter_deque`` and may ``popleft`` from it;
          runs alone (no concurrent reader or writer in the same wave).

        **Future editors MUST preserve thread safety**:

        1. Do NOT add new writers to ``self.filter_deque`` from any ``pull``
           path other than the filter sink pad; that would introduce a
           same-deque write race across same-wave pulls.
        2. Do NOT introduce additional element-level state mutated from
           ``pull`` paths outside of per-pad-keyed containers.
        3. ``self.filters`` is reassigned during adaptation in ``internal()``;
           that is fine because ``internal`` runs alone, but if you split
           adaptation logic into ``pull`` you must move ``self.filters`` to
           per-call local state.

    Args:
        filter_sink_name: Name of the sink pad receiving filter updates.
            Defaults to "filters".
        verbose: Enables diagnostic logging of filter scheduling decisions.
        transition_profile: A TransitionProfile instance defining the
            blending strategy between the old and new filters. Options are:

            - CosSquaredTransition (default)
            - DiscreteTransition
            - ReverseDiscreteTransition
            - LinearTransition
            - PlanckTaperTransition
        filter_dtype: Optional[Any], the filter dtype. Must be specified if
            the filters differ from the dtype of the incoming data, so that
            the output dtype can be inferred correctly. If None, it is
            assumed that the output dtype is equal to the input dtype.

    Raises:
        ValueError:
            Raises a value error if more than one filter update is passed per
            stride, or if a filter update would change the output dtype
            mid-stream
    """

    thread_safe = True

    filter_sink_name: str = "filters"
    verbose: bool = False
    transition_profile: TransitionProfile = field(default_factory=CosSquaredTransition)
    filter_dtype: Optional[Any] = None

    @property
    def static_unaligned_sink_pads(self) -> list[str]:  # type: ignore[override]
        """
        Mark the filters pad as asynchronous.
        This prevents the audio thread from blocking if filter updates are slow.
        """
        return [self.filter_sink_name]

    @property
    def static_sink_pads(self) -> list[str]:  # type: ignore[override]
        # Required for framework pad validation.
        return [self.filter_sink_name]

    def configure(self) -> None:
        # Standard Correlate configuration handles audio alignment.
        super().configure()

        # Setup the secondary filter input sink.
        self.filter_pad = self.snks[self.filter_sink_name]
        self.input_frame_types[self.filter_sink_name] = EventFrame
        self.filter_deque: Deque[FilterState] = deque()

        # Seed initial state if filters were provided at init.
        if self.filters is not None:
            self.filter_deque.append(
                FilterState(
                    offset=0, noffset=TIME_MAX, data=numpy.asarray(self.filters)
                )
            )

    def validate(self) -> None:
        # Overrides base Correlate's @validator.one_to_one: this element has a
        # second (filter) sink pad, so it validates its own aligned-pad count.
        # filters=None is permitted (gaps-until-first-filter startup).
        assert len(self.aligned_sink_pads) == 1 and len(self.source_pads) == 1, (
            f"Correlate requires exactly one aligned sink pad and one "
            f"source pad, got {len(self.aligned_sink_pads)} aligned sink "
            f"pads and {len(self.source_pads)} source pads"
        )
        assert self.sample_rate != -1, "Sample rate must be specified (not -1)"

    def output_prototype(self, pad: SourcePad) -> Array:
        # Correlate derives the prototype from real filters when present,
        # else inherits the input; filter_dtype only layers on top of that.
        derived = super().output_prototype(pad)
        if self.filter_dtype is None:
            # No dtype information for filters available, assume that
            # input dtype is not changed by them
            return derived
        if self.filters is not None:
            # Derived from real filters: flag a conflicting declaration
            if derived.dtype != self.filter_dtype:
                raise ValueError(
                    f"filter_dtype {self.filter_dtype!r} conflicts with dtype "
                    f"{derived.dtype!r} inferred from filters; either remove "
                    f"filter_dtype or change filters to the declared dtype"
                )
            return derived
        # Gap startup: apply the declared dtype to the inherited prototype
        xp = array_namespace(derived)
        assert xp is not None
        return xp.zeros((0,), dtype=self.filter_dtype) * derived

    @property
    def filters_cur(self) -> Optional[Array]:
        """Returns the coefficients of the active (oldest) filter."""
        return self.filter_deque[0].data if self.filter_deque else None

    def _extract_all_filters(self, frame: TimeSpanFrame) -> list[Tuple[int, Array]]:
        """
        Extract all valid filter coefficients and their offsets from an EventFrame.
        """
        results: list[Tuple[int, Array]] = []
        if not frame or not frame.data:
            return results
        for buf in frame:
            if not isinstance(buf, EventBuffer):
                continue
            for event in buf:
                if event.data is not None:
                    results.append((buf.offset, numpy.asarray(event.data)))
        return results

    def pull(self, pad: SinkPad, frame: TimeSpanFrame) -> None:
        """
        Pull data from sinks. Handles asynchronous filter updates.
        """
        # Always delegate to parent to ensure main audio sink processing.
        super().pull(pad, frame)

        # Identify filter updates by comparing pad names
        if pad.name != self.filter_pad.name or frame.is_gap:
            return

        updates = self._extract_all_filters(frame)
        if not updates:
            return

        # A filter update must not change the output shape or dtype: the
        # stream spec is locked on the first output frame, so a mid-stream
        # change would crash downstream with an opaque spec mismatch. Compare
        # against the current active filter, or the declared filter_dtype
        # during gap startup (no filters yet).
        cur = self.filters_cur
        expected_dtype = cur.dtype if cur is not None else self.filter_dtype

        for offset, data in updates:
            if cur is not None and cur.shape != data.shape:
                raise ValueError(
                    "New filters must have the same shape as existing filters"
                )
            if expected_dtype is not None:
                xp = array_namespace(data)
                assert xp is not None
                if xp.result_type(data) != xp.result_type(expected_dtype):
                    raise ValueError(
                        f"New filters have dtype {data.dtype}, which would "
                        f"change the output dtype mid-stream (expected: "
                        f"{xp.result_type(expected_dtype)}). Provide filters "
                        f"matching the dtype of the initial filters, or the "
                        f"declared filter_dtype."
                    )

            # Enforcement of Principle 2 (LWW):
            # Replace any existing filter at exactly the same offset.
            self.filter_deque = deque(
                f for f in self.filter_deque if f.offset != offset
            )

            # Add to queue
            self.filter_deque.append(
                FilterState(offset=offset, noffset=TIME_MAX, data=data)
            )

        # Enforcement of Principle 1: maintain chronologically sorted queue.
        self.filter_deque = deque(sorted(self.filter_deque, key=lambda f: f.offset))

    def internal(self) -> None:
        """
        Adaptive internal loop: manages filter lifecycle and handles mid-stride
        transitions.
        """
        # Call TSTransform.internal (skipping Correlate.internal) to perform alignment.
        super(Correlate, self).internal()
        if not self._is_aligned:
            return

        try:
            _, input_frame = self.next_input()
            _, output_collector = self.next_output()
        except (ValueError, IndexError, StopIteration):
            return

        T_out_start = output_collector.offset
        T_out_end = output_collector.offset + output_collector.noffset

        # Enforcement of Principle 1 (Handover Pruning):
        # We prune filters that have expired before the current stride begins.
        # We keep the oldest filter if its validity window abuts or overlaps
        # the current stride, allowing for transitions into the next state.
        while len(self.filter_deque) > 1:
            f1 = self.filter_deque[1]
            if f1.offset < T_out_start:
                self.filter_deque.popleft()
            else:
                break

        # Identify all filters valid for the current audio window [T_start, T_end).
        intersecting_filters = []
        for i, f in enumerate(self.filter_deque):
            f_start = f.offset
            f_end = (
                self.filter_deque[i + 1].offset
                if i < len(self.filter_deque) - 1
                else TIME_MAX
            )
            if f_start < T_out_end and f_end >= T_out_start:
                intersecting_filters.append((f, f_start, f_end))

        if len(intersecting_filters) <= 1:
            # Case 1: Steady-state or Zero-Order Hold.
            active_f = next(
                (f for f, fs, fe in intersecting_filters if fs <= T_out_start < fe),
                None,
            )
            if active_f is None:
                active_f = (
                    intersecting_filters[0][0]
                    if intersecting_filters
                    else (self.filter_deque[0] if self.filter_deque else None)
                )

            if active_f:
                self.filters = active_f.data
                self.shape = self.filters.shape
                self._transform(input_frame, output_collector)
            else:
                self.filters = None
                self._transform(input_frame, output_collector)
        else:
            # Case 2: Transition detected within the current stride.
            # Derive the array-API namespace from the intersecting filter data
            # (numpy for this element's declared backend); fall back to numpy.
            backend = (
                array_namespace(*(f.data for f, _fs, _fe in intersecting_filters))
                or numpy
            )
            self._process_adaptive(
                input_frame,
                output_collector,
                intersecting_filters,
                backend,
                T_out_start,
                T_out_end,
            )

        # Ensure frame is fully populated to its expected duration
        if not output_collector._buffers:
            # Entirely empty: generate a single gap buffer spanning the stride
            out_samples = Offset.tosamples(output_collector.noffset, self.sample_rate)
            out_buf = SeriesBuffer(
                offset=output_collector.offset,
                sample_rate=self.sample_rate,
                data=None,
                shape=self.shape[:-1] + (out_samples,),
            )
            out_buf = out_buf.copy(is_gap=True)
            output_collector.append(out_buf)
        else:
            # Check if last buffer reaches the end of the frame
            frame_end = output_collector.offset + output_collector.noffset
            last_end = output_collector._buffers[-1].end_offset
            if last_end < frame_end:
                gap_noffset = frame_end - last_end
                out_samples = Offset.tosamples(gap_noffset, self.sample_rate)
                out_buf = SeriesBuffer(
                    offset=last_end,
                    sample_rate=self.sample_rate,
                    data=None,
                    shape=self.shape[:-1] + (out_samples,),
                )
                out_buf = out_buf.copy(is_gap=True)
                output_collector.append(out_buf)

        output_collector.close()

    def _process_adaptive(
        self,
        input_frame: TSFrame,
        output_collector: TSCollectFrame,
        intersecting_filters: list[Tuple[FilterState, int, int]],
        backend: Any,
        T_out_start: int,
        T_out_end: int,
    ):
        """
        Unified adaptive processor implementing Principle 4.

        This method segments the current audio stride into computation intervals
        where the filter state is either stationary or undergoing a transition.
        """
        from sgnts.base.slice_tools import TSSlice

        profile = self.transition_profile
        is_discrete = isinstance(profile, DiscreteTransition)

        # 1. Define computation segments [seg_start, seg_end, f_old, f_new, prof]
        segments: list[
            Tuple[
                int,
                int,
                Optional[FilterState],
                Optional[FilterState],
                TransitionProfile,
            ]
        ] = []
        if is_discrete:
            # Split the stride into hard segments based on filter arrival boundaries.
            boundaries = sorted(
                list(
                    set(
                        f_start
                        for _, f_start, _ in intersecting_filters
                        if T_out_start < f_start < T_out_end
                    )
                )
            )
            curr = T_out_start
            for b in boundaries:
                # Find active filter for this hard stationary segment.
                active_f, _, _ = next(
                    item for item in intersecting_filters if item[1] <= curr < item[2]
                )
                segments.append((curr, b, None, active_f, DiscreteTransition()))
                curr = b
            # Last stationary segment of the current stride.
            active_f, _, _ = next(
                item for item in intersecting_filters if item[1] <= curr < item[2]
            )
            segments.append((curr, T_out_end, None, active_f, DiscreteTransition()))
        else:
            # Continuous: exactly one segment for the whole stride (smooth crossfade).
            if len(intersecting_filters) > 2:
                warnings.warn(
                    "Rapid filter updates detected.", RuntimeWarning, stacklevel=2
                )
            f_cur, _, _ = intersecting_filters[0]
            f_new, _, _ = intersecting_filters[-1]
            segments.append((T_out_start, T_out_end, f_cur, f_new, profile))

        # 2. Unified Stride Loop: process each segment sequentially.
        for seg_start, seg_end, s_f_old, s_f_new, s_prof in segments:
            # Enforcement of Principle 3 (Clock Coherence):
            # Quantize GPS boundaries to the nearest integer sample relative
            # to global max rate to prevent sub-sample phase artifacts.
            stride = Offset.MAX_RATE // self.sample_rate
            s1, s2 = int(round(seg_start / stride)), int(round(seg_end / stride))
            t1, t2 = s1 * stride, s2 * stride
            if t1 >= t2:
                continue

            # Determine dominant filter for alignment parameters.
            f_active = s_f_new if s_f_new is not None else s_f_old
            if f_active:
                self.shape = f_active.data.shape
            overlap_samples = max(0, self.shape[-1] - 1)
            overlap_offsets = Offset.fromsamples(overlap_samples, self.sample_rate)
            comp_samples = int(round(overlap_samples - self.latency))
            comp_offsets = Offset.fromsamples(comp_samples, self.sample_rate)

            # Define the required audio slice for this segment.
            # We must pull (N-1) samples of history BEFORE the segment start.
            req_slice = TSSlice(t1 - overlap_offsets + comp_offsets, t2 + comp_offsets)

            for buf in input_frame:
                sl = buf.slice & req_slice
                # Defensive: the adapter delivers exactly the segment's audio
                # window, so every buffer intersects req_slice.
                if not sl or not sl.isfinite():  # pragma: no cover
                    continue

                sub_buf = buf.sub_buffer(sl)
                out_samples = (
                    Offset.tosamples(sub_buf.noffset, self.sample_rate)
                    - overlap_samples
                )
                # Defensive: segments shorter than the overlap are already
                # dropped by the t1 >= t2 quantization guard above.
                if out_samples <= 0:  # pragma: no cover
                    continue

                # Output offset account for internal convolver delay (shift).
                out_offset = sub_buf.offset + (overlap_offsets - comp_offsets)

                if sub_buf.is_gap:
                    out_buf = sub_buf.copy(
                        offset=out_offset,
                        data=None,
                        shape=self.shape[:-1] + (out_samples,),
                        is_gap=True,
                    )
                    output_collector.append(out_buf)
                    continue

                # --- Core Weighted Sum Math ---
                data_to_corr = sub_buf.data

                # Compute output from previous state (if required by profile).
                data_old = None
                if s_f_old is not None and not s_prof.skip_old:
                    self.filters = s_f_old.data
                    self.shape = self.filters.shape
                    data_old = self.corr(data_to_corr)[..., :out_samples]

                # Compute output from target state.
                data_new = None
                if s_f_new is not None and not s_prof.skip_new:
                    self.filters = s_f_new.data
                    self.shape = self.filters.shape
                    data_new = self.corr(data_to_corr)[..., :out_samples]

                # Blend outputs based on Strategy Pattern.
                w_old, w_new = s_prof.get_weights(out_samples, backend)

                if data_old is None:
                    data = w_new * data_new
                elif data_new is None:
                    data = w_old * data_old
                else:
                    data = w_old * data_old + w_new * data_new

                # Latch the target filter state for future stationary strides.
                self.filters = (
                    s_f_new.data if s_f_new else (s_f_old.data if s_f_old else None)
                )
                if self.filters is not None:
                    self.shape = self.filters.shape

                out_buf = sub_buf.copy(
                    offset=out_offset, data=data, shape=data.shape, is_gap=False
                )
                output_collector.append(out_buf)

filters_cur property

Returns the coefficients of the active (oldest) filter.

static_unaligned_sink_pads property

Mark the filters pad as asynchronous. This prevents the audio thread from blocking if filter updates are slow.

internal()

Adaptive internal loop: manages filter lifecycle and handles mid-stride transitions.

Source code in src/sgnts/transforms/correlate.py

def internal(self) -> None:
    """
    Adaptive internal loop: manages filter lifecycle and handles mid-stride
    transitions.
    """
    # Call TSTransform.internal (skipping Correlate.internal) to perform alignment.
    super(Correlate, self).internal()
    if not self._is_aligned:
        return

    try:
        _, input_frame = self.next_input()
        _, output_collector = self.next_output()
    except (ValueError, IndexError, StopIteration):
        return

    T_out_start = output_collector.offset
    T_out_end = output_collector.offset + output_collector.noffset

    # Enforcement of Principle 1 (Handover Pruning):
    # We prune filters that have expired before the current stride begins.
    # We keep the oldest filter if its validity window abuts or overlaps
    # the current stride, allowing for transitions into the next state.
    while len(self.filter_deque) > 1:
        f1 = self.filter_deque[1]
        if f1.offset < T_out_start:
            self.filter_deque.popleft()
        else:
            break

    # Identify all filters valid for the current audio window [T_start, T_end).
    intersecting_filters = []
    for i, f in enumerate(self.filter_deque):
        f_start = f.offset
        f_end = (
            self.filter_deque[i + 1].offset
            if i < len(self.filter_deque) - 1
            else TIME_MAX
        )
        if f_start < T_out_end and f_end >= T_out_start:
            intersecting_filters.append((f, f_start, f_end))

    if len(intersecting_filters) <= 1:
        # Case 1: Steady-state or Zero-Order Hold.
        active_f = next(
            (f for f, fs, fe in intersecting_filters if fs <= T_out_start < fe),
            None,
        )
        if active_f is None:
            active_f = (
                intersecting_filters[0][0]
                if intersecting_filters
                else (self.filter_deque[0] if self.filter_deque else None)
            )

        if active_f:
            self.filters = active_f.data
            self.shape = self.filters.shape
            self._transform(input_frame, output_collector)
        else:
            self.filters = None
            self._transform(input_frame, output_collector)
    else:
        # Case 2: Transition detected within the current stride.
        # Derive the array-API namespace from the intersecting filter data
        # (numpy for this element's declared backend); fall back to numpy.
        backend = (
            array_namespace(*(f.data for f, _fs, _fe in intersecting_filters))
            or numpy
        )
        self._process_adaptive(
            input_frame,
            output_collector,
            intersecting_filters,
            backend,
            T_out_start,
            T_out_end,
        )

    # Ensure frame is fully populated to its expected duration
    if not output_collector._buffers:
        # Entirely empty: generate a single gap buffer spanning the stride
        out_samples = Offset.tosamples(output_collector.noffset, self.sample_rate)
        out_buf = SeriesBuffer(
            offset=output_collector.offset,
            sample_rate=self.sample_rate,
            data=None,
            shape=self.shape[:-1] + (out_samples,),
        )
        out_buf = out_buf.copy(is_gap=True)
        output_collector.append(out_buf)
    else:
        # Check if last buffer reaches the end of the frame
        frame_end = output_collector.offset + output_collector.noffset
        last_end = output_collector._buffers[-1].end_offset
        if last_end < frame_end:
            gap_noffset = frame_end - last_end
            out_samples = Offset.tosamples(gap_noffset, self.sample_rate)
            out_buf = SeriesBuffer(
                offset=last_end,
                sample_rate=self.sample_rate,
                data=None,
                shape=self.shape[:-1] + (out_samples,),
            )
            out_buf = out_buf.copy(is_gap=True)
            output_collector.append(out_buf)

    output_collector.close()

pull(pad, frame)

Pull data from sinks. Handles asynchronous filter updates.

Source code in src/sgnts/transforms/correlate.py

def pull(self, pad: SinkPad, frame: TimeSpanFrame) -> None:
    """
    Pull data from sinks. Handles asynchronous filter updates.
    """
    # Always delegate to parent to ensure main audio sink processing.
    super().pull(pad, frame)

    # Identify filter updates by comparing pad names
    if pad.name != self.filter_pad.name or frame.is_gap:
        return

    updates = self._extract_all_filters(frame)
    if not updates:
        return

    # A filter update must not change the output shape or dtype: the
    # stream spec is locked on the first output frame, so a mid-stream
    # change would crash downstream with an opaque spec mismatch. Compare
    # against the current active filter, or the declared filter_dtype
    # during gap startup (no filters yet).
    cur = self.filters_cur
    expected_dtype = cur.dtype if cur is not None else self.filter_dtype

    for offset, data in updates:
        if cur is not None and cur.shape != data.shape:
            raise ValueError(
                "New filters must have the same shape as existing filters"
            )
        if expected_dtype is not None:
            xp = array_namespace(data)
            assert xp is not None
            if xp.result_type(data) != xp.result_type(expected_dtype):
                raise ValueError(
                    f"New filters have dtype {data.dtype}, which would "
                    f"change the output dtype mid-stream (expected: "
                    f"{xp.result_type(expected_dtype)}). Provide filters "
                    f"matching the dtype of the initial filters, or the "
                    f"declared filter_dtype."
                )

        # Enforcement of Principle 2 (LWW):
        # Replace any existing filter at exactly the same offset.
        self.filter_deque = deque(
            f for f in self.filter_deque if f.offset != offset
        )

        # Add to queue
        self.filter_deque.append(
            FilterState(offset=offset, noffset=TIME_MAX, data=data)
        )

    # Enforcement of Principle 1: maintain chronologically sorted queue.
    self.filter_deque = deque(sorted(self.filter_deque, key=lambda f: f.offset))

Correlate dataclass

Bases: TSTransform

              flowchart TD
              sgnts.transforms.correlate.Correlate[Correlate]
              sgnts.base.base.TSTransform[TSTransform]
              sgnts.base.base.TimeSeriesMixin[TimeSeriesMixin]

                              sgnts.base.base.TSTransform --> sgnts.transforms.correlate.Correlate
                                sgnts.base.base.TimeSeriesMixin --> sgnts.base.base.TSTransform
                



              click sgnts.transforms.correlate.Correlate href "" "sgnts.transforms.correlate.Correlate"
              click sgnts.base.base.TSTransform href "" "sgnts.base.base.TSTransform"
              click sgnts.base.base.TimeSeriesMixin href "" "sgnts.base.base.TimeSeriesMixin"
            

Correlates input data with a fixed or dynamic filter.

This element performs a standard multi-channel correlation: Out = Data * Filter. It uses the sgn-ts AudioAdapter to manage overlap-save convolution history, ensuring that N-1 samples of history are maintained across stride boundaries.

Parameters:

Name	Type	Description	Default
sample_rate	int	The audio sample rate in Hz.	required
filters	Optional[Array]	Initial filter coefficients. Should be shape (channels, taps). If None, the element defaults to (1, 1) until an update arrives.	None
latency	int	The output timing offset, in samples: an output computed from input up to time t is timestamped t - latency. The value that aligns a feature with its true time depends on the filter's group delay (see Notes). It is a physical (timestamp) latency in the output channel, not an algorithmic delay in this element. Common choices for an N-tap filter: 0: causal, for a minimum-phase filter (energy at the front). (N - 1) / 2: symmetric, linear-phase filter. N - 1: anti-causal (energy at the back).	0
taps_reversed	bool	If True, the filter taps are pre-reversed so that scipy.correlate(data, taps_reversed) = convolution with original taps. This is standard for asymmetric (e.g., minimum-phase) FIR filters. When True, the temporal invariant is adjusted so that latency=0 still means "causal / zero-latency".	False
pad_zeros	bool	boolean, whether to pad with zeros on startup or not. This determines whether the pipeline stalls initially or just starts producing values (with data zero-padded to match the filter length).	False
method	Literal['auto', 'direct', 'fft']	str, the convolution backend passed to scipy.signal.correlate: "auto" (default), "direct", or "fft". scipy's "auto" heuristic can mis-select for overlap-save input shapes at small strides (picking "direct" when "fft" is several times faster for long filters), so exposing this lets callers force the fast path.	'auto'

Notes

Latency and output timing. latency shifts the timestamps written on the output buffers. Two distinct kinds of latency must be kept apart:

• Physical (timestamp) latency is introduced here, and is real. The published output is labeled latency samples behind, so a testpoint on the output frame, or a channel sent to a sink, reports exactly that latency, whatever follows this element.
• Algorithmic latency (waiting in the execution loop) is not introduced here. The element runs the same valid-mode correlation over history the adapter already holds, leaves the sample values unchanged, and emits each buffer as soon as it otherwise would; nothing waits. A downstream element that synchronizes its inputs by timestamp (via AdapterConfig) may wait for a relabeled buffer to line up, but that is a separate, downstream effect, not always enforced.

Each output depends on a window of N input samples, and latency chooses which one dates it: 0 the newest, N - 1 the oldest, (N - 1) / 2 the center. The linear-phase value (N - 1) / 2 places a symmetric filter's features at their true time; the reference covers the general, frequency-dependent case.

For the underlying signal processing, see the Signal Processing Fundamentals reference: Latency defines the physical and algorithmic kinds with worked examples, and Phase Delay and Group Delay explains why the delay tau is in general frequency dependent.

Thread safety. Marked thread_safe = True. With Pipeline.run(threaded=N) the pad callbacks (pull, new, internal) for this element are dispatched onto worker threads.

Pad layout: 1 sink + 1 source pad (enforced by @validator.one_to_one). There is therefore no same-element pull/new concurrency to worry about: only one pull and one new ever run at a time on this element. internal always runs alone (single InternalPad).

Where the GIL-releasing work lives: internal() calls scipy.signal.correlate, which releases the GIL, so this element delivers significant wall-clock speedup when there are multiple parallel correlation branches in the graph and their elements all opt in.

State touched per call:

• pull (inherited TimeSeriesMixin.pull): writes per-pad-keyed dicts (inbufs[pad], metadata[pad]); ORs self.at_EOS (idempotent for booleans).
• new (inherited TSTransform.new): read-only lookup in self.outframes.
• internal: reads self.filters (set in configure() and read-only afterwards) and self.shape; writes the next output frame.

Future editors MUST preserve thread safety: do not relax the one-to-one constraint without re-auditing self.filters access. Do not introduce element-level state that is mutated from pull/new outside of per-pad-keyed containers.

Source code in src/sgnts/transforms/correlate.py

@dataclass(kw_only=True)
class Correlate(TSTransform):
    """
    Correlates input data with a fixed or dynamic filter.

    This element performs a standard multi-channel correlation: Out = Data * Filter.
    It uses the sgn-ts AudioAdapter to manage overlap-save convolution history,
    ensuring that N-1 samples of history are maintained across stride boundaries.

    Args:
        sample_rate: The audio sample rate in Hz.
        filters: Initial filter coefficients. Should be shape (channels, taps).
            If None, the element defaults to (1, 1) until an update arrives.
        latency: The output timing offset, in samples: an output computed from
            input up to time ``t`` is timestamped ``t - latency``. The value
            that aligns a feature with its true time depends on the filter's
            group delay (see Notes). It is a physical (timestamp) latency in the
            output channel, not an algorithmic delay in this element. Common
            choices for an N-tap filter:

            - ``0``: causal, for a minimum-phase filter (energy at the front).
            - ``(N - 1) / 2``: symmetric, linear-phase filter.
            - ``N - 1``: anti-causal (energy at the back).
        taps_reversed: If True, the filter taps are pre-reversed so that
            scipy.correlate(data, taps_reversed) = convolution with original taps.
            This is standard for asymmetric (e.g., minimum-phase) FIR filters.
            When True, the temporal invariant is adjusted so that latency=0
            still means "causal / zero-latency".
        pad_zeros:
            boolean, whether to pad with zeros on startup or not. This
            determines whether the pipeline stalls initially or just
            starts producing values (with data zero-padded to match the
            filter length).
        method:
            str, the convolution backend passed to scipy.signal.correlate:
            "auto" (default), "direct", or "fft". scipy's "auto" heuristic
            can mis-select for overlap-save input shapes at small strides
            (picking "direct" when "fft" is several times faster for long
            filters), so exposing this lets callers force the fast path.

    Notes:
        **Latency and output timing.** ``latency`` shifts the timestamps
        written on the output buffers. Two distinct kinds of latency must be
        kept apart:

        - *Physical (timestamp) latency* is introduced here, and is real. The
          published output is labeled ``latency`` samples behind, so a
          testpoint on the output frame, or a channel sent to a sink, reports
          exactly that latency, whatever follows this element.
        - *Algorithmic latency* (waiting in the execution loop) is not
          introduced here. The element runs the same valid-mode correlation
          over history the adapter already holds, leaves the sample values
          unchanged, and emits each buffer as soon as it otherwise would;
          nothing waits. A downstream element that synchronizes its inputs by
          timestamp (via ``AdapterConfig``) may wait for a relabeled buffer to
          line up, but that is a separate, downstream effect, not always
          enforced.

        Each output depends on a window of ``N`` input samples, and ``latency``
        chooses which one dates it: ``0`` the newest, ``N - 1`` the oldest,
        ``(N - 1) / 2`` the center. The linear-phase value ``(N - 1) / 2``
        places a symmetric filter's features at their true time; the reference
        covers the general, frequency-dependent case.

        For the underlying signal processing, see the
        [Signal Processing Fundamentals](../../signal-processing/index.md)
        reference: [Latency](../../signal-processing/latency.md) defines the
        physical and algorithmic kinds with worked examples, and
        [Phase Delay and Group Delay](../../signal-processing/group-delay.md)
        explains why the delay ``tau`` is in general frequency dependent.

        **Thread safety.** Marked ``thread_safe = True``. With
        ``Pipeline.run(threaded=N)`` the pad callbacks (``pull``, ``new``,
        ``internal``) for this element are dispatched onto worker threads.

        Pad layout: 1 sink + 1 source pad (enforced by
        ``@validator.one_to_one``). There is therefore no same-element
        ``pull``/``new`` concurrency to worry about: only one ``pull`` and one
        ``new`` ever run at a time on this element. ``internal`` always runs
        alone (single ``InternalPad``).

        Where the GIL-releasing work lives: ``internal()`` calls
        ``scipy.signal.correlate``, which releases the GIL, so this element
        delivers significant wall-clock speedup when there are multiple parallel
        correlation branches in the graph and their elements all opt in.

        State touched per call:

        - ``pull`` (inherited ``TimeSeriesMixin.pull``): writes per-pad-keyed
          dicts (``inbufs[pad]``, ``metadata[pad]``); ORs ``self.at_EOS``
          (idempotent for booleans).
        - ``new`` (inherited ``TSTransform.new``): read-only lookup in
          ``self.outframes``.
        - ``internal``: reads ``self.filters`` (set in ``configure()`` and
          read-only afterwards) and ``self.shape``; writes the next output
          frame.

        **Future editors MUST preserve thread safety**: do not relax the
        one-to-one constraint without re-auditing ``self.filters`` access. Do
        not introduce element-level state that is mutated from ``pull``/``new``
        outside of per-pad-keyed containers.
    """

    thread_safe = True

    # scipy.signal.correlate / windows (numpy). Inherited by AdaptiveCorrelate.
    backends = frozenset({"numpy"})

    sample_rate: int
    filters: Optional[Array] = None
    latency: int = 0
    taps_reversed: bool = False
    pad_zeros: bool = False
    method: Literal["auto", "direct", "fft"] = "auto"

    def configure(self) -> None:
        """
        Setup alignment based on filter length (N) and user latency property.

        The SGN-TS temporal invariant is: T_out = T_in + (N-1) - L
        Where L is the compensation (shift).
        To achieve T_out = T_in + self.latency, we set L = (N-1) - self.latency.

        When taps_reversed=True, pre-reversed taps + scipy.correlate = convolution.
        Convolution reverses the filter's time axis, so a causal kernel (energy at
        index 0) appears anti-causal (energy at index N-1) to the correlator. We
        compensate by mapping: effective_latency = (N-1) - user_latency.
        This way latency=0 means "zero latency" regardless of tap ordering.
        """
        if self.filters is not None:
            self.shape = self.filters.shape
            overlap_samples = max(0, self.shape[-1] - 1)
        else:
            # Null Initial Condition:
            self.shape = (1, 1)
            overlap_samples = 0

        # When taps are pre-reversed (correlate = convolution), the correlator
        # sees the energy at the opposite end of the kernel. Remap latency so
        # that the user's intent (0 = causal) is preserved.
        effective_latency = self.latency
        if self.taps_reversed and overlap_samples > 0:
            effective_latency = overlap_samples - self.latency

        # Alignment logic:
        # 'overlap' ensures we have (N-1) samples of history for 'valid' correlation.
        # 'L' (compensation) = (N-1) - effective_latency
        comp_samples = overlap_samples - effective_latency

        self.adapter_config.alignment(
            overlap=(Offset.fromsamples(overlap_samples, self.sample_rate), 0),
            shift=-Offset.fromsamples(comp_samples, self.sample_rate),
        )
        # Pad with zeros on startup only if requested; otherwise the adapter
        # stalls until it has real history.
        self.adapter_config.on_startup(pad_zeros=self.pad_zeros)

        self.sink_pad = self.sink_pads[0]
        self.source_pad = self.source_pads[0]

    @validator.one_to_one
    def validate(self) -> None:
        pass

    def output_prototype(self, pad: SourcePad) -> Array:
        proto = self.input_prototype(self.sink_pad.pad_name)
        if self.filters is not None:
            xp = array_namespace(self.filters)
            assert xp is not None
            return xp.reshape(self.filters, (-1,))[:0] * proto
        # No filters, assume that input dtype is not changed by them
        return proto

    def corr(self, data: Array) -> Array:
        """
        Perform the mathematical correlation.

        Uses scipy.signal.correlate(mode='valid') which effectively
        reverses the filter taps. ``self.method`` selects scipy's
        convolution backend ("auto"/"direct"/"fft").
        """
        if self.filters is None:
            raise ValueError("Cannot correlate without filters")

        if len(self.filters.shape) == 1:
            return scipy.signal.correlate(
                data, self.filters, mode="valid", method=self.method
            )

        # Row count and output shape come from the filters in hand so that
        # >= 3-dim banks apply every filter row (not just shape[0] of the
        # flattened rows). The reshape is local: repeated calls (and the
        # documented read-only ``self.filters`` contract) keep the original
        # bank.
        shape = self.filters.shape
        if shape[-1] != self.shape[-1]:
            # Defensive: the element's output span and adapter overlap were
            # configured for self.shape; correlating with differently-sized
            # filters would silently produce wrong results.
            raise ValueError(
                f"filters of shape {shape} are inconsistent with the "
                f"configured filter shape {self.shape}; the element's "
                "output span and adapter overlap were built for the "
                "configured shape"
            )
        f_reshaped = numpy.asarray(self.filters).reshape(-1, shape[-1])
        os = []
        for j in range(f_reshaped.shape[0]):
            # Multi-channel data: map filter row j to data channel j. A single
            # (1-D) data stream is correlated against every filter row
            # (filter-bank semantics).
            d = data[j] if len(data.shape) > 1 else data
            os.append(
                scipy.signal.correlate(
                    d, f_reshaped[j], mode="valid", method=self.method
                )
            )
        return numpy.vstack(os).reshape(shape[:-1] + (-1,))

    def _transform(self, input_frame: TSFrame, output_frame: TSCollectFrame) -> None:
        """
        Core transform implementation for non-adaptive segments.

        Iterates through input buffers and applies the current active filter.
        When the adapter provides overlap-padded buffers, correlation mode='valid'
        naturally contracts non-gap data to the correct output size. Gap buffers
        must be explicitly capped to the output frame's remaining capacity since
        they bypass correlation.
        """
        curr_offset = output_frame.offset
        output_end = output_frame.offset + output_frame.noffset
        for buf in input_frame:
            assert buf.sample_rate == self.sample_rate
            out_samples = Offset.tosamples(buf.noffset, self.sample_rate)

            if buf.is_gap or self.filters is None:
                # Cap gap duration to remaining output frame capacity.
                # The adapter's overlap padding inflates input buffers beyond
                # the expected output duration; correlation handles this for
                # data, but gaps must be capped explicitly.
                remaining = Offset.tosamples(output_end - curr_offset, self.sample_rate)
                gap_samples = min(out_samples, remaining)
                if gap_samples <= 0:
                    break
                out_buf = buf.copy(
                    offset=curr_offset,
                    data=None,
                    shape=self.shape[:-1] + (gap_samples,),
                    is_gap=True,
                )
                output_frame.append(out_buf)
                curr_offset += Offset.fromsamples(gap_samples, self.sample_rate)
            else:
                # Perform correlation.
                data = self.corr(buf.data)
                # Slice to the specific duration of this buffer.
                data = data[..., :out_samples]
                out_buf = buf.copy(
                    offset=curr_offset,
                    data=data,
                    shape=data.shape,
                    is_gap=False,
                )
                output_frame.append(out_buf)
                curr_offset += buf.noffset

    def internal(self) -> None:
        """
        Standard SGN-TS internal loop: pulls aligned input and executes transform.
        """
        super().internal()
        _, output_collector = self.next_output()
        _, input_frame = self.next_input()
        self._transform(input_frame, output_collector)

        # Ensure frame is fully populated to its expected duration
        if not output_collector._buffers:
            # Entirely empty: generate a single gap buffer spanning the stride
            out_samples = Offset.tosamples(output_collector.noffset, self.sample_rate)
            out_buf = SeriesBuffer(
                offset=output_collector.offset,
                sample_rate=self.sample_rate,
                data=None,
                shape=self.shape[:-1] + (out_samples,),
            )
            out_buf = out_buf.copy(is_gap=True)
            output_collector.append(out_buf)
        else:
            # Check if last buffer reaches the end of the frame
            frame_end = output_collector.offset + output_collector.noffset
            last_end = output_collector._buffers[-1].end_offset
            # Defensive: the aligned adapter fills the frame exactly, so the
            # last produced buffer already reaches frame_end. Kept as a safety
            # net for partial fills but unreachable via the normal pipeline.
            if last_end < frame_end:  # pragma: no cover
                gap_noffset = frame_end - last_end
                out_samples = Offset.tosamples(gap_noffset, self.sample_rate)
                out_buf = SeriesBuffer(
                    offset=last_end,
                    sample_rate=self.sample_rate,
                    data=None,
                    shape=self.shape[:-1] + (out_samples,),
                )
                out_buf = out_buf.copy(is_gap=True)
                output_collector.append(out_buf)

        output_collector.close()

configure()

Setup alignment based on filter length (N) and user latency property.

The SGN-TS temporal invariant is: T_out = T_in + (N-1) - L Where L is the compensation (shift). To achieve T_out = T_in + self.latency, we set L = (N-1) - self.latency.

When taps_reversed=True, pre-reversed taps + scipy.correlate = convolution. Convolution reverses the filter's time axis, so a causal kernel (energy at index 0) appears anti-causal (energy at index N-1) to the correlator. We compensate by mapping: effective_latency = (N-1) - user_latency. This way latency=0 means "zero latency" regardless of tap ordering.

Source code in src/sgnts/transforms/correlate.py

def configure(self) -> None:
    """
    Setup alignment based on filter length (N) and user latency property.

    The SGN-TS temporal invariant is: T_out = T_in + (N-1) - L
    Where L is the compensation (shift).
    To achieve T_out = T_in + self.latency, we set L = (N-1) - self.latency.

    When taps_reversed=True, pre-reversed taps + scipy.correlate = convolution.
    Convolution reverses the filter's time axis, so a causal kernel (energy at
    index 0) appears anti-causal (energy at index N-1) to the correlator. We
    compensate by mapping: effective_latency = (N-1) - user_latency.
    This way latency=0 means "zero latency" regardless of tap ordering.
    """
    if self.filters is not None:
        self.shape = self.filters.shape
        overlap_samples = max(0, self.shape[-1] - 1)
    else:
        # Null Initial Condition:
        self.shape = (1, 1)
        overlap_samples = 0

    # When taps are pre-reversed (correlate = convolution), the correlator
    # sees the energy at the opposite end of the kernel. Remap latency so
    # that the user's intent (0 = causal) is preserved.
    effective_latency = self.latency
    if self.taps_reversed and overlap_samples > 0:
        effective_latency = overlap_samples - self.latency

    # Alignment logic:
    # 'overlap' ensures we have (N-1) samples of history for 'valid' correlation.
    # 'L' (compensation) = (N-1) - effective_latency
    comp_samples = overlap_samples - effective_latency

    self.adapter_config.alignment(
        overlap=(Offset.fromsamples(overlap_samples, self.sample_rate), 0),
        shift=-Offset.fromsamples(comp_samples, self.sample_rate),
    )
    # Pad with zeros on startup only if requested; otherwise the adapter
    # stalls until it has real history.
    self.adapter_config.on_startup(pad_zeros=self.pad_zeros)

    self.sink_pad = self.sink_pads[0]
    self.source_pad = self.source_pads[0]

corr(data)

Perform the mathematical correlation.

Uses scipy.signal.correlate(mode='valid') which effectively reverses the filter taps. self.method selects scipy's convolution backend ("auto"/"direct"/"fft").

Source code in src/sgnts/transforms/correlate.py

def corr(self, data: Array) -> Array:
    """
    Perform the mathematical correlation.

    Uses scipy.signal.correlate(mode='valid') which effectively
    reverses the filter taps. ``self.method`` selects scipy's
    convolution backend ("auto"/"direct"/"fft").
    """
    if self.filters is None:
        raise ValueError("Cannot correlate without filters")

    if len(self.filters.shape) == 1:
        return scipy.signal.correlate(
            data, self.filters, mode="valid", method=self.method
        )

    # Row count and output shape come from the filters in hand so that
    # >= 3-dim banks apply every filter row (not just shape[0] of the
    # flattened rows). The reshape is local: repeated calls (and the
    # documented read-only ``self.filters`` contract) keep the original
    # bank.
    shape = self.filters.shape
    if shape[-1] != self.shape[-1]:
        # Defensive: the element's output span and adapter overlap were
        # configured for self.shape; correlating with differently-sized
        # filters would silently produce wrong results.
        raise ValueError(
            f"filters of shape {shape} are inconsistent with the "
            f"configured filter shape {self.shape}; the element's "
            "output span and adapter overlap were built for the "
            "configured shape"
        )
    f_reshaped = numpy.asarray(self.filters).reshape(-1, shape[-1])
    os = []
    for j in range(f_reshaped.shape[0]):
        # Multi-channel data: map filter row j to data channel j. A single
        # (1-D) data stream is correlated against every filter row
        # (filter-bank semantics).
        d = data[j] if len(data.shape) > 1 else data
        os.append(
            scipy.signal.correlate(
                d, f_reshaped[j], mode="valid", method=self.method
            )
        )
    return numpy.vstack(os).reshape(shape[:-1] + (-1,))

internal()

Standard SGN-TS internal loop: pulls aligned input and executes transform.

Source code in src/sgnts/transforms/correlate.py

def internal(self) -> None:
    """
    Standard SGN-TS internal loop: pulls aligned input and executes transform.
    """
    super().internal()
    _, output_collector = self.next_output()
    _, input_frame = self.next_input()
    self._transform(input_frame, output_collector)

    # Ensure frame is fully populated to its expected duration
    if not output_collector._buffers:
        # Entirely empty: generate a single gap buffer spanning the stride
        out_samples = Offset.tosamples(output_collector.noffset, self.sample_rate)
        out_buf = SeriesBuffer(
            offset=output_collector.offset,
            sample_rate=self.sample_rate,
            data=None,
            shape=self.shape[:-1] + (out_samples,),
        )
        out_buf = out_buf.copy(is_gap=True)
        output_collector.append(out_buf)
    else:
        # Check if last buffer reaches the end of the frame
        frame_end = output_collector.offset + output_collector.noffset
        last_end = output_collector._buffers[-1].end_offset
        # Defensive: the aligned adapter fills the frame exactly, so the
        # last produced buffer already reaches frame_end. Kept as a safety
        # net for partial fills but unreachable via the normal pipeline.
        if last_end < frame_end:  # pragma: no cover
            gap_noffset = frame_end - last_end
            out_samples = Offset.tosamples(gap_noffset, self.sample_rate)
            out_buf = SeriesBuffer(
                offset=last_end,
                sample_rate=self.sample_rate,
                data=None,
                shape=self.shape[:-1] + (out_samples,),
            )
            out_buf = out_buf.copy(is_gap=True)
            output_collector.append(out_buf)

    output_collector.close()

CosSquaredTransition dataclass

Bases: TransitionProfile

              flowchart TD
              sgnts.transforms.correlate.CosSquaredTransition[CosSquaredTransition]
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

                              sgnts.transforms.correlate.TransitionProfile --> sgnts.transforms.correlate.CosSquaredTransition
                


              click sgnts.transforms.correlate.CosSquaredTransition href "" "sgnts.transforms.correlate.CosSquaredTransition"
              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Standard Cosine-Squared crossfade.

Provides C1 continuity at the boundaries and constant power summation for uncorrelated noise. This is the preferred profile for smooth, adiabatic filter updates.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class CosSquaredTransition(TransitionProfile):
    """
    Standard Cosine-Squared crossfade.

    Provides C1 continuity at the boundaries and constant power
    summation for uncorrelated noise. This is the preferred profile
    for smooth, adiabatic filter updates.
    """

    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        # Using scipy for high-precision window generation.
        # We use a 2*n cosine window and take the first half.
        win_new = (scipy.signal.windows.cosine(2 * n, sym=True) ** 2)[:n]
        return 1.0 - win_new, win_new

DiscreteTransition dataclass

Bases: TransitionProfile

              flowchart TD
              sgnts.transforms.correlate.DiscreteTransition[DiscreteTransition]
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

                              sgnts.transforms.correlate.TransitionProfile --> sgnts.transforms.correlate.DiscreteTransition
                


              click sgnts.transforms.correlate.DiscreteTransition href "" "sgnts.transforms.correlate.DiscreteTransition"
              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Hard switch at the boundary.

This profile ensures that the output is piecewise stationary, matching the target filter exactly at the transition offset without any crossfading.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class DiscreteTransition(TransitionProfile):
    """
    Hard switch at the boundary.

    This profile ensures that the output is piecewise stationary,
    matching the target filter exactly at the transition offset
    without any crossfading.
    """

    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        # Hard jump: old filter has zero weight, new filter has unity weight.
        return backend.zeros((n,)), backend.ones((n,))

    @property
    def skip_old(self) -> bool:
        # Optimization: no need to correlate with the old filter.
        return True

FilterState dataclass

Internal immutable record of a filter's temporal validity.

Attributes:

Name	Type	Description
offset	int	The start time (in framework ticks) of the filter.
noffset	int	The duration (in framework ticks) of pre-calculated validity.
data	Array	The filter coefficients (taps).

Source code in src/sgnts/transforms/correlate.py

@dataclass(frozen=True)
class FilterState:
    """
    Internal immutable record of a filter's temporal validity.

    Attributes:
        offset: The start time (in framework ticks) of the filter.
        noffset: The duration (in framework ticks) of pre-calculated validity.
        data: The filter coefficients (taps).
    """

    offset: int
    noffset: int
    data: Array

LinearTransition dataclass

Bases: TransitionProfile

              flowchart TD
              sgnts.transforms.correlate.LinearTransition[LinearTransition]
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

                              sgnts.transforms.correlate.TransitionProfile --> sgnts.transforms.correlate.LinearTransition
                


              click sgnts.transforms.correlate.LinearTransition href "" "sgnts.transforms.correlate.LinearTransition"
              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Linear constant-voltage crossfade.

Simple arithmetic blending of filter outputs. Useful for low-complexity scenarios or debugging.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class LinearTransition(TransitionProfile):
    """
    Linear constant-voltage crossfade.

    Simple arithmetic blending of filter outputs. Useful for low-complexity
    scenarios or debugging.
    """

    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        t = backend.arange(n) / (n - 1) if n > 1 else backend.zeros((n,))
        return 1.0 - t, t

PlanckTaperTransition dataclass

Bases: TransitionProfile

              flowchart TD
              sgnts.transforms.correlate.PlanckTaperTransition[PlanckTaperTransition]
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

                              sgnts.transforms.correlate.TransitionProfile --> sgnts.transforms.correlate.PlanckTaperTransition
                


              click sgnts.transforms.correlate.PlanckTaperTransition href "" "sgnts.transforms.correlate.PlanckTaperTransition"
              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Planck-taper crossfade.

Uses the Planck-taper window, a C-infinity (all derivatives continuous) smooth step. Because the transition and every one of its derivatives vanish at both boundaries, it produces the gentlest spectral leakage of the available profiles during a filter handover.

See McKechan, Robinson & Sathyaprakash (2010), arXiv:1003.2939.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class PlanckTaperTransition(TransitionProfile):
    """
    Planck-taper crossfade.

    Uses the Planck-taper window, a C-infinity (all derivatives continuous)
    smooth step. Because the transition and every one of its derivatives
    vanish at both boundaries, it produces the gentlest spectral leakage of
    the available profiles during a filter handover.

    See McKechan, Robinson & Sathyaprakash (2010), arXiv:1003.2939.
    """

    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        # Sample the open interval (0, 1) at bin midpoints so the endpoint
        # singularities of the Planck-taper exponent are never evaluated.
        x = (backend.arange(n) + 0.5) / n
        # expit(-z) == 1 / (1 + exp(z)) is the numerically stable logistic
        # form of the Planck-taper rising edge (0 -> 1 across the window).
        z = 1.0 / x - 1.0 / (1.0 - x)
        win_new = scipy.special.expit(-z)
        return 1.0 - win_new, win_new

ReverseDiscreteTransition dataclass

Bases: TransitionProfile

              flowchart TD
              sgnts.transforms.correlate.ReverseDiscreteTransition[ReverseDiscreteTransition]
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

                              sgnts.transforms.correlate.TransitionProfile --> sgnts.transforms.correlate.ReverseDiscreteTransition
                


              click sgnts.transforms.correlate.ReverseDiscreteTransition href "" "sgnts.transforms.correlate.ReverseDiscreteTransition"
              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Identity switch (keeps old filter).

Primarily used for testing skip_new optimization paths.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class ReverseDiscreteTransition(TransitionProfile):
    """
    Identity switch (keeps old filter).

    Primarily used for testing skip_new optimization paths.
    """

    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        return backend.ones((n,)), backend.zeros((n,))

    @property
    def skip_new(self) -> bool:
        return True

TransitionProfile dataclass

Bases: ABC

              flowchart TD
              sgnts.transforms.correlate.TransitionProfile[TransitionProfile]

              

              click sgnts.transforms.correlate.TransitionProfile href "" "sgnts.transforms.correlate.TransitionProfile"
            

Base class for filter transition strategies.

This follows the Strategy Pattern to allow different types of blending (Cos2, Linear, Discrete, Planck-taper) between old and new filters when an update occurs mid-stride.

Source code in src/sgnts/transforms/correlate.py

@dataclass
class TransitionProfile(ABC):
    """
    Base class for filter transition strategies.

    This follows the Strategy Pattern to allow different types of
    blending (Cos2, Linear, Discrete, Planck-taper) between old and new
    filters when an update occurs mid-stride.
    """

    @abstractmethod
    def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
        """
        Returns (old_weights, new_weights) for a stride of length n.

        Args:
            n: Number of samples in the transition window.
            backend: The array backend (NumPy/Torch) to use.
        """
        pass

    @property
    def skip_old(self) -> bool:
        """Optimization: if True, the old filter correlation can be skipped."""
        return False

    @property
    def skip_new(self) -> bool:
        """Optimization: if True, the new filter correlation can be skipped."""
        return False

skip_new property

Optimization: if True, the new filter correlation can be skipped.

skip_old property

Optimization: if True, the old filter correlation can be skipped.

get_weights(n, backend) abstractmethod

Returns (old_weights, new_weights) for a stride of length n.

Parameters:

Name	Type	Description	Default
n	int	Number of samples in the transition window.	required
backend	Any	The array backend (NumPy/Torch) to use.	required

Source code in src/sgnts/transforms/correlate.py

@abstractmethod
def get_weights(self, n: int, backend: Any) -> Tuple[Any, Any]:
    """
    Returns (old_weights, new_weights) for a stride of length n.

    Args:
        n: Number of samples in the transition window.
        backend: The array backend (NumPy/Torch) to use.
    """
    pass