make_channelize_poly_stream#

Streaming polyphase channelizer: channelize a signal delivered in segments, equivalent to a one-shot channelize_poly over the concatenated stream

Added in version 1.1.0.

struct ChannelizePolyStreamParams#

Construction-time parameters for a streaming polyphase channelizer object.

Aggregate with designated-initializer support, e.g. make_channelize_poly_stream<float>(h, {.num_channels = 8, .decimation_factor = 4}, exec). A params struct is used so future options can be added without changing call sites.

Public Members

index_t num_channels = 1#: Number of channels M (positive; 1 is the degenerate single-channel FIR)

index_t decimation_factor = 1#: Decimation factor D (positive, <= num_channels)

template<typename InType, typename FilterOp, typename Exec = cudaExecutor> auto matx::make_channelize_poly_stream(const FilterOp &filter, const ChannelizePolyStreamParams &params, Exec exec = {})#

Create a streaming polyphase channelizer object.

The object channelizes an arbitrarily long signal, delivered in segments of any (possibly varying) size, into params.num_channels channels decimated by params.decimation_factor. Callers provide input segments via feed() and call flush() once at the end of the stream. The concatenation of the produced [blocks, num_channels] outputs equals a single one-shot channelize_poly(signal, filter, num_channels, decimation_factor) over the whole signal. As with channelize_poly, the output element type must be complex.

The object owns a small retained-history buffer sized from the filter, num_channels, and decimation_factor. No allocation scales with the segment size. Each call to feed() or flush() accepts an output buffer shaped [rows, num_channels] and writes the produced blocks to its leading rows. The output buffer must have at least max_output(max_input_segment_size) rows. If the maximum segment size is known a priori, then a single output buffer can be allocated and reused for all calls.

Each feed() / flush() call writes the blocks to the front of the output buffer and returns the number of blocks written (which may be 0), so no dynamic memory is allocated during the call. Consume the produced region slice(out, {0, 0}, {count, num_channels}) before reusing the output buffer. A zero-length slice is not valid, so create and use the slice only when count > 0.

Template Parameters:

InType – Sample type of the input stream (as in make_tensor<T>)
FilterOp – Type of the filter operator (deduced)
Exec – Executor type (CUDA or host; deduced, cudaExecutor by default)

Parameters:

filter – FIR prototype filter (1D, length >= 1). The object materializes a copy of the filter at construction, so any filter operator (including a transform expression) is evaluated once and need not outlive the object.
params – Stream parameters; see ChannelizePolyStreamParams
exec – Executor bound to this stream’s lifetime. All feed()/flush() work runs on it. For CUDA executors the retain buffer is stream-ordered device memory. For host executors the retain buffer is system-allocated memory. The executor (and, for CUDA, its stream) must outlive the object.

Returns:

A streaming object exposing feed(), flush(), max_output(), reset(), history_len(), and num_channels()

template<typename InType, typename FilterOp, typename Exec> class ChannelizePolyStream#

Streaming polyphase channelizer object; construct via make_channelize_poly_stream().

Channelizes an arbitrarily long signal delivered in segments of any (possibly varying) size into rank-2 [blocks, num_channels] outputs, where each block (row) is one output time step across all channels. The only stream state is a small retained-history buffer; no allocation scales with the segment size. All work runs asynchronously on the executor bound at construction. An object serves one stream at a time and is not thread-safe.

Public Functions

ChannelizePolyStream(const ChannelizePolyStream&) = delete#: Streaming objects are not copyable: a copy would share the retained history buffer while tracking its state independently, corrupting both streams. Move construction/assignment transfers ownership and is allowed.

ChannelizePolyStream &operator=(const ChannelizePolyStream&) = delete#: Streaming objects are not copyable: a copy would share the retained history buffer while tracking its state independently, corrupting both streams. Move construction/assignment transfers ownership and is allowed.

ChannelizePolyStream(ChannelizePolyStream&&) = default#: Move-construct, transferring ownership of the stream state.

ChannelizePolyStream &operator=(ChannelizePolyStream&&) = default#: Move-assign, transferring ownership of the stream state.

inline index_t num_channels() const#

Number of channels the stream produces (the output column count).

Returns:: Number of channels

inline index_t history_len() const#

Maximum number of trailing input samples retained between calls.

Informational; the caller does not size anything with it. feed() concatenates the retained history internally.

Returns:: Maximum retained history length in samples

inline index_t max_output(index_t new_len) const#

Upper bound on the output blocks a single feed() or flush() call can produce.

Size one reusable [max_output(largest_input_segment_size), num_channels] output buffer; it is then large enough for every feed() of up to that many samples and for flush().

Parameters:: new_len – Largest segment size that will be passed to feed()
Returns:: Maximum output blocks (rows) a single call may produce

inline void reset()#

Restart the stream for a new input signal.

Drops all retained history and clears the end-of-stream state set by flush().

template<typename InOp, typename OutTensor> inline index_t feed(const InOp &new_samples, OutTensor &out)#

Feed a segment of new samples and receive the number of output blocks it produces.

Emits every not-yet-emitted block whose decimation_factor input samples have fully arrived. The per-call count varies with the segment size (it can be zero). Blocks are written to the leading rows of out; the return value is the number of blocks written. Consume slice(out, {0, 0}, {count, M}) (the produced rows) before reusing out. Runs asynchronously on the object’s executor. Throws matxInvalidParameter if called after flush(). Use reset() to start a new stream.

Template Parameters:

InOp – 1D input operator type (deduced)
OutTensor – 2D output tensor type (deduced)

Parameters:

new_samples – New signal samples (1D, non-empty). Any MatX operator is accepted and its lifecycle is run each call. A transform-valued segment (e.g. ifft(…)) therefore works but allocates and evaluates a per-call temporary; for hot streaming loops prefer a directly-evaluable segment (a tensor, view, or generator) or materialize once and reuse.
out – Output buffer shaped [>= max_output(input_segment_size), num_channels]; throws matxInvalidSize on a channel-count mismatch or if the row count is smaller than the produced block count

Returns:

Number of blocks (rows) written to the front of out (may be 0). A zero-length slice is not valid, so create and use slice(out, {0, 0}, {count, num_channels}) only when count > 0.

template<typename OutTensor> inline index_t flush(OutTensor &out)#

Emit the end-of-stream trailing block, if any.

When the total stream length is not a multiple of decimation_factor, the final block is only partially covered by real samples. flush() emits that final block (computed with implicit zero padding on the right), matching the final block of the one-shot channelize_poly. When the total length is a multiple of decimation_factor, every block was fully covered and already emitted, and flush() produces zero rows. The first call emits the trailing block and ends the stream. Subsequent calls return 0, and feed() throws until reset() starts a new stream. A flush() that throws (for example, an undersized or mis-shaped output buffer) does not end the stream and can be retried. Runs asynchronously on the object’s executor.

Template Parameters:: OutTensor – 2D output tensor type (deduced)
Parameters:: out – Output buffer shaped [>= max_output(input_segment_size), num_channels]; throws matxInvalidSize on a channel-count mismatch or if the row count is smaller than the produced block count
Returns:: Number of blocks (rows) written to the front of out (may be 0). A zero-length slice is not valid, so create and use slice(out, {0, 0}, {count, num_channels}) only when count > 0.

Examples#

auto channelize_stream = make_channelize_poly_stream<T>(h, {.num_channels = M, .decimation_factor = D}, exec);
auto output_frame = make_tensor<CT>({channelize_stream.max_output(a.chunk), M});

index_t off = 0, nchunks = 0;
for (index_t g = 0; g < N; g += a.chunk) {
  const index_t in_length = std::min(a.chunk, N - g);
  auto in_chunk = slice(sig, {g}, {g + in_length});
  const index_t cnt = channelize_stream.feed(in_chunk, output_frame); // blocks emitted

  // For real applications, consume slice(output_frame, {0, 0}, {cnt, M}) here

  // For validation only, copy the frame into the full-length buffer.
  if (cnt > 0) {
    (slice(full, {off, 0}, {off + cnt, M}) = slice(output_frame, {0, 0}, {cnt, M})).run(exec);
  }
  off += cnt;
  ++nchunks;
}
// End of stream: flush the trailing (edge-padded) block, if any.
const index_t tcnt = channelize_stream.flush(output_frame);
if (tcnt > 0) {
  (slice(full, {off, 0}, {off + tcnt, M}) = slice(output_frame, {0, 0}, {tcnt, M})).run(exec);
}
off += tcnt;

make_channelize_poly_stream

Contents

make_channelize_poly_stream#

Examples#