"""Trunking control channels: recognising them, and refusing to sit on them. A trunked system keeps one frequency transmitting a data stream around the clock so that every radio in the fleet knows which channel each conversation has been assigned to. There is no speech on it and it never stops, which makes it both the strongest and the most useless signal in the band: a scanner that treats it as content parks on it for the full record limit, then finds it again on the very next sweep. """ import numpy as np import pytest from bandsaunter.classify import (SignalFeatures, classify, trunk_control, CONTROL_SIGNATURES) from bandsaunter.config import ScanConfig from bandsaunter.quality import CATEGORIES, assess from bandsaunter.ranges import parse_range_list from bandsaunter.scanner import Scanner, ScannerCallbacks from bandsaunter.simulator import SimulatedDevice, VirtualTransmitter as V FS = 96_000.0 def features(**over) -> SignalFeatures: """A control channel as the feature extractor sees one. The numbers are those measured from a real 856.561 MHz SMARTNET capture, so a change that stops this being recognised has broken the case the detector was written for. """ f = SignalFeatures(sample_rate=32_000.0, n_samples=320_000, duration=10.0) f.bandwidth = 7_437.5 f.env_cv = 0.0268 f.ook_contrast_db = 5.411 f.duty_cycle = 0.898 f.freq_modes = 2 f.mode_spacing = 5_326.7 f.level_dwell = 0.5244 f.baud = 3_600.1 f.baud_stability = 1.0 f.baud_strength = 51.7 f.snr_db = 38.58 for k, v in over.items(): setattr(f, k, v) return f def fsk(baud, levels, seconds=2.0, fs=FS, deviation=2_600.0, gaps=False): """Continuous multi-level FSK -- the shape every control channel has. Built from the simulator's own transmitter so the transmit shaping is the same as everywhere else: an unshaped square-edged stream is a signal no licensed radio would radiate, and its harmonics fool any symbol-rate estimator. """ mode = "fsk2" if levels == 2 else "fsk4" tx = V(100e6, mode, 1.0, 12_500.0, f"ctl{baud:.0f}x{levels}", baud=float(baud), deviation=deviation) x = tx.generate(0.0, int(seconds * fs), fs) if gaps: # key it on and off, as a voice call would env = np.ones(x.size) for i in range(0, x.size, int(fs * 0.5)): env[i:i + int(fs * 0.2)] = 0.0 x = (x * env).astype(np.complex64) return x # -- the detector on its own ------------------------------------------------- def test_a_real_smartnet_control_channel_is_recognised(): c = trunk_control(features(), 856_561_096.0) assert c is not None assert "SMARTNET" in c.system assert c.confidence >= 0.9 assert not c.ambiguous def test_the_name_says_control_channel_in_words(): c = trunk_control(features(), 856_561_096.0) assert c.describe().endswith("trunking control channel") def test_being_in_a_trunked_band_is_corroboration_not_a_requirement(): """Trunking is licensed all over the spectrum, so the shape has to be enough on its own -- the allocation only raises confidence.""" inband = trunk_control(features(), 856_561_096.0) out = trunk_control(features(), 33_000_000.0) assert out is not None assert inband.confidence > out.confidence @pytest.mark.parametrize("baud,levels,name", [ (3_600.0, 2, "SMARTNET"), (9_600.0, 2, "EDACS"), ]) def test_unambiguous_symbol_rates_are_believed_at_once(baud, levels, name): """Nothing but a control channel sends these without pause, so waiting to see whether it stops would only waste the receiver's time.""" c = trunk_control(features(baud=baud, freq_modes=levels), 856_500_000.0, continuous_for=0.0) assert c is not None and name in c.system @pytest.mark.parametrize("baud,levels", [(4_800.0, 4), (2_400.0, 4), (1_200.0, 2)]) def test_shapes_shared_with_digital_voice_wait_for_the_carrier_to_prove_itself( baud, levels): """P25, DMR and NXDN control channels look exactly like a call on the same system. Only one of the two is still transmitting a minute later.""" f = features(baud=baud, freq_modes=levels, mode_spacing=1_800.0) assert trunk_control(f, 856_500_000.0, continuous_for=5.0) is None late = trunk_control(f, 856_500_000.0, continuous_for=30.0) assert late is not None and late.ambiguous def test_patience_is_adjustable(): f = features(baud=4_800.0, freq_modes=4, mode_spacing=1_800.0) assert trunk_control(f, 0.0, continuous_for=8.0, min_seconds=5.0) is not None assert trunk_control(f, 0.0, continuous_for=8.0, min_seconds=60.0) is None def test_a_bursty_data_signal_is_not_a_control_channel(): """A pager page or a packet frame has the right symbol rate and the wrong duty cycle: it stops, and a control channel never does.""" assert trunk_control(features(duty_cycle=0.2, ook_contrast_db=28.0), 929_000_000.0) is None def test_an_off_rate_data_stream_is_left_alone(): assert trunk_control(features(baud=5_000.0), 856_500_000.0) is None def test_a_wandering_symbol_rate_is_not_trusted(): """The cyclostationary estimator always returns *some* peak; a control channel's does not move.""" assert trunk_control(features(baud_stability=0.3), 856_500_000.0) is None def test_voice_never_looks_like_one(): """Speech swings the envelope around between syllables.""" assert trunk_control(features(env_cv=0.55), 856_500_000.0) is None def test_every_signature_is_reachable(): """A typo in the table would silently disable one system for ever.""" for baud, levels, name, conf, ambiguous in CONTROL_SIGNATURES: f = features(baud=baud, freq_modes=levels, mode_spacing=1_800.0) c = trunk_control(f, 0.0, continuous_for=999.0) assert c is not None and c.system == name, name # -- through the classifier -------------------------------------------------- def test_classify_names_the_control_channel_rather_than_the_modulation(): """"2-FSK" is true and useless; the point of spotting one is to say so.""" cls = classify(fsk(3_600.0, 2), FS, freq_hz=856_561_096.0, snr_db=40.0) assert cls.control is not None assert "control channel" in cls.label.lower() assert cls.reasons[0].startswith("unbroken 3600 baud") def test_a_keyed_digital_voice_call_is_not_called_a_control_channel(): cls = classify(fsk(4_800.0, 4, gaps=True), FS, freq_hz=856_500_000.0, snr_db=40.0) assert cls.control is None # -- through the content gate ------------------------------------------------ def test_trunk_is_a_content_category_of_its_own(): assert "trunk" in CATEGORIES def test_trunk_is_not_accepted_by_default(): assert "trunk" not in ScanConfig().accept def test_the_verdict_calls_it_trunk_and_refuses_to_keep_it(): cls = classify(fsk(3_600.0, 2), FS, freq_hz=856_561_096.0, snr_db=40.0) a = assess(cls, np.zeros(16_000), 16_000.0) assert a.category == "trunk" assert not a.accept assert a.control is not None assert "control channel" in a.reason def test_keep_control_puts_it_back_in_the_digital_pile(): """Someone feeding a decoder wants the control channel recorded.""" cls = classify(fsk(3_600.0, 2), FS, freq_hz=856_561_096.0, snr_db=40.0) a = assess(cls, np.zeros(16_000), 16_000.0, skip_control=False) assert a.category == "digital" # -- end to end -------------------------------------------------------------- def control_channel_scan(tmp_path, **over): tx = [V(856_562_500, "fsk2", 0.45, 12_500, "control", baud=3_600.0, deviation=2_600.0)] dev = SimulatedDevice(transmitters=tx).open() cfg = ScanConfig(ranges=parse_range_list("856.4M-856.7M"), output_dir=str(tmp_path), record_seconds=8.0, hang_seconds=1.0, min_record_seconds=0.3, threshold_db=12, dwell_seconds=0.05, max_cycles=1, revisit_seconds=0.2, verify_seconds=0.5) for k, v in over.items(): setattr(cfg, k, v) hits, notes = [], [] s = Scanner(cfg, device=dev, callbacks=ScannerCallbacks( on_record_end=hits.append, on_record_note=notes.append)) s.prepare() s.run() return s, hits, notes def test_a_control_channel_is_never_saved(tmp_path): s, hits, _ = control_channel_scan(tmp_path) assert hits, "the transmitter was never found at all" assert not any(h.kept for h in hits) assert any(h.category == "trunk" for h in hits), [h.category for h in hits] assert list(tmp_path.glob("*.wav")) == [] def test_it_is_abandoned_long_before_the_record_limit(tmp_path): """The whole point: not holding the receiver on a channel with nothing to hear. Eight seconds were allowed; it should leave in about one.""" s, hits, _ = control_channel_scan(tmp_path) trunk = [h for h in hits if h.category == "trunk"] assert trunk assert trunk[0].duration < 4.0, trunk[0].duration def test_the_display_is_told_what_it_is(tmp_path): s, hits, notes = control_channel_scan(tmp_path) assert notes, "nothing was sent to the live display" assert notes[0].startswith("TRUNK: ") assert "SMARTNET" in notes[0] def test_the_frequency_and_system_are_reported_in_the_status_line(tmp_path): tx = [V(856_562_500, "fsk2", 0.45, 12_500, "control", baud=3_600.0, deviation=2_600.0)] dev = SimulatedDevice(transmitters=tx).open() cfg = ScanConfig(ranges=parse_range_list("856.4M-856.7M"), output_dir=str(tmp_path), record_seconds=8.0, hang_seconds=1.0, threshold_db=12, dwell_seconds=0.05, max_cycles=1, verify_seconds=0.5) msgs = [] s = Scanner(cfg, device=dev, callbacks=ScannerCallbacks(on_status=msgs.append)) s.prepare() s.run() trunk = [m for m in msgs if m.startswith("TRUNK ")] assert trunk, msgs assert "856.5" in trunk[0] and "control channel" in trunk[0] assert s.stats.control_channels def test_keep_control_records_it_like_any_other_data(tmp_path): s, hits, _ = control_channel_scan(tmp_path, skip_control=False) kept = [h for h in hits if h.kept] assert kept, [(h.category, h.stop_reason) for h in hits] assert kept[0].category == "digital" def test_lockout_control_writes_it_down_when_asked(tmp_path): s, hits, _ = control_channel_scan(tmp_path, lockout_control=True, save_lockouts=False) assert s.cfg.lockout, "the control channel was not locked out" assert s._is_locked(856_562_500) def test_lockout_control_is_off_by_default(tmp_path): s, hits, _ = control_channel_scan(tmp_path) assert not s.cfg.lockout def test_it_is_skipped_even_with_the_content_gate_off(tmp_path): """--keep-everything turns off *judging*, not spotting. A control channel is still recognised, and the receiver still leaves promptly.""" s, hits, notes = control_channel_scan(tmp_path, require_signal=False) assert notes and notes[0].startswith("TRUNK: ") trunk = [h for h in hits if h.category == "trunk"] assert trunk and not trunk[0].kept assert trunk[0].duration < 4.0, trunk[0].duration assert list(tmp_path.glob("*.wav")) == [] def test_a_simulated_lockout_never_reaches_the_real_settings(tmp_path, capsys): """The demo band is invented. A lock-out taken from it must not be written into a settings file that a real scan will later read -- it would sit there for ever, skipping whatever genuine signal happened to land near a made-up frequency.""" from bandsaunter.cli import main settings = tmp_path / "config.yaml" settings.write_text("record_seconds: 4.0\nsave_lockouts: true\n") rc = main(["scan", "--simulate", "-r", "856.4M-856.7M", "-o", str(tmp_path / "out"), "--plain", "--cycles", "1", "--record", "8", "--lockout-control", "--profile", str(settings)]) assert rc == 0 out = capsys.readouterr().out assert "locked out" in out and "for this run" in out, out after = settings.read_text() assert "lockout:" not in after, after assert "856" not in after, after