comnum_spec/blocks/
cpm.rs

1//! Continuous-phase modulation: the modulator and its detectors.
2//!
3//! `cpm_tx` replaces the symbols of a real constellation (PAM of order $M$, or BPSK)
4//! by the constant-envelope signal $e^{j \phi(t)}$ of [`comnum_waveform::cpm`]: the
5//! levels of the constellation, in increasing order, are the CPM symbols
6//! $\pm 1, \pm 3, \dots$. `cpm_rx` decides the symbols back, by the maximum-likelihood
7//! sequence detector or the one-symbol differential detector, and hands the metrics
8//! the decided constellation points.
9
10use super::schema::ParamType;
11use super::{Block, Extension, Frame};
12use comnum_core::rng::Rng;
13use comnum_core::{Error, Result, ensure};
14use comnum_mapping::constellation::Constellation;
15use comnum_waveform::cpm::{Cpm, CpmConfig, FrequencyPulse, ModulationIndex, msk_bits, msk_precode};
16use serde::Deserialize;
17use serde_json::Value as Json;
18
19/// The blocks of this module: their documentation and their builders.
20pub(super) fn blocks() -> Vec<Extension> {
21    vec![
22        builtin!(
23            "cpm_tx",
24            "Continuous-phase modulator of the symbols of a real constellation (CPFSK, MSK, GMSK, raised-cosine CPM): a unit-envelope signal at sps samples per symbol.",
25            CpmSpec,
26            needs [Samples, Constellation],
27            gives [],
28            |q, _, _| Ok(Box::new(CpmTx(q))),
29        ),
30        builtin!(
31            "cpm_rx",
32            "CPM detector (maximum-likelihood sequence detection over the phase trellis, or one-symbol differential detection) back to the symbols.",
33            CpmRxSpec,
34            needs [Samples, Constellation],
35            gives [],
36            |q, _, _| Ok(Box::new(CpmRx::new(q))),
37        ),
38    ]
39}
40
41block_params! {
42    struct CpmSpec {
43        h: String ["1/2"], "modulation index p/q, as a fraction (1/2 with a rectangular pulse of length 1 is MSK)";
44        pulse: String ["rec"] one_of ["rec", "rc", "gmsk"],
45            "frequency pulse: rectangular (1REC is CPFSK), raised cosine, or Gaussian (GMSK, with bt)";
46        length: usize [1] >= 1, "length L of the pulse in symbols";
47        bt: f64 [0.3] > 0, "bandwidth–time product of the Gaussian pulse";
48        sps: usize [4] >= 2, "samples per symbol";
49        precode: bool [false],
50            "MSK precoding (binary, h 1/2, rec, length 1): the bits are the boundary signs of the equivalent OQPSK signal, and cpm_rx reads them from the phase states of the ML path, with the BER of BPSK";
51    }
52}
53
54impl CpmSpec {
55    /// The modulator of order `order`.
56    fn cpm(&self, order: usize) -> Result<Cpm> {
57        let h: ModulationIndex = self.h.parse().map_err(|e: Error| e.context("cpm"))?;
58        CpmConfig {
59            order,
60            h_num: h.num,
61            h_den: h.den,
62            pulse: FrequencyPulse::from_name(&self.pulse, self.bt)?,
63            length: self.length,
64            sps: self.sps,
65        }
66        .build()
67    }
68
69    /// The modulator, checked for MSK when `precode` is set.
70    fn checked(&self, order: usize, precode: bool, block: &str) -> Result<Cpm> {
71        let cpm = self.cpm(order)?;
72        let c = cpm.config();
73        let msk = order == 2 && c.h_num == 1 && c.h_den == 2 && c.pulse == FrequencyPulse::Rectangular && c.length == 1;
74        ensure(!precode || msk, || format!("{block}: precode needs MSK (BPSK, h: 1/2, pulse: rec, length: 1)"))?;
75        Ok(cpm)
76    }
77}
78
79/// The CPM symbol of every constellation index (the rank of its level), and the
80/// constellation index of every CPM symbol.
81fn levels(c: &Constellation, block: &str) -> Result<(Vec<usize>, Vec<usize>)> {
82    let points = c.points();
83    ensure(points.iter().all(|p| p.im.abs() < 1e-9), || {
84        format!("{block}: the constellation must be real (PAM or BPSK), got {:?}", c.kind())
85    })?;
86    let mut order: Vec<usize> = (0..points.len()).collect();
87    order.sort_by(|&a, &b| points[a].re.total_cmp(&points[b].re));
88    let mut rank = vec![0; points.len()];
89    for (r, &i) in order.iter().enumerate() {
90        rank[i] = r;
91    }
92    Ok((rank, order))
93}
94
95struct CpmTx(CpmSpec);
96
97impl Block for CpmTx {
98    fn apply(&self, s: &mut Frame, _: &mut Rng) -> Result<()> {
99        s.require_one_sps("cpm_tx")?;
100        let c = s.constellation("cpm_tx")?;
101        let (rank, _) = levels(c, "cpm_tx")?;
102        let cpm = self.0.checked(c.order(), self.0.precode, "cpm_tx")?;
103        let mut symbols: Vec<usize> = c.decide(&s.samples).into_iter().map(|i| rank[i]).collect();
104        if self.0.precode {
105            let bits: Vec<u8> = symbols.iter().map(|&r| r as u8).collect();
106            symbols = msk_precode(&bits);
107        }
108        s.samples = cpm.modulate(&symbols)?;
109        s.sps = self.0.sps as f64;
110        Ok(())
111    }
112}
113
114block_params! {
115    struct CpmRxSpec {
116        h: String ["1/2"], "as cpm_tx";
117        pulse: String ["rec"] one_of ["rec", "rc", "gmsk"], "as cpm_tx";
118        length: usize [1] >= 1, "as cpm_tx";
119        bt: f64 [0.3] > 0, "as cpm_tx";
120        sps: usize [4] >= 2, "as cpm_tx";
121        detector: CpmDetector ["ml"] one_of ["ml", "differential"],
122            "ml (Viterbi over the phase trellis, coherent) or differential (sign of the one-symbol phase increment, after a Gaussian low-pass of rx_bt)";
123        rx_bt: Option<f64> [absent] > 0, "bandwidth–time product of the receive low-pass of the differential detector (none: no filter)";
124        precode: bool [false], "as cpm_tx (with the ml detector)";
125    }
126}
127
128/// The detector of `cpm_rx`.
129#[derive(Clone, Copy, PartialEq, Eq, Deserialize)]
130#[serde(rename_all = "lowercase")]
131enum CpmDetector {
132    /// Viterbi over the phase trellis.
133    Ml,
134    /// Sign of the one-symbol phase increment.
135    Differential,
136}
137
138impl ParamType for CpmDetector {
139    fn schema() -> Json {
140        String::schema()
141    }
142}
143
144struct CpmRx {
145    spec: CpmRxSpec,
146    differential: bool,
147}
148
149impl CpmRx {
150    fn new(spec: CpmRxSpec) -> CpmRx {
151        let differential = spec.detector == CpmDetector::Differential;
152        CpmRx { spec, differential }
153    }
154
155    /// The MSK precoding is read from the ML path.
156    fn precode(&self) -> bool {
157        self.spec.precode && !self.differential
158    }
159}
160
161impl Block for CpmRx {
162    fn apply(&self, s: &mut Frame, _: &mut Rng) -> Result<()> {
163        let c = s.constellation("cpm_rx")?;
164        let (_, index_of) = levels(c, "cpm_rx")?;
165        ensure(s.sps == self.spec.sps as f64, || {
166            format!("cpm_rx: expects {} samples per symbol (as cpm_tx), the signal has {}", self.spec.sps, s.sps)
167        })?;
168        let tx = CpmSpec {
169            h: self.spec.h.clone(),
170            pulse: self.spec.pulse.clone(),
171            length: self.spec.length,
172            bt: self.spec.bt,
173            sps: self.spec.sps,
174            precode: self.spec.precode,
175        };
176        let cpm = tx.checked(c.order(), self.precode(), "cpm_rx")?;
177        let decided = if self.differential {
178            cpm.differential_detect(&s.samples, self.spec.rx_bt)?
179        } else if self.precode() {
180            msk_bits(&cpm.ml_detect_path(&s.samples)?.phase_states).into_iter().map(usize::from).collect()
181        } else {
182            cpm.ml_detect(&s.samples)?
183        };
184        s.samples = decided.iter().map(|&r| c.points()[index_of[r]]).collect();
185        s.sps = 1.0;
186        Ok(())
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::super::{BuildContext, build};
193    use comnum_core::rng::Rng;
194    use serde_json::json;
195
196    #[test]
197    fn msk_round_trip_is_exact_without_noise() {
198        let ctx = BuildContext::default();
199        let mut rng = Rng::new(3);
200        let mut s = super::Frame::default();
201        let source = build(&json!({"type": "source", "constellation": "BPSK", "n_symbols": 500}), &ctx).unwrap();
202        source.apply(&mut s, &mut rng).unwrap();
203        let tx_symbols = s.samples.clone();
204        build(&json!({"type": "cpm_tx", "sps": 4}), &ctx).unwrap().apply(&mut s, &mut rng).unwrap();
205        assert_eq!(s.samples.len(), 2000);
206        assert!(s.samples.iter().all(|v| (v.norm() - 1.0).abs() < 1e-12), "constant envelope");
207        build(&json!({"type": "cpm_rx", "sps": 4}), &ctx).unwrap().apply(&mut s, &mut rng).unwrap();
208        assert_eq!(s.samples, tx_symbols);
209    }
210
211    #[test]
212    fn a_complex_constellation_is_refused() {
213        let ctx = BuildContext::default();
214        let mut rng = Rng::new(3);
215        let mut s = super::Frame::default();
216        build(&json!({"type": "source", "constellation": "QPSK", "n_symbols": 100}), &ctx)
217            .unwrap()
218            .apply(&mut s, &mut rng)
219            .unwrap();
220        let e = build(&json!({"type": "cpm_tx"}), &ctx).unwrap().apply(&mut s, &mut rng).unwrap_err();
221        assert!(e.to_string().contains("real"), "{e}");
222        assert!(build(&json!({"type": "cpm_tx", "h": "half"}), &ctx).unwrap().apply(&mut s, &mut rng).is_err());
223    }
224}
comnum_spec/blocks/cpm.rs

comnum_spec/blocks/
cpm.rs
