orion-sdr 0.0.69

// Copyright (c) 2026 G & R Associates LLC
// SPDX-License-Identifier: MIT OR Apache-2.0

// src/demodulate/dvb_t_stream.rs
//
// Streaming DVB-T frame receiver: a `feed`/`flush` front end over the batch
// `DvbTFrameDemod`, mirroring `OfdmFrameStreamDemod`. A DVB-T signal is a
// continuous run of fixed-size frames with no preamble, so the receiver
// accumulates IQ, guard-interval-acquires the next frame at the front of the
// buffer, decodes it, drains its samples, and loops — holding a partially-arrived
// frame until a later `feed` completes it.
//
// The frame geometry (symbols per frame, payload length per frame) is fixed by
// the link and supplied at construction, exactly as the batch demod takes
// `n_symbols`/`payload_len` — a cold DVB-T receiver knows the mode from
// configuration (or an out-of-band TPS lock) and confirms it via the TPS word.

use super::dvb_t_frame::{DvbTFrameDemod, DvbTRxError, DvbTRxFrame};
use super::dvb_t_probe::DvbTRxProbe;
use crate::sync::dvb_t_gi_sync;
use crate::waveform::dvb_t::{DVB_T_N_FFT, DvbTFrameParams};
use num_complex::Complex32 as C32;

/// A streaming DVB-T receiver. Push IQ with [`feed`](Self::feed); it
/// guard-interval-acquires and decodes each fixed-size frame as its samples
/// arrive, returning the completed ones. [`flush`](Self::flush) runs a final pass
/// over the residual buffer.
pub struct DvbTFrameStreamDemod {
    /// The per-frame demodulator (carries the params + the integer-CFO flag).
    demod: DvbTFrameDemod,
    /// OFDM symbols per frame (from the paired modulator's `DvbTFrame::n_symbols`).
    n_symbols: usize,
    /// Payload byte count per frame (for trimming), from the modulator.
    payload_len: usize,
    /// Samples per OFDM symbol (`n_fft + cp_len`).
    sps: usize,
    /// Accumulated, not-yet-consumed IQ.
    buf: Vec<C32>,
}

impl DvbTFrameStreamDemod {
    /// Builds a streaming receiver for a link whose frames are `n_symbols` OFDM
    /// symbols carrying `payload_len` payload bytes each, under `params`.
    pub fn new(params: DvbTFrameParams, n_symbols: usize, payload_len: usize) -> Self {
        let cp_len = params.config().carrier_plan.cp_len();
        Self {
            demod: DvbTFrameDemod::new(params),
            n_symbols,
            payload_len,
            sps: DVB_T_N_FFT + cp_len,
            buf: Vec::new(),
        }
    }

    /// Enables (or disables) internal integer-CFO correction on the underlying
    /// frame demod — a link-constant knob set once at construction (see
    /// [`DvbTFrameDemod::with_integer_cfo_correction`]). Each decoded frame then has
    /// its whole-subcarrier offset removed internally.
    pub fn with_integer_cfo_correction(mut self, on: bool) -> Self {
        self.demod = self.demod.with_integer_cfo_correction(on);
        self
    }

    /// Slides the FFT window `backoff` samples earlier into the guard on the
    /// underlying frame demod — the receiver half of the TX spectral-shaping
    /// pair (see [`DvbTFrameDemod::with_rx_window_backoff`]). Without it a
    /// stream carrying a symbol taper or a baseband mask is received with the
    /// shaping inside the FFT window rather than outside it. Capped at
    /// [`DVB_T_MAX_RX_WINDOW_BACKOFF`](crate::waveform::dvb_t::DVB_T_MAX_RX_WINDOW_BACKOFF)
    /// by the scattered-pilot grid, not by the guard.
    ///
    /// Guard-interval acquisition is unaffected: the back-off moves only the
    /// per-symbol FFT window, not the frame's sample boundaries, so the
    /// front-of-buffer GI search and the per-frame drain are unchanged.
    pub fn with_rx_window_backoff(mut self, backoff: usize) -> Self {
        self.demod = self.demod.with_rx_window_backoff(backoff);
        self
    }

    /// Enables (or disables) the BER rungs on each decoded frame's diagnostics
    /// (see [`DvbTFrameDemod::with_error_rates`]). Off by default — it costs one
    /// encode chain per frame.
    pub fn with_error_rates(mut self, on: bool) -> Self {
        self.demod = self.demod.with_error_rates(on);
        self
    }

    /// Accumulated (not-yet-consumed) sample count.
    pub fn len(&self) -> usize {
        self.buf.len()
    }

    pub fn is_empty(&self) -> bool {
        self.buf.is_empty()
    }

    /// Read-only view of the accumulated IQ buffer.
    pub fn view_buf(&self) -> &[C32] {
        &self.buf
    }

    /// Discards all accumulated samples.
    pub fn clear(&mut self) {
        self.buf.clear();
    }

    /// Samples one frame occupies once acquired (`n_symbols · sps`).
    fn frame_samples(&self) -> usize {
        self.n_symbols * self.sps
    }

    /// Feeds IQ samples and returns any frames (or errors) that completed.
    pub fn feed(&mut self, iq: &[C32]) -> Vec<Result<DvbTRxFrame, DvbTRxError>> {
        self.buf.extend_from_slice(iq);
        self.drain(None)
    }

    /// Runs a final decode pass over the residual buffer (end of stream). Same
    /// semantics as [`feed`](Self::feed) with no new samples.
    pub fn flush(&mut self) -> Vec<Result<DvbTRxFrame, DvbTRxError>> {
        self.drain(None)
    }

    /// [`feed`](Self::feed), additionally filling `probe` with each frame's
    /// equalized data-carrier symbols and per-coded-bit correction map — the two
    /// quantities a constellation / decoder display needs. See [`DvbTRxProbe`].
    ///
    /// `probe` is **cleared first**, then refilled with everything this call
    /// produced; its allocations are retained, so probing a steady stream does
    /// not reallocate.
    ///
    /// **The gate is the choice of method, not a flag.** There is no
    /// `want_probe` field, so the unprobed [`feed`](Self::feed) gains no runtime
    /// branch and no receiver state can disagree with what the caller believes
    /// it enabled. A viewer that toggles its pane calls the other method.
    ///
    /// Costs, per frame: one encode chain (the same one
    /// [`with_error_rates`](Self::with_error_rates) runs — asking for both pays
    /// for it once), one further inner encode to re-derive what the decoder
    /// decided, and two buffer fills totalling ~1 MB for a 68-symbol frame.
    /// Frames that fail their Reed–Solomon check still contribute their symbols,
    /// with an empty correction map.
    ///
    /// This does **not** enable the EVM or BER rungs of
    /// [`DvbTRxDiagnostics`](crate::demodulate::DvbTRxDiagnostics); those remain
    /// gated by [`with_error_rates`](Self::with_error_rates), which composes
    /// with this freely.
    pub fn feed_probed(
        &mut self,
        iq: &[C32],
        probe: &mut DvbTRxProbe,
    ) -> Vec<Result<DvbTRxFrame, DvbTRxError>> {
        probe.clear();
        self.buf.extend_from_slice(iq);
        self.drain(Some(probe))
    }

    /// [`flush`](Self::flush) with the probe of [`feed_probed`](Self::feed_probed).
    pub fn flush_probed(
        &mut self,
        probe: &mut DvbTRxProbe,
    ) -> Vec<Result<DvbTRxFrame, DvbTRxError>> {
        probe.clear();
        self.drain(Some(probe))
    }

    /// Repeatedly acquires and decodes frames from the front of the buffer,
    /// consuming their samples, until no further complete frame is present.
    fn drain(
        &mut self,
        mut probe: Option<&mut DvbTRxProbe>,
    ) -> Vec<Result<DvbTRxFrame, DvbTRxError>> {
        let mut out = Vec::new();
        while let FrameStep::Decoded(result, consume_to) = self.try_one_frame(probe.as_deref_mut())
        {
            self.buf.drain(..consume_to);
            out.push(result);
        }
        out
    }

    /// Attempts to acquire and decode one frame at the front of the buffer.
    fn try_one_frame(&mut self, probe: Option<&mut DvbTRxProbe>) -> FrameStep {
        let n_fft = DVB_T_N_FFT;
        let cp_len = self.sps - n_fft;
        let fs = self.demod.params().config().fs;

        // The GI search spans one symbol period; the frame's data then needs
        // `n_symbols · sps` more samples past the acquired start. Wait until the
        // worst case (start near the end of the search window) can be present.
        let need = self.sps + self.frame_samples();
        if self.buf.len() < need {
            return FrameStep::NeedMore;
        }

        // Acquire the symbol boundary at the front (search one symbol period).
        let Some(acq) = dvb_t_gi_sync(&self.buf, n_fft, cp_len, fs, self.sps) else {
            return FrameStep::NeedMore;
        };
        let start = acq.start_sample;
        let consume_to = start + self.frame_samples();
        if consume_to > self.buf.len() {
            return FrameStep::NeedMore;
        }

        // Decode the frame from its acquired start (the batch demod re-locks the
        // CP at offset 0 of this slice, which is idempotent; with integer-CFO
        // correction enabled it also removes any whole-subcarrier offset there).
        let decoded = match probe {
            Some(p) => {
                self.demod
                    .decode_probed(&self.buf[start..], self.n_symbols, self.payload_len, p)
            }
            None => self
                .demod
                .decode(&self.buf[start..], self.n_symbols, self.payload_len),
        };
        match decoded {
            Ok(frame) => FrameStep::Decoded(Ok(frame), consume_to),
            // A genuine decode failure on a fully-present frame: report it and
            // consume past this frame so the stream advances (no re-locking the
            // same corrupt occurrence forever).
            Err(e) => FrameStep::Decoded(Err(e), consume_to),
        }
    }
}

/// The outcome of one attempt to decode a frame from the front of the buffer.
enum FrameStep {
    /// A frame (or a decode error on a complete frame) plus how many samples to
    /// drain from the front.
    Decoded(Result<DvbTRxFrame, DvbTRxError>, usize),
    /// Not enough samples buffered yet — hold and retry on the next `feed`.
    NeedMore,
}