diff --git a/README.md b/README.md index 8fecabf..1e7d56a 100644 --- a/README.md +++ b/README.md @@ -577,6 +577,30 @@ reliably: The decoder runs its own CW detector over the captured IQ, so Morse is found even when the recording itself was made in FM or SSB. +### Single sideband + +SSB needs no `--mode usb`. Nothing else does either, but SSB is the mode where +it would matter: FM and AM detectors do not care where in their passband a +signal sits, while an SSB demodulator is a filter that opens at the suppressed +carrier. Tune to the middle of the voice — which is where a detector naturally +lands, a couple of kilohertz up — and its lower half is filtered off while the +rest comes out shifted down by the error. That is the mistuned sound, and it +makes the recording useless rather than merely imperfect. + +So the carrier is measured rather than assumed. Speech puts most of its power +in the first few hundred hertz above the carrier, so an SSB signal's occupied +band is lopsided: the loud end is the carrier end. That locates the carrier to +within about a hundred hertz and names the sideband at the same time — energy +bunched at the low edge is upper sideband, at the high edge lower. + +The frequency in the filename is therefore the carrier, the one you would dial +into a radio, not the middle of the voice. + +Where the signal has no lean to read — a data mode inside an SSB segment, or a +steady tone — the band plan decides, and it is right where the folklore is +wrong: 60 m and the HF utility bands are upper sideband well below the 10 MHz +that "LSB below, USB above" splits on. + ## Output Everything lands in one directory, named diff --git a/bandsaunter/classify.py b/bandsaunter/classify.py index abfa337..f153060 100755 --- a/bandsaunter/classify.py +++ b/bandsaunter/classify.py @@ -21,7 +21,7 @@ from .dsp import (db, instantaneous_frequency, occupied_bandwidth, spectral_flatness, welch_psd) __all__ = ["classify", "Classification", "SignalFeatures", "extract_features", - "CTCSS_TONES", "detect_ctcss"] + "CTCSS_TONES", "detect_ctcss", "SSBAlignment", "ssb_alignment"] def _pow2_floor(n: int, cap: int = 1 << 16) -> int: @@ -978,3 +978,94 @@ def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0, label=label, family=family, confidence=round(min(0.99, score), 3), reasons=reasons, alternatives=alts, suggested_mode=mode, features=f, ) + + +# -------------------------------------------------------------------------- +# SSB alignment +# -------------------------------------------------------------------------- + +# How far below the first audio energy the suppressed carrier sits. Transmit +# filters start passing somewhere around 200-300 Hz, and our own SSB filter +# opens at 200 Hz, so guessing a little low costs nothing while guessing high +# clips the bottom of the voice. +CARRIER_GUARD_HZ = 250.0 + + +@dataclass +class SSBAlignment: + """Where an SSB signal's suppressed carrier is, and which sideband it is.""" + + lower_hz: float # low edge of the occupied band, about the centre + upper_hz: float # high edge + sideband: str # "usb" or "lsb", from which way the band leans + confidence: float # 0 = perfectly symmetric, 1 = all to one side + width_hz: float + + def carrier_for(self, sideband: str) -> float: + """Carrier offset in Hz, given a sideband -- measured or assumed.""" + if sideband == "lsb": + return self.upper_hz + CARRIER_GUARD_HZ + return self.lower_hz - CARRIER_GUARD_HZ + + +def ssb_alignment(x: np.ndarray, sample_rate: float, + hint_bw: float = 0.0) -> SSBAlignment | None: + """Find the suppressed carrier of an SSB signal centred near DC. + + Every other mode tolerates a kilohertz of tuning error -- an FM + discriminator and an AM envelope detector do not care where in their + passband the signal sits. SSB does not: its demodulator is a filter that + opens at the carrier, so tuning to the middle of the voice throws away + everything below that point and shifts what is left down by the error. + + Speech puts most of its power in the first few hundred hertz above the + carrier, so an SSB signal's occupied band is lopsided, and which way it + leans identifies the sideband without recourse to any convention: energy + bunched at the low edge is upper sideband, at the high edge lower. + Returns None when the band is too odd to read, leaving the caller to fall + back on what the band plan says. + """ + nfft = min(32768, _pow2_floor(int(sample_rate / 100.0))) + if x.size < nfft or nfft < 256: + return None + _, psd = welch_psd(x, nfft) + psd_db = db(psd) + freqs = np.fft.fftshift(np.fft.fftfreq(nfft, 1.0 / sample_rate)) + win = np.abs(freqs) <= max(hint_bw * 2.5, 8_000.0) + if not np.any(win): + return None + floor = float(np.percentile(psd_db[win], 25.0)) + peak = float(psd_db[win].max()) + hot = win & (psd_db > max(floor + 8.0, peak - 20.0)) + idx = np.flatnonzero(hot) + if idx.size < 8: + return None + + f_hot = freqs[idx] + # Percentiles rather than the extremes: one stray bin in the skirt would + # otherwise put the carrier a kilohertz out, and the carrier is the whole + # point of the measurement. + lo = float(np.percentile(f_hot, 2.0)) + hi = float(np.percentile(f_hot, 98.0)) + width = hi - lo + if not (400.0 <= width <= 8_000.0): + return None + + # Where the power actually sits inside that band, by weight rather than by + # count, so a wide quiet skirt cannot outvote the loud part of the voice. + w = np.maximum(psd[idx], 0.0) + total = float(w.sum()) + if total <= 0: + return None + order = np.argsort(f_hot) + f_sorted, w_sorted = f_hot[order], w[order] + cumulative = np.cumsum(w_sorted) / total + median = float(f_sorted[int(np.searchsorted(cumulative, 0.5))]) + median = min(max(median, lo), hi) + + to_lo, to_hi = median - lo, hi - median + upper = to_lo < to_hi + confidence = abs(to_hi - to_lo) / width if width > 0 else 0.0 + return SSBAlignment(lower_hz=lo, upper_hz=hi, + sideband="usb" if upper else "lsb", + confidence=float(confidence), width_hz=float(width)) diff --git a/bandsaunter/demod.py b/bandsaunter/demod.py index 5a992c7..e9e34ab 100755 --- a/bandsaunter/demod.py +++ b/bandsaunter/demod.py @@ -224,10 +224,19 @@ class SSBDemod(Demodulator): """Single-sideband via a one-sided complex band-pass, then take the real part.""" upper: bool = True + # Where the suppressed carrier sits relative to the tuned centre. The + # filter below opens at the carrier and everything under it is discarded, + # so this is the one demodulator that has to be told exactly where the + # signal begins; the scanner measures it and passes it in. Shifting here, + # at the IF rate, costs a fraction of what shifting the full-rate stream + # would, and the IF passband is far wider than any plausible offset. + carrier_offset_hz: float = 0.0 def _setup(self) -> None: self._zi = None self._taps = None + self._carrier = (Mixer(self.carrier_offset_hz, self.if_rate) + if self.carrier_offset_hz else None) def _sideband_taps(self): if self._taps is None: @@ -242,6 +251,8 @@ class SSBDemod(Demodulator): return self._taps def _demod(self, x: np.ndarray) -> np.ndarray: + if self._carrier is not None: + x = self._carrier(x) taps = self._sideband_taps() y, self._zi = sps.lfilter(taps, [1.0], x, zi=self._zi) return 2.0 * y.real diff --git a/bandsaunter/scanner.py b/bandsaunter/scanner.py index 7b3f125..032353a 100755 --- a/bandsaunter/scanner.py +++ b/bandsaunter/scanner.py @@ -21,7 +21,7 @@ import numpy as np from . import dsp from .bandplan import fmt_hz, presets_covering -from .classify import classify +from .classify import classify, ssb_alignment from .config import ScanConfig from .demod import make_demodulator from .device import RtlSdrDevice, RtlSdrError @@ -458,9 +458,50 @@ class Scanner: return "am" if carrier_ratio < 0.15 and env_cv > 0.6: # No carrier left, and the envelope carrying the whole signal. + # Which sideband is settled later, by measurement; the band plan + # is a better guess than the HF convention in the meantime, since + # it knows the segments the convention gets wrong -- 60 m and the + # HF utility bands are upper sideband well below 10 MHz. + if fallback in ("usb", "lsb"): + return fallback return "lsb" if freq_hz < 10_000_000 else "usb" return fallback + # A sideband read this weakly is no better than the band plan's word. + _SIDEBAND_CONFIDENCE = 0.12 + + def _align_ssb(self, x: np.ndarray, mode: str, + det: Detection, fine_bw: float) -> tuple[str, float]: + """Put an SSB capture on the suppressed carrier rather than the voice. + + Detection reports the centroid of the energy, which is what every + other mode wants: an FM discriminator and an AM envelope detector do + not care where in their passband the signal sits. SSB is demodulated + by a filter that opens at the carrier, so centring on the middle of + the voice cuts off its lower half and shifts the rest down by a + couple of kilohertz -- the mistuned sound that makes SSB unusable + rather than merely imperfect. + + Returns the sideband to demodulate and the carrier's offset in hertz. + """ + if mode not in ("usb", "lsb", "ssb"): + return mode, 0.0 + hint = mode if mode in ("usb", "lsb") else "" + al = ssb_alignment(x, self.device.sample_rate, fine_bw) + if al is None: + return hint or "usb", 0.0 + if al.confidence >= self._SIDEBAND_CONFIDENCE or not hint: + sideband = al.sideband + else: + sideband = hint + offset = al.carrier_for(sideband) + # A correction larger than the signal itself means the band was read + # wrong, and moving that far would tune away from it. + if abs(offset) > max(4_000.0, al.width_hz): + return sideband, 0.0 + det.frequency += offset + return sideband, offset + def _lo_offset(self, freq: float, bw: float) -> float: """How far to offset the LO so the DC spike misses the signal.""" if self.device.direct_sampling_mode != 0: @@ -522,6 +563,7 @@ class Scanner: self._error(exc) probe = None + carrier_offset = 0.0 try: n_probe = int(self.device.sample_rate * cfg.probe_seconds) probe = (self.device.read_stream(n_probe) if streamed @@ -541,11 +583,14 @@ class Scanner: mode = self._demod_from_signal(measured, det.frequency, mode, bw, signal_bw=fine_bw, snr_db=det.snr_db) + mode, carrier_offset = self._align_ssb(measured, mode, det, fine_bw) except RtlSdrError as exc: self._error(exc) demod = make_demodulator(mode, self.device.sample_rate, bw, - cfg.audio_rate) + cfg.audio_rate, + **({"carrier_offset_hz": carrier_offset} + if carrier_offset else {})) rec = Recording( root=Path(cfg.output_dir).expanduser(), diff --git a/bandsaunter/simulator.py b/bandsaunter/simulator.py index cd371e9..1eaafec 100755 --- a/bandsaunter/simulator.py +++ b/bandsaunter/simulator.py @@ -26,14 +26,20 @@ _SPEECH_CACHE: dict = {} def _speech_loop(seed: int, pitch: float = 120.0, seconds: float = 8.0, - phrases: bool = True) -> np.ndarray: + phrases: bool = True, + band: tuple[float, float] | None = None) -> np.ndarray: """Render a loop of synthetic speech at :data:`_SPEECH_RATE` (cached). Glottal pulse train with pitch drift and jitter, shaped by three swept formant resonators, gated into syllables with pauses between phrases -- the structure that separates a voice from a tone or a hum. + + ``band`` applies the transmitter's audio filter. An SSB transmitter is + filtered to a few kilohertz before the modulator -- that filter is what + makes the signal single-sideband in the first place -- so without it the + simulated signal is several times wider than anything on the air. """ - key = (seed, round(pitch, 1), seconds, phrases) + key = (seed, round(pitch, 1), seconds, phrases, band) cached = _SPEECH_CACHE.get(key) if cached is not None: return cached @@ -85,6 +91,11 @@ def _speech_loop(seed: int, pitch: float = 120.0, seconds: float = 8.0, phrase = (np.sin(2 * np.pi * 0.35 * t) > -0.5).astype(float) k = max(1, int(0.02 * fs)) out *= sps.lfilter(np.ones(k) / k, [1.0], phrase) + if band is not None: + lo, hi = band + taps = sps.firwin(255, [lo, min(hi, fs / 2 * 0.95)], + pass_zero=False, fs=fs) + out = sps.filtfilt(taps, [1.0], out) out /= max(np.abs(out).max(), 1e-9) _SPEECH_CACHE[key] = out @@ -151,7 +162,10 @@ class VirtualTransmitter: out = ((1.0 + 0.6 * self._audio(t)) * np.exp(1j * self._advance(np.zeros(n), fs))) elif m in ("usb", "lsb"): - audio = self._audio(t) + # Filtered to the transmitter's audio passband first: the width of + # an SSB signal is exactly the width of that filter. + hi = self.bandwidth if 1_000.0 < self.bandwidth < 6_000.0 else 2_800.0 + audio = self._audio(t, band=(300.0, hi)) from scipy.signal import hilbert an = hilbert(audio) out = an if m == "usb" else np.conj(an) @@ -181,7 +195,8 @@ class VirtualTransmitter: self._phase = float(ph[-1] % (2.0 * np.pi)) if ph.size else self._phase return ph - def _audio(self, t: np.ndarray) -> np.ndarray: + def _audio(self, t: np.ndarray, + band: tuple[float, float] | None = None) -> np.ndarray: """Speech on the RF time base. A synthetic talker (glottal pulses through moving formants, gated into @@ -196,7 +211,8 @@ class VirtualTransmitter: # short over carries speech rather than whatever silence happened to # line up with it. ptt = self.period_seconds > 0 and self.on_seconds > 0 - loop = _speech_loop(self._seed, self.pitch_hz, phrases=not ptt) + loop = _speech_loop(self._seed, self.pitch_hz, phrases=not ptt, + band=band) if ptt: shifted = t + self.phase_offset into_over = shifted % self.period_seconds diff --git a/tests/test_ssb.py b/tests/test_ssb.py new file mode 100644 index 0000000..8d2a970 --- /dev/null +++ b/tests/test_ssb.py @@ -0,0 +1,160 @@ +"""Single sideband: finding the carrier, reading the sideband, and the audio. + +SSB is the one mode where tuning has to be exact. An FM discriminator and an +AM envelope detector do not care where in their passband a signal sits, but an +SSB demodulator is a filter that opens at the suppressed carrier: tune to the +middle of the voice instead and its lower half is filtered away while the rest +comes out shifted down by the error. +""" +import numpy as np +import pytest + +from bandsaunter.classify import ssb_alignment +from bandsaunter.config import ScanConfig +from bandsaunter.demod import make_demodulator +from bandsaunter.ranges import parse_range_list +from bandsaunter.scanner import Scanner, ScannerCallbacks +from bandsaunter.simulator import (SimulatedDevice, VirtualTransmitter as V, + _speech_loop) + + +FS = 256_000.0 + + +def ssb(audio, fs, sideband="usb"): + """Analytic signal: single sideband with its carrier at 0 Hz.""" + from scipy.signal import hilbert + an = hilbert(audio).astype(np.complex64) + return an if sideband == "usb" else np.conj(an) + + +def at_offset(x, hz, fs=FS): + """The same signal seen by a receiver tuned `hz` above its carrier.""" + n = np.arange(x.size) + return (x * np.exp(-2j * np.pi * hz * n / fs)).astype(np.complex64) + + +def speech(seconds=0.8, fs=FS, band=(300.0, 2_800.0)): + """The simulator's talker, resampled onto the test's sample rate.""" + from scipy.signal import resample_poly + loop = _speech_loop(7, 130.0, band=band) + audio = resample_poly(loop, int(fs), 16_000) + n = int(seconds * fs) + return np.tile(audio, int(n / audio.size) + 1)[:n] + + +# --------------------------------------------------------------------------- +# Reading the signal +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("sideband", ["usb", "lsb"]) +@pytest.mark.parametrize("centre", [0.0, 1500.0, -1200.0, 2450.0]) +def test_the_carrier_is_found_wherever_the_receiver_sits(sideband, centre): + """Speech leans towards its carrier, which locates it and names the side.""" + x = at_offset(ssb(speech(), FS, sideband), centre) + al = ssb_alignment(x, FS, 3_000.0) + assert al is not None + assert al.sideband == sideband + # The carrier is at -centre relative to where the receiver is tuned. + assert al.carrier_for(al.sideband) == pytest.approx(-centre, abs=250.0) + + +def test_a_signal_with_no_lean_gives_no_opinion(): + """Flat noise cannot say which sideband it is, and must not pretend to. + + The caller falls back on the band plan, which does know. + """ + from scipy.signal import butter, lfilter + rng = np.random.default_rng(0) + b, a = butter(6, [300 / (FS / 2), 2_700 / (FS / 2)], btype="band") + flat = lfilter(b, a, rng.standard_normal(int(FS * 0.8))) + al = ssb_alignment(ssb(flat, FS, "usb"), FS, 3_000.0) + assert al is None or al.confidence < 0.12 + + +def test_something_far_too_wide_is_not_read_as_ssb(): + rng = np.random.default_rng(1) + wide = rng.standard_normal(int(FS * 0.5)) + 1j * rng.standard_normal(int(FS * 0.5)) + assert ssb_alignment(wide.astype(np.complex64), FS, 3_000.0) is None + + +# --------------------------------------------------------------------------- +# Recovering the audio +# --------------------------------------------------------------------------- + +def tones(audio, rate, want): + """Level in dB of each wanted frequency in a block of audio.""" + spec = np.abs(np.fft.rfft(audio * np.hanning(audio.size))) + freqs = np.fft.rfftfreq(audio.size, 1.0 / rate) + top = spec.max() + return [20.0 * np.log10(max(spec[np.argmin(np.abs(freqs - f))], 1e-12) + / max(top, 1e-12)) for f in want] + + +def two_tone(fs=FS, seconds=1.5): + t = np.arange(int(fs * seconds)) / fs + return (np.exp(2j * np.pi * 500.0 * t) + + 0.7 * np.exp(2j * np.pi * 1800.0 * t)).astype(np.complex64) + + +def test_ssb_audio_is_intact_when_the_carrier_offset_is_given(): + """Two known tones must come back at the frequencies they went in at.""" + x = at_offset(two_tone(), 2_000.0) # receiver 2 kHz above carrier + d = make_demodulator("usb", int(FS), 2_800, 16_000, + carrier_offset_hz=-2_000.0) + audio = d.process(x)[-16_000:] + levels = tones(audio, d.audio_rate, [500.0, 1800.0]) + assert max(levels) > -3.0, levels + assert min(levels) > -12.0, levels + + +def test_without_the_offset_the_same_capture_is_ruined(): + """The bug this guards: the voice filtered away and the rest shifted.""" + x = at_offset(two_tone(), 2_000.0) + d = make_demodulator("usb", int(FS), 2_800, 16_000) + audio = d.process(x)[-16_000:] + assert max(tones(audio, d.audio_rate, [500.0, 1800.0])) < -20.0 + + +# --------------------------------------------------------------------------- +# End to end +# --------------------------------------------------------------------------- + +def run(tmp_path, freq, mode, ranges, **over): + dev = SimulatedDevice(transmitters=[V(freq, mode, 0.4, 2_800, "ssb tx")]).open() + cfg = ScanConfig(ranges=parse_range_list(ranges), output_dir=str(tmp_path), + record_seconds=3.0, hang_seconds=1.0, min_record_seconds=0.3, + threshold_db=12, dwell_seconds=0.05, max_cycles=2, + revisit_seconds=0.2) + for k, v in over.items(): + setattr(cfg, k, v) + hits = [] + s = Scanner(cfg, device=dev, callbacks=ScannerCallbacks(on_record_end=hits.append)) + s.prepare() + s.run() + kept = [h for h in hits if h.kept] + assert kept, "nothing recorded" + return kept[0] + + +def test_an_ssb_capture_lands_on_the_carrier(tmp_path): + """Not on the middle of the voice, which is 2.5 kHz away from it.""" + h = run(tmp_path, 14_250_000, "usb", "14.2M-14.3M") + assert h.mode == "usb", h.classification + assert h.frequency == pytest.approx(14_250_000, abs=400.0) + + +def test_lower_sideband_lands_on_its_carrier_too(tmp_path): + h = run(tmp_path, 3_800_000, "lsb", "3.75M-3.85M") + assert h.mode == "lsb", h.classification + assert h.frequency == pytest.approx(3_800_000, abs=400.0) + + +def test_the_sideband_is_not_guessed_from_the_frequency(tmp_path): + """60 m is upper sideband, well below the 10 MHz the convention splits on. + + Deciding by convention alone demodulated this as LSB. + """ + h = run(tmp_path, 5_357_000, "usb", "5.3M-5.4M") + assert h.mode == "usb", h.classification + assert "USB" in h.classification