dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Recognise trunking control channels, and refuse to sit on them

Browse source 
A trunked system keeps one frequency transmitting a data stream around
the clock so its radios know where each conversation has been put.  There
is no speech on it and it never stops, which makes it the strongest and
most useless signal in the band: the scanner parked on 856.561 MHz for
the full record limit, saved four minutes of buzzing, and found it again
on the next sweep.

Five signatures, matched against a constant-envelope stream that never
pauses: 3600 baud two-level (Motorola SMARTNET/SmartZone), 9600 (EDACS),
1200 (MPT-1327), 4800 four-level (P25 or DMR Tier III), 2400 (NXDN).
The first two are believed at once -- nothing else sends at those rates
without pausing.  The rest share their shape with a digital voice call on
the same system, so they wait for the carrier to run unbroken past
--control-seconds, longer than a conversation goes without a breath.
Being in a trunked allocation raises confidence but is never required;
trunking is licensed on business pairs all over the spectrum.

One is named on screen, abandoned within a second or so, and its capture
deleted.  --keep-control records them for a decoder; --lockout-control
writes them into the lock-out list.

Three things had to be fixed to get there.

The simulator's "pseudo-random" symbols were a counter: multiplying the
symbol index by an odd constant and taking it modulo the level count
returns the low bits, so two-level FSK came out 0,1,0,1.  Every FSK test
in the suite was measuring a tone.  Its FSK is now shaped the way GFSK
and C4FM shape a stream, too, square-edged keying being a signal no
licensed transmitter would radiate.

The symbol-rate estimator locked onto harmonics -- 3600 baud read as
18000 -- because a transition impulse train is a comb of equal lines; it
now walks down to the fundamental.  The squared envelope is no longer a
candidate: it is not a transition signal, and its DC lobe made every
random OOK signal measure ninety baud.  The search starts at 200 Hz
rather than 40, below which it was reading drift, which is how a bare
carrier was awarded a symbol rate.  And a clean two-level signal counted
zero discriminator levels, because its modes land in the first and last
histogram bin, where find_peaks cannot see them.

Separately: locking out a frequency wrote to the settings file even under
--no-config, which has no settings file by definition.  It now writes
only where it read from, and --simulate never writes at all -- an
invented frequency would sit in a real config for ever, skipping whatever
genuine signal happened to land near it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
This commit is contained in:
The Dust Council 2026-08-22 14:22:49 -07:00
parent ba6c925351
commit 4a272eb1d5
14 changed files with 984 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

86
README.md
View file

@ -450,8 +450,13 @@ only for:
| `cw` | a keyed carrier whose timing resolves as Morse | | `cw` | a keyed carrier whose timing resolves as Morse |
| `digital` | an identified keying scheme: discrete FSK levels, an M-PSK phase line, or on-off keying -- corroborated by a symbol rate | | `digital` | an identified keying scheme: discrete FSK levels, an M-PSK phase line, or on-off keying -- corroborated by a symbol rate |
Two more categories exist but are not accepted by default: `carrier`
(unmodulated, real but empty) and `trunk` (a [trunking control
channel](#trunked-systems-and-their-control-channels)).
Everything else -- static, hum, switch-mode power supply harmonics, clock Everything else -- static, hum, switch-mode power supply harmonics, clock
spurs, bare carriers -- is discarded, and the files it wrote are deleted. spurs, bare carriers, trunking control channels -- is discarded, and the files
it wrote are deleted.
The check runs *while* the capture is still going, so interference is dropped The check runs *while* the capture is still going, so interference is dropped
after a second or two instead of holding the receiver for the whole record after a second or two instead of holding the receiver for the whole record
@ -574,6 +579,62 @@ Each result carries a confidence and the reasoning behind it:
Low SNR reduces confidence rather than producing a confident wrong answer. Low SNR reduces confidence rather than producing a confident wrong answer.
### Trunked systems and their control channels
Police, fire and most large business radio in the US runs on *trunked*
systems. Rather than giving each department a frequency of its own, the system
owns a pool of channels and hands one out per conversation. For that to work,
one frequency is given over entirely to a data stream that runs day and night
telling every radio in the fleet where to go next. That frequency is the
**control channel**.
It is the worst thing a scanner can find: loud, perfectly steady, never
silent, and with nothing on it to hear — just a harsh buzz. Left to itself a
scanner parks on it for the whole record limit, saves the file, and finds it
again on the next sweep, for as long as it runs.
bandsaunter recognises one and moves on, usually within a second or two:
```
TRUNK 856.561096 MHz -- Motorola SMARTNET / SmartZone (Type I/II) control channel, 3600 baud -- skipping
```
In the live display the recording panel turns yellow and says `TRUNK:` with
the system name instead of `REC`, and the end-of-run summary lists every
control channel found and where it was.
What identifies one is a constant-envelope data stream that never pauses, at a
symbol rate belonging to a known trunking standard:
| Symbol rate | Levels | System |
|---|---|---|
| 3600 baud | 2 | Motorola SMARTNET / SmartZone (Type I/II) |
| 9600 baud | 2 | EDACS / ProVoice |
| 1200 baud | 2 | MPT-1327 |
| 4800 baud | 4 | P25 or DMR Tier III |
| 2400 baud | 4 | NXDN / NEXEDGE |
The first two are called immediately — nothing else transmits at those rates
without pausing. The rest share their shape with an ordinary digital voice
call on the same system, so they are only judged to be a control channel once
the carrier has run unbroken for `--control-seconds` (20 s by default), which
is longer than a real conversation goes without taking a breath. Raise it if
digital voice is being skipped by mistake.
Sitting in a band where trunking is common raises confidence but is never
required — trunking is licensed on business pairs all over the spectrum, so
the shape of the signal has to be enough on its own.
```bash
bandsaunter scan -b 800-trunked # control channels named and skipped
bandsaunter scan -b 800-trunked --keep-control # record them (for a decoder)
bandsaunter scan -b 800-trunked --lockout-control # never look at them again
```
`--lockout-control` adds each one to the lock-out list as it is found; with
lock-out saving on (the default) that list is written to your settings file
and survives a restart.
### CW / Morse ### CW / Morse
Keyed carriers are decoded to text. The speed is measured from the signal, so Keyed carriers are decoded to text. The speed is measured from the signal, so
@ -863,11 +924,18 @@ span is taken exactly as written, since a noisy stretch of spectrum has a
definite width rather than a point with a guess around it. Ranges accept the definite width rather than a point with a guess around it. Ranges accept the
same forms as everywhere else — `450M-455M`, `450-455M`, `88M to 108M`. same forms as everywhere else — `450M-455M`, `450-455M`, `88M to 108M`.
`--lockout-control` adds each trunking control channel to the list as it is
found. Two runs never write anything back: `--no-config` has no settings file
to write to, since the point of the flag is to leave the saved settings alone;
and `--simulate` is looking at an invented band, whose frequencies would sit in
a real settings file for ever, skipping whatever genuine signal happened to
land near one. Both still lock out for the run in hand, and say so.
## Built-in help ## Built-in help
Press `h` in the menus for topics covering setup, how the sweep works, why Press `h` in the menus for topics covering setup, how the sweep works, why
nothing (or too much) is being recorded, capturing conversations, where files nothing (or too much) is being recorded, capturing conversations, where files
go, HF reception and the keys available during a scan. Typing a setting name go, trunked systems, HF reception and the keys available during a scan. Typing a setting name
there explains that setting instead. there explains that setting instead.
From the command line, `bandsaunter config --describe <setting>` does the same, From the command line, `bandsaunter config --describe <setting>` does the same,
@ -925,8 +993,15 @@ signal type, which is also what the test suite runs against:
```bash ```bash
bandsaunter scan -r 144M-148M --simulate bandsaunter scan -r 144M-148M --simulate
bandsaunter scan -r 856.4M-856.7M --simulate # the control channel, skipped
``` ```
The demo band holds 2 m FM voice with a CTCSS tone, a repeater, a CW beacon,
NOAA weather radio, airband AM, an FM broadcast station, P25-style digital
voice, a POCSAG pager, a bare carrier, a 433 MHz ISM remote, and a SMARTNET
control channel that never stops transmitting — because that last one is only
interesting if it behaves the way the real thing does.
## Testing ## Testing
```bash ```bash
@ -946,7 +1021,12 @@ then gated into syllables and phrases -- because sine tones would not exercise
the speech detector at all. SSB transmitters are filtered to their audio the speech detector at all. SSB transmitters are filtered to their audio
passband first, since that filter is what makes a signal single-sideband, and passband first, since that filter is what makes a signal single-sideband, and
without it the simulated signal was several times wider than anything on the without it the simulated signal was several times wider than anything on the
air. air. Its FSK transmitters are shaped the way GFSK and C4FM shape a symbol
stream, because square-edged keying is a signal no licensed radio would
radiate, and its symbols are genuinely pseudo-random: an earlier version
multiplied the symbol index by an odd constant and took it modulo the level
count, which returns the low bits of a counter -- 0, 1, 0, 1 -- so every FSK
test was measuring a tone rather than data.
Its transmitters seed themselves deterministically, so a test that fails can be Its transmitters seed themselves deterministically, so a test that fails can be
made to fail again -- the one thing needed to fix it. made to fail again -- the one thing needed to fix it.

2
bandsaunter/__init__.py
View file

@ -9,7 +9,7 @@ and transcribing speech.
# 2026-08-21_02 is the second build made on the 21st. The revision is padded # 2026-08-21_02 is the second build made on the 21st. The revision is padded
# to two digits so versions sort as text. # to two digits so versions sort as text.
VERSION_DATE = "2026-08-22" VERSION_DATE = "2026-08-22"
VERSION_REVISION = 1 VERSION_REVISION = 2
__version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}" __version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}"

173
bandsaunter/classify.py
View file

@ -21,7 +21,8 @@ from .dsp import (db, instantaneous_frequency,
occupied_bandwidth, spectral_flatness, welch_psd) occupied_bandwidth, spectral_flatness, welch_psd)
__all__ = ["classify", "Classification", "SignalFeatures", "extract_features", __all__ = ["classify", "Classification", "SignalFeatures", "extract_features",
"CTCSS_TONES", "detect_ctcss", "SSBAlignment", "ssb_alignment"] "CTCSS_TONES", "detect_ctcss", "SSBAlignment", "ssb_alignment",
"ControlChannel", "trunk_control", "CONTROL_SIGNATURES"]
def _pow2_floor(n: int, cap: int = 1 << 16) -> int: def _pow2_floor(n: int, cap: int = 1 << 16) -> int:
@ -114,6 +115,7 @@ class Classification:
alternatives: list[tuple[str, float]] = field(default_factory=list) alternatives: list[tuple[str, float]] = field(default_factory=list)
suggested_mode: str = "nfm" suggested_mode: str = "nfm"
features: SignalFeatures | None = None features: SignalFeatures | None = None
control: "ControlChannel | None" = None
def summary(self) -> str: def summary(self) -> str:
pct = int(round(self.confidence * 100)) pct = int(round(self.confidence * 100))
@ -142,8 +144,18 @@ def _otsu(values: np.ndarray, bins: int = 128) -> float:
return float(centres[int(np.argmax(var_between))]) return float(centres[int(np.argmax(var_between))])
# Nothing this program identifies keys slower than a few hundred baud -- the
# slowest named system is 512 baud POCSAG -- and below a couple of hundred
# hertz the transition signals are dominated by slow drift in gain and
# frequency, whose spectrum is a lobe running down to DC rather than a line.
# Searching from 40 Hz meant a bare carrier was routinely awarded a "symbol
# rate" of fifty-something baud, taken from the bottom edge of that lobe.
SYMBOL_RATE_FLOOR_HZ = 200.0
def _cyclic_line(feature: np.ndarray, fs: float, def _cyclic_line(feature: np.ndarray, fs: float,
lo_hz: float = 40.0, hi_hz: float | None = None): lo_hz: float = SYMBOL_RATE_FLOOR_HZ,
hi_hz: float | None = None):
"""Find the strongest periodic line in a nonnegative feature signal. """Find the strongest periodic line in a nonnegative feature signal.
Symbol transitions sit on a symbol-rate grid, so the transition-magnitude Symbol transitions sit on a symbol-rate grid, so the transition-magnitude
@ -171,6 +183,30 @@ def _cyclic_line(feature: np.ndarray, fs: float,
k = int(np.argmax(sub)) k = int(np.argmax(sub))
peak = sub[k] peak = sub[k]
med = np.median(sub) + 1e-12 med = np.median(sub) + 1e-12
if peak <= 0:
return 0.0, 0.0
# Symbol transitions are impulses on the symbol grid, and an impulse train
# has a *comb* of lines: the symbol rate and every multiple of it, all of
# comparable height. Taking the tallest therefore returns a harmonic as
# often as the fundamental -- 3600 baud came back as 18000. So walk back
# down the comb and report the lowest sub-multiple that still carries a
# line of its own.
df = float(freqs[1] - freqs[0])
for m in range(8, 1, -1):
want = subf[k] / m
if want < lo_hz:
continue
j = int(round((want - subf[0]) / df))
w = max(1, int(round(0.02 * want / df))) # +/- 2%, for a little drift
lo_i, hi_i = max(0, j - w), min(sub.size, j + w + 1)
if hi_i <= lo_i:
continue
window = sub[lo_i:hi_i]
height = float(window.max())
if height > 0.30 * peak and height > 6.0 * med:
i = lo_i + int(np.argmax(window))
return float(subf[i]), float(20.0 * math.log10(height / med))
return float(subf[k]), float(20.0 * math.log10(peak / med)) return float(subf[k]), float(20.0 * math.log10(peak / med))
@ -182,6 +218,12 @@ def _count_modes(values: np.ndarray, weights: np.ndarray | None = None,
lo, hi = np.percentile(values, [1.0, 99.0]) lo, hi = np.percentile(values, [1.0, 99.0])
if hi <= lo: if hi <= lo:
return 0, 0.0, np.zeros(0) return 0, 0.0, np.zeros(0)
# Leave room at both ends. find_peaks cannot return the first or last
# bin, and a clean two-level FSK signal puts its two modes exactly there:
# the levels *are* the 1st and 99th percentiles. Without the margin such
# a signal counts zero modes, and the cleaner it is the worse it gets.
margin = 0.10 * (hi - lo)
lo, hi = lo - margin, hi + margin
hist, edges = np.histogram(values, bins=bins, range=(lo, hi), weights=weights) hist, edges = np.histogram(values, bins=bins, range=(lo, hi), weights=weights)
hist = hist.astype(np.float64) hist = hist.astype(np.float64)
if hist.sum() == 0: if hist.sum() == 0:
@ -564,15 +606,21 @@ def extract_features(x: np.ndarray, sample_rate: float,
f.psk_order, f.psk_strength = order, strength f.psk_order, f.psk_strength = order, strength
# ---- symbol rate --------------------------------------------------- # ---- symbol rate ---------------------------------------------------
# For FSK the transition magnitude of the discriminator carries the line; # Transition magnitude: for FSK the discriminator's, for anything that
# for linear modulations the squared envelope does. # keys its amplitude the envelope's. Both are impulse trains sitting on
# the symbol grid, which is what puts a line at the symbol rate.
#
# The squared envelope used to be offered as a third candidate. It is not
# a transition signal: for data-carrying NRZ its spectrum is a sinc lobe
# peaking at DC, so the lowest bin in the search band beat the band median
# by 40-odd dB and won every time. Every random OOK signal measured about
# 90 baud because of it.
cand = [] cand = []
if ifreq.size > 512: if ifreq.size > 512:
cand.append((_cyclic_line(np.abs(np.diff(ifreq)), sample_rate), cand.append((_cyclic_line(np.abs(np.diff(ifreq)), sample_rate),
np.abs(np.diff(ifreq)))) np.abs(np.diff(ifreq))))
cand.append((_cyclic_line(np.abs(np.diff(env)), sample_rate), cand.append((_cyclic_line(np.abs(np.diff(env)), sample_rate),
np.abs(np.diff(env)))) np.abs(np.diff(env))))
cand.append((_cyclic_line(env ** 2, sample_rate), env ** 2))
(baud, strength), winner = max(cand, key=lambda c: c[0][1]) (baud, strength), winner = max(cand, key=lambda c: c[0][1])
if strength > 8.0: if strength > 8.0:
f.baud, f.baud_strength = baud, strength f.baud, f.baud_strength = baud, strength
@ -925,16 +973,118 @@ def _marine_channel(freq_hz: float) -> str:
return "" return ""
# ---------------------------------------------------------------------------
# Trunked-radio control channels
# ---------------------------------------------------------------------------
# A trunked system keeps one channel permanently transmitting a data stream
# that tells the radios in the fleet which channel each conversation has been
# assigned. There is no speech on it and it never stops, so a scanner that
# treats it as a signal parks on it for as long as the record limit allows and
# then finds it again on the next sweep. It is the single most common way for
# an unattended scan to fill a disk with nothing.
#
# Each entry is (baud, levels, name, confidence, ambiguous). ``ambiguous``
# marks a shape that digital *voice* also uses, which must therefore be held
# to a much longer run of unbroken carrier before it is called a control
# channel -- a P25 talkgroup and a P25 control channel look alike for the
# first few seconds, and only the control channel is still there a minute
# later.
CONTROL_SIGNATURES: tuple[tuple[float, int, str, float, bool], ...] = (
(3600.0, 2, "Motorola SMARTNET / SmartZone (Type I/II)", 0.90, False),
(9600.0, 2, "EDACS / ProVoice", 0.80, False),
(1200.0, 2, "MPT-1327", 0.62, True),
(4800.0, 4, "P25 or DMR Tier III", 0.75, True),
(2400.0, 4, "NXDN / NEXEDGE", 0.65, True),
)
# Allocations where trunked systems live in the US. Being inside one is
# corroboration, never a requirement: trunking turns up on licensed business
# pairs all over the spectrum, and the shape of the signal is the real
# evidence.
_TRUNKED_BANDS = (
(136.0, 174.0), (380.0, 400.0), (406.0, 420.0), (450.0, 470.0),
(470.0, 512.0), (758.0, 775.0), (788.0, 805.0), (806.0, 824.0),
(851.0, 869.0), (896.0, 902.0), (935.0, 941.0),
)
@dataclass
class ControlChannel:
"""A trunking control channel, and how sure we are of it."""
system: str
confidence: float
reason: str
baud: float = 0.0
ambiguous: bool = False
def describe(self) -> str:
return f"{self.system} trunking control channel"
def _continuous_data(f: SignalFeatures) -> bool:
"""True when the signal is an unbroken, constant-envelope data stream.
Every control channel looks like this. Bursty data -- a pager page, an
ACARS message, a packet frame -- does not, and neither does voice, which
swings the envelope around as the speaker pauses.
"""
return (f.env_cv < 0.30 and f.ook_contrast_db < 12.0
and f.duty_cycle > 0.75 and f.freq_modes >= 2
and f.mode_spacing > 500.0 and f.level_dwell > 0.30)
def trunk_control(f: SignalFeatures, freq_hz: float = 0.0,
continuous_for: float = 0.0,
min_seconds: float = 20.0) -> ControlChannel | None:
"""Identify a trunking control channel from its shape and symbol rate.
``continuous_for`` is how many seconds of unbroken carrier this capture
has seen. Signatures shared with digital voice are only believed once
that passes ``min_seconds``; the unambiguous symbol rates -- nothing but a
control channel sends 3600 baud two-level FSK without pause -- are
believed straight away.
"""
if f.baud <= 0 or f.baud_stability < 0.60 or f.baud_strength < 12.0:
return None
if not _continuous_data(f):
return None
levels = 4 if f.freq_modes >= 4 else 2
for baud, want_levels, name, conf, ambiguous in CONTROL_SIGNATURES:
if want_levels != levels or not _baud_near(f, baud, 0.06):
continue
if ambiguous and continuous_for < min_seconds:
return None
reason = (f"unbroken {baud:.0f} baud {want_levels}-level data with no "
f"speech and no gaps -- a trunking control channel")
if ambiguous:
reason += f", still transmitting after {continuous_for:.0f} s"
mhz = freq_hz / 1e6
if any(lo <= mhz <= hi for lo, hi in _TRUNKED_BANDS):
conf = min(0.95, conf + 0.05)
reason += " in a trunked allocation"
return ControlChannel(system=name, confidence=conf, reason=reason,
baud=f.baud, ambiguous=ambiguous)
return None
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Entry point # Entry point
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0, def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0,
snr_db: float = 0.0, analyse_audio: bool = True) -> Classification: snr_db: float = 0.0, analyse_audio: bool = True,
continuous_for: float = 0.0,
control_seconds: float = 20.0) -> Classification:
"""Identify what kind of signal ``x`` is. """Identify what kind of signal ``x`` is.
``x`` should be complex baseband centred on the signal. ``freq_hz`` is the ``x`` should be complex baseband centred on the signal. ``freq_hz`` is the
real-world centre frequency and is used only for the known-system lookup. real-world centre frequency and is used only for the known-system lookup.
``continuous_for`` is how long the carrier has been up without a break,
which is what separates a trunking control channel from the digital voice
call it otherwise resembles.
""" """
f = extract_features(x, sample_rate, snr_db=snr_db) f = extract_features(x, sample_rate, snr_db=snr_db)
filtered = f.extras.pop("_filtered", x) filtered = f.extras.pop("_filtered", x)
@ -956,6 +1106,9 @@ def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0,
label = f"SSB voice ({mode.upper()})" label = f"SSB voice ({mode.upper()})"
reasons.append(f"{mode.upper()} assumed from the band convention") reasons.append(f"{mode.upper()} assumed from the band convention")
control = trunk_control(f, freq_hz, continuous_for=continuous_for,
min_seconds=control_seconds)
system = _identify_system(freq_hz, f, family) system = _identify_system(freq_hz, f, family)
if system: if system:
sys_label, sys_conf, sys_reason = system sys_label, sys_conf, sys_reason = system
@ -966,6 +1119,13 @@ def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0,
else: else:
reasons.append(f"also consistent with {sys_label}") reasons.append(f"also consistent with {sys_label}")
# A control channel outranks the generic name for its modulation: "2-FSK"
# is true but useless, and the whole point of spotting one is to say so.
if control is not None:
label = control.describe()
score = max(score, control.confidence)
reasons.insert(0, control.reason)
# Low SNR means low trust, whatever the rules said. # Low SNR means low trust, whatever the rules said.
if f.snr_db < 8: if f.snr_db < 8:
score *= 0.65 score *= 0.65
@ -977,6 +1137,7 @@ def classify(x: np.ndarray, sample_rate: float, freq_hz: float = 0.0,
return Classification( return Classification(
label=label, family=family, confidence=round(min(0.99, score), 3), label=label, family=family, confidence=round(min(0.99, score), 3),
reasons=reasons, alternatives=alts, suggested_mode=mode, features=f, reasons=reasons, alternatives=alts, suggested_mode=mode, features=f,
control=control,
) )

12
bandsaunter/cli.py
View file

@ -325,6 +325,13 @@ def cmd_scan(args) -> int:
"hardware.[/grey62]") "hardware.[/grey62]")
return 1 return 1
if args.simulate and cfg.save_lockouts:
# The demo band is invented. A lock-out taken from it would sit in
# the real settings file for ever, skipping whatever genuine signal
# happened to land near a made-up frequency. Locking out still works
# for the run in hand; it is only the writing back that is refused.
cfg.save_lockouts = False
scanner = Scanner(cfg, device=device, callbacks=ScannerCallbacks()) scanner = Scanner(cfg, device=device, callbacks=ScannerCallbacks())
try: try:
scanner.prepare() scanner.prepare()
@ -435,6 +442,11 @@ def _print_summary(scanner: Scanner) -> None:
if worker.dropped: if worker.dropped:
bits.append(f"{worker.dropped} skipped, the recogniser fell behind") bits.append(f"{worker.dropped} skipped, the recogniser fell behind")
console.print(f"[grey62] {', '.join(bits)}[/grey62]") console.print(f"[grey62] {', '.join(bits)}[/grey62]")
if st.control_channels:
console.print(f"[yellow] {len(st.control_channels)} trunking control "
f"channel(s) skipped:[/yellow]")
for hz, name in sorted(st.control_channels.items()):
console.print(f"[grey62] {fmt_hz(hz):>14} {name}[/grey62]")
if st.rejected_by_category: if st.rejected_by_category:
drops = ", ".join(f"{n} {cat}" drops = ", ".join(f"{n} {cat}"
for cat, n in sorted(st.rejected_by_category.items(), for cat, n in sorted(st.rejected_by_category.items(),

17
bandsaunter/config.py
View file

@ -83,6 +83,11 @@ class ScanConfig:
verify_max_seconds: float = 6.0 # give up on a contentless capture by here verify_max_seconds: float = 6.0 # give up on a contentless capture by here
max_detections_per_step: int = 4 max_detections_per_step: int = 4
# -- trunking control channels --------------------------------------
skip_control: bool = True # spot them, abandon them, move on
control_seconds: float = 20.0 # unbroken carrier a look-alike must hold
lockout_control: bool = False # also add them to the lock-out list
# -- radio ---------------------------------------------------------- # -- radio ----------------------------------------------------------
device_index: int = 0 device_index: int = 0
sample_rate: int = 2_048_000 sample_rate: int = 2_048_000
@ -288,10 +293,16 @@ def remember_lockouts(cfg: "ScanConfig",
Returns where it was written, or None if it could not be -- locking out a Returns where it was written, or None if it could not be -- locking out a
birdie must never be the thing that ends a scan. birdie must never be the thing that ends a scan.
A run started with --no-config has no settings file, and writing one
anyway would be the opposite of what was asked: the flag exists to leave
the saved settings alone. So without a source to write back to, nothing
is written.
""" """
path = getattr(cfg, "_source_path", None) source = getattr(cfg, "_source_path", None)
path = Path(path) if path else (Path(directory or DEFAULT_CONFIG_DIR) if not source and directory is None:
/ "config.yaml") return None
path = Path(source) if source else (Path(directory) / "config.yaml")
try: try:
data = {} data = {}
if path.exists(): if path.exists():

31
bandsaunter/quality.py
View file

@ -6,6 +6,7 @@ capture is assessed for *content* before it is kept:
* **voice** -- speech structure in the demodulated audio * **voice** -- speech structure in the demodulated audio
* **cw** -- a keyed carrier whose timing resolves as Morse * **cw** -- a keyed carrier whose timing resolves as Morse
* **digital** -- a symbol rate, discrete FSK levels, or an M-PSK phase line * **digital** -- a symbol rate, discrete FSK levels, or an M-PSK phase line
* **trunk** -- a trunking control channel: continuous data, never any speech
* **carrier** -- a steady unmodulated carrier (real, but carries nothing) * **carrier** -- a steady unmodulated carrier (real, but carries nothing)
* **noise** -- no structure at all: static, interference, receiver artefacts * **noise** -- no structure at all: static, interference, receiver artefacts
@ -22,7 +23,7 @@ import numpy as np
__all__ = ["VoiceMetrics", "voice_metrics", "Assessment", "assess", __all__ = ["VoiceMetrics", "voice_metrics", "Assessment", "assess",
"CATEGORIES", "RAYLEIGH_CV"] "CATEGORIES", "RAYLEIGH_CV"]
CATEGORIES = ("voice", "cw", "digital", "carrier", "noise") CATEGORIES = ("voice", "cw", "digital", "trunk", "carrier", "noise")
# Envelope coefficient of variation for complex Gaussian noise: |x| is # Envelope coefficient of variation for complex Gaussian noise: |x| is
# Rayleigh distributed, so std/mean is exactly sqrt(4/pi - 1). # Rayleigh distributed, so std/mean is exactly sqrt(4/pi - 1).
@ -216,6 +217,7 @@ class Assessment:
reason: str = "" reason: str = ""
voice: VoiceMetrics = field(default_factory=VoiceMetrics) voice: VoiceMetrics = field(default_factory=VoiceMetrics)
noise_likeness: float = 0.0 noise_likeness: float = 0.0
control: object = None # ControlChannel when this is a control channel
def describe(self) -> str: def describe(self) -> str:
return f"{self.category} ({self.score:.2f}): {self.reason}" return f"{self.category} ({self.score:.2f}): {self.reason}"
@ -293,7 +295,7 @@ def digital_structure(f) -> tuple[float, list[str]]:
def assess(classification, audio: np.ndarray, audio_rate: float, def assess(classification, audio: np.ndarray, audio_rate: float,
morse=None, min_voice: float = 0.45, freq_hz: float = 0.0, morse=None, min_voice: float = 0.45, freq_hz: float = 0.0,
accept: tuple[str, ...] = ("voice", "cw", "digital"), accept: tuple[str, ...] = ("voice", "cw", "digital"),
min_score: float = 0.45) -> Assessment: min_score: float = 0.45, skip_control: bool = True) -> Assessment:
"""Decide what a capture contains and whether it should be kept. """Decide what a capture contains and whether it should be kept.
Deliberately does not route on the classifier's label. The label is a Deliberately does not route on the classifier's label. The label is a
@ -313,32 +315,43 @@ def assess(classification, audio: np.ndarray, audio_rate: float,
a.voice = vm a.voice = vm
dig_score, dig_bits = digital_structure(f) dig_score, dig_bits = digital_structure(f)
# 1. Morse that actually decoded. # 1. A trunking control channel, before anything else. It is a positive
if morse is not None and morse.is_morse: # identification rather than a failure to find content -- the stream is
# real data, and would otherwise be kept as "digital" and hold the
# receiver for the full record time on a channel with nothing to hear.
if skip_control and getattr(classification, "control", None) is not None:
c = classification.control
a.category = "trunk"
a.control = c
a.score = float(c.confidence)
a.reason = f"{c.describe()}: {c.reason}"
# 2. Morse that actually decoded.
elif morse is not None and morse.is_morse:
a.category = "cw" a.category = "cw"
a.score = float(morse.confidence) a.score = float(morse.confidence)
a.reason = f'Morse decoded at {morse.wpm:.0f} WPM: "{morse.text.strip()[:40]}"' a.reason = f'Morse decoded at {morse.wpm:.0f} WPM: "{morse.text.strip()[:40]}"'
# 2. Speech, whatever the modulation was called. # 3. Speech, whatever the modulation was called.
elif vm.score >= min_voice: elif vm.score >= min_voice:
a.category = "voice" a.category = "voice"
a.score = vm.score a.score = vm.score
a.reason = f"speech in the audio ({vm.describe()})" a.reason = f"speech in the audio ({vm.describe()})"
# 3. Symbol structure. # 4. Symbol structure.
elif dig_score > 0.3: elif dig_score > 0.3:
a.category = "digital" a.category = "digital"
a.score = min(0.95, 0.35 + dig_score) a.score = min(0.95, 0.35 + dig_score)
a.reason = "digital modulation: " + ", ".join(dig_bits) a.reason = "digital modulation: " + ", ".join(dig_bits)
# 4. A keyed carrier whose timing would not resolve as Morse. # 5. A keyed carrier whose timing would not resolve as Morse.
elif fam == "cw" and f.ook_contrast_db > 12: elif fam == "cw" and f.ook_contrast_db > 12:
a.category = "cw" a.category = "cw"
a.score = 0.55 a.score = 0.55
a.reason = (f"keyed carrier, {f.ook_contrast_db:.0f} dB on/off contrast " a.reason = (f"keyed carrier, {f.ook_contrast_db:.0f} dB on/off contrast "
"(timing did not resolve as Morse)") "(timing did not resolve as Morse)")
# 5. Broadcast FM gets a lenient path because most of it is music, which # 6. Broadcast FM gets a lenient path because most of it is music, which
# has no speech pitch track to find. That leniency is confined to # has no speech pitch track to find. That leniency is confined to
# signals that really are broadcast: in the FM band, or carrying a # signals that really are broadcast: in the FM band, or carrying a
# 19 kHz stereo pilot. Without the restriction, any wideband hump -- # 19 kHz stereo pilot. Without the restriction, any wideband hump --
@ -351,7 +364,7 @@ def assess(classification, audio: np.ndarray, audio_rate: float,
a.reason = ("programme audio on a wideband FM carrier" a.reason = ("programme audio on a wideband FM carrier"
+ (" with a 19 kHz stereo pilot" if f.stereo_pilot else "")) + (" with a 19 kHz stereo pilot" if f.stereo_pilot else ""))
# 6. A carrier that is really just a carrier. # 7. A carrier that is really just a carrier.
elif fam == "carrier" or (f.am_depth < 0.004 and f.fdev_rms < 150 elif fam == "carrier" or (f.am_depth < 0.004 and f.fdev_rms < 150
and f.ook_contrast_db < 6): and f.ook_contrast_db < 6):
a.category = "carrier" a.category = "carrier"

107
bandsaunter/scanner.py
View file

@ -65,6 +65,7 @@ class ScanStats:
dropped_samples: int = 0 dropped_samples: int = 0
truncated: int = 0 # captures cut off mid-transmission truncated: int = 0 # captures cut off mid-transmission
rejected_by_category: dict = field(default_factory=dict) rejected_by_category: dict = field(default_factory=dict)
control_channels: dict = field(default_factory=dict) # Hz -> system name
@property @property
def elapsed(self) -> float: def elapsed(self) -> float:
@ -80,6 +81,7 @@ class ScannerCallbacks:
on_record_start: callable = None # (Detection, Recording) on_record_start: callable = None # (Detection, Recording)
on_record_tick: callable = None # (Recording, elapsed, signal_present, snr) on_record_tick: callable = None # (Recording, elapsed, signal_present, snr)
on_record_end: callable = None # (HitRecord) on_record_end: callable = None # (HitRecord)
on_record_note: callable = None # (str) -- a label for the live capture
on_cycle: callable = None # (cycle_number) on_cycle: callable = None # (cycle_number)
on_status: callable = None # (str) on_status: callable = None # (str)
on_error: callable = None # (Exception) on_error: callable = None # (Exception)
@ -136,11 +138,14 @@ class Scanner:
with self._lock: with self._lock:
self._lockouts.append(entry.interval(self.cfg.lockout_width)) self._lockouts.append(entry.interval(self.cfg.lockout_width))
self.cfg.lockout.append(entry) self.cfg.lockout.append(entry)
if self.cfg.save_lockouts:
path = remember_lockouts(self.cfg) path = remember_lockouts(self.cfg) if self.cfg.save_lockouts else None
self._status(f"locked out {entry.describe()}" # Announced either way: the lock-out took effect whether or not it was
+ (f", remembered in {path.name}" if path else # written down, and saying nothing when it was not is how an operator
" (could not be saved)")) # ends up thinking the key did not work.
self._status(f"locked out {entry.describe()}"
+ (f", remembered in {path.name}" if path
else " for this run"))
def _key(self, freq_hz: float) -> int: def _key(self, freq_hz: float) -> int:
return int(round(freq_hz / self._bucket)) return int(round(freq_hz / self._bucket))
@ -628,8 +633,11 @@ class Scanner:
last_signal_at = 0.0 last_signal_at = 0.0
last_content_at = 0.0 last_content_at = 0.0
ever_signal = False ever_signal = False
was_present = False
signal_since = 0.0 # start of the current unbroken run
content_ever = False content_ever = False
checks_done = 0 checks_done = 0
continuous_for = 0.0
peak_snr = det.snr_db peak_snr = det.snr_db
stop_reason = "unknown" stop_reason = "unknown"
next_check = cfg.verify_seconds next_check = cfg.verify_seconds
@ -668,9 +676,16 @@ class Scanner:
snr = self._band_snr(x, bw) snr = self._band_snr(x, bw)
present = snr >= (squelch_off if ever_signal else squelch_on) present = snr >= (squelch_off if ever_signal else squelch_on)
if present: if present:
if not was_present:
signal_since = elapsed - x.size / fs
last_signal_at = elapsed last_signal_at = elapsed
ever_signal = True ever_signal = True
peak_snr = max(peak_snr, snr) peak_snr = max(peak_snr, snr)
was_present = present
# How long the carrier has been up without a break. A
# trunking control channel never lets this reset; a
# conversation on the same system always does.
continuous_for = elapsed - signal_since if present else 0.0
audio, iq = demod.step(x) audio, iq = demod.step(x)
rec.write_audio(audio, present=present, at=elapsed) rec.write_audio(audio, present=present, at=elapsed)
@ -685,9 +700,13 @@ class Scanner:
# Judge the content while it is still arriving so static and # Judge the content while it is still arriving so static and
# interference are dropped after a second or two instead of # interference are dropped after a second or two instead of
# holding the receiver for the full record time. # holding the receiver for the full record time. The check
if (cfg.require_signal and cfg.classify and ever_signal # also runs with the content gate off, because spotting a
and elapsed >= next_check): # control channel is worth doing whether or not anything is
# being thrown away for lack of content; in that case its only
# power is to end the capture, never to reject one.
if ((cfg.require_signal or cfg.skip_control) and cfg.classify
and ever_signal and elapsed >= next_check):
# Check more often once a gap has opened, so a channel # Check more often once a gap has opened, so a channel
# held open by static is released promptly instead of # held open by static is released promptly instead of
# waiting out a full verify interval. # waiting out a full verify interval.
@ -697,10 +716,23 @@ class Scanner:
window = max(2.0, cfg.verify_seconds) window = max(2.0, cfg.verify_seconds)
_, _, verdict = self._analyse( _, _, verdict = self._analyse(
rec, demod, det.frequency, peak_snr, rec, demod, det.frequency, peak_snr,
since=max(0.0, elapsed - window)) since=max(0.0, elapsed - window),
if verdict is not None: continuous_for=continuous_for)
if verdict is not None and cfg.require_signal:
checks_done += 1 checks_done += 1
if verdict.accept: if verdict is not None:
if verdict.category == "trunk":
# A positive identification, not a failure to find
# content: there is nothing to hear on a control
# channel however long we wait, so leave now
# rather than serving out verify_max_seconds.
rejected_early = verdict
stop_reason = verdict.reason
self._note_control(det.frequency, verdict)
break
if not cfg.require_signal:
pass # the gate is off; nothing to judge
elif verdict.accept:
content_ever = True content_ever = True
last_content_at = elapsed last_content_at = elapsed
elif not content_ever: elif not content_ever:
@ -806,6 +838,13 @@ class Scanner:
hit.content_reason = rejected_early.reason hit.content_reason = rejected_early.reason
hit.noise_likeness = round(rejected_early.noise_likeness, 3) hit.noise_likeness = round(rejected_early.noise_likeness, 3)
hit.voice_score = round(rejected_early.voice.score, 3) hit.voice_score = round(rejected_early.voice.score, 3)
control = rejected_early.control
if control is not None:
hit.classification = control.describe()
hit.family = "control"
hit.baud = control.baud
hit.confidence = control.confidence
hit.reasons = [control.reason]
elif not ever_signal: elif not ever_signal:
hit.category = "vanished" hit.category = "vanished"
hit.stop_reason = "false trigger (no signal after retuning)" hit.stop_reason = "false trigger (no signal after retuning)"
@ -814,7 +853,15 @@ class Scanner:
self._reject(rec, hit) self._reject(rec, hit)
return hit return hit
verdict = self._identify(rec, hit, demod) verdict = self._identify(rec, hit, demod, continuous_for=continuous_for)
if verdict is not None and verdict.category == "trunk":
# Recognised only at the end -- a short capture, or one that ran
# before the first content check. Same outcome, said the same way.
hit.kept = False
hit.stop_reason = verdict.reason
self._note_control(det.frequency, verdict)
self._reject(rec, hit)
return hit
if cfg.require_signal and verdict is not None and not verdict.accept: if cfg.require_signal and verdict is not None and not verdict.accept:
hit.kept = False hit.kept = False
hit.stop_reason = f"no signal content: {verdict.reason}" hit.stop_reason = f"no signal content: {verdict.reason}"
@ -862,6 +909,27 @@ class Scanner:
pass pass
return hit return hit
def _note_control(self, freq_hz: float, verdict) -> None:
"""Announce a trunking control channel, and optionally lock it out.
Said out loud every time rather than once: an operator watching a
band wants to know *which* frequencies are control channels, and
there is usually more than one.
"""
control = getattr(verdict, "control", None)
name = control.system if control is not None else "trunked system"
baud = f", {control.baud:.0f} baud" if control is not None else ""
self.stats.control_channels[round(freq_hz)] = name
self._status(f"TRUNK {fmt_hz(freq_hz)} -- {name} control channel"
f"{baud} -- skipping")
if self.cb.on_record_note:
try:
self.cb.on_record_note(f"TRUNK: {name}")
except Exception:
pass
if self.cfg.lockout_control and not self._is_locked(freq_hz):
self.lockout(freq_hz)
def _reject(self, rec: Recording, hit: HitRecord) -> None: def _reject(self, rec: Recording, hit: HitRecord) -> None:
"""Throw away a capture and everything it wrote.""" """Throw away a capture and everything it wrote."""
rec.discard() rec.discard()
@ -875,7 +943,8 @@ class Scanner:
pass pass
def _analyse(self, rec: Recording, demod, freq_hz: float, snr_db: float, def _analyse(self, rec: Recording, demod, freq_hz: float, snr_db: float,
final: bool = False, since: float | None = None): final: bool = False, since: float | None = None,
continuous_for: float = 0.0):
"""Classify what has been captured so far and judge whether to keep it. """Classify what has been captured so far and judge whether to keep it.
Returns ``(classification, morse, assessment)``. Called periodically Returns ``(classification, morse, assessment)``. Called periodically
@ -889,7 +958,9 @@ class Scanner:
if iq.size < 2048: if iq.size < 2048:
return None, None, None return None, None, None
try: try:
cls = classify(iq, demod.if_rate, freq_hz=freq_hz, snr_db=snr_db) cls = classify(iq, demod.if_rate, freq_hz=freq_hz, snr_db=snr_db,
continuous_for=continuous_for,
control_seconds=self.cfg.control_seconds)
except Exception as exc: except Exception as exc:
self._error(exc) self._error(exc)
return None, None, None return None, None, None
@ -916,7 +987,8 @@ class Scanner:
return assess(cls, audio, audio_rate, morse=morse, freq_hz=freq_hz, return assess(cls, audio, audio_rate, morse=morse, freq_hz=freq_hz,
min_voice=cfg.min_voice_score, min_voice=cfg.min_voice_score,
accept=tuple(cfg.accept), accept=tuple(cfg.accept),
min_score=cfg.min_signal_score) min_score=cfg.min_signal_score,
skip_control=cfg.skip_control)
def _decode_cw(self, iq: np.ndarray, demod): def _decode_cw(self, iq: np.ndarray, demod):
"""Run a dedicated CW detector over the captured IQ. """Run a dedicated CW detector over the captured IQ.
@ -971,12 +1043,13 @@ class Scanner:
meta_path=Path(hit.meta_path) if hit.meta_path else None) meta_path=Path(hit.meta_path) if hit.meta_path else None)
def _identify(self, rec: Recording, hit: HitRecord, demod, def _identify(self, rec: Recording, hit: HitRecord, demod,
analysis=None) -> Assessment | None: analysis=None, continuous_for: float = 0.0) -> Assessment | None:
"""Attach the classification, Morse text and content verdict to a hit.""" """Attach the classification, Morse text and content verdict to a hit."""
if not self.cfg.classify: if not self.cfg.classify:
return None return None
cls, morse, verdict = analysis if analysis is not None else \ cls, morse, verdict = analysis if analysis is not None else \
self._analyse(rec, demod, hit.frequency, hit.snr_db, final=True) self._analyse(rec, demod, hit.frequency, hit.snr_db, final=True,
continuous_for=continuous_for)
if cls is None: if cls is None:
return None return None

63
bandsaunter/settings.py
View file

@ -170,6 +170,7 @@ _TABLE: tuple[Setting, ...] = (
"voice - speech structure in the demodulated audio. " "voice - speech structure in the demodulated audio. "
"cw - a keyed carrier whose timing resolves as Morse. " "cw - a keyed carrier whose timing resolves as Morse. "
"digital - an identified keying scheme with a steady symbol rate. " "digital - an identified keying scheme with a steady symbol rate. "
"trunk - a trunked system's control channel. "
"carrier - a steady unmodulated carrier. " "carrier - a steady unmodulated carrier. "
"noise - everything else.", "noise - everything else.",
choices=CONTENT_CATEGORIES, flags=("--accept",), metavar="LIST", choices=CONTENT_CATEGORIES, flags=("--accept",), metavar="LIST",
@ -200,6 +201,29 @@ _TABLE: tuple[Setting, ...] = (
"conversation short.", "conversation short.",
unit="s", minimum=0.5, flags=("--verify-max",), metavar="SEC"), unit="s", minimum=0.5, flags=("--verify-max",), metavar="SEC"),
# -- trunking -------------------------------------------------------------
S("skip_control", "Skip trunk control channels", "Trunking", "bool",
"spot a trunked system's control channel and move straight on",
"A trunked radio system keeps one channel transmitting data day and "
"night so the radios know where each conversation has been put. There "
"is no speech on it and it never stops.",
flags=("--skip-control",),
off_flags=("--keep-control", "--no-skip-control")),
S("control_seconds", "Control channel patience", "Trunking", "float",
"unbroken carrier a digital-voice look-alike must hold",
"P25, DMR and NXDN control channels have the same shape as a digital "
"voice call on the same system. A carrier still running unbroken after "
"this long is a control channel, because a conversation would have "
"paused by now.",
unit="s", minimum=1.0, flags=("--control-seconds",), metavar="SEC"),
S("lockout_control", "Lock out control channels", "Trunking", "bool",
"add every control channel found to the lock-out list",
"Off by default: a control channel is cheap to recognise and skip, so "
"there is no need to write it down. Turn it on to stop revisiting it "
"at all.",
flags=("--lockout-control",),
off_flags=("--no-lockout-control",)),
# -- receiver ------------------------------------------------------------- # -- receiver -------------------------------------------------------------
S("device_index", "Device index", "Receiver", "int", S("device_index", "Device index", "Receiver", "int",
"which dongle to use when more than one is attached", "which dongle to use when more than one is attached",
@ -515,10 +539,40 @@ _GUIDANCE: dict[str, str] = {
"however empty.", "however empty.",
"accept": "accept":
"Which kinds of transmission are worth keeping: speech, Morse, " "Which kinds of transmission are worth keeping: speech, Morse, "
"data, plain unmodulated carriers, and noise. Most people want " "data, trunking control channels, plain unmodulated carriers, and "
"voice, cw and digital. Add carrier if you are hunting beacons or " "noise. Most people want voice, cw and digital. Add carrier if you "
"interference sources, and noise only for diagnosing why nothing is " "are hunting beacons or interference sources, trunk if you are "
"being recorded.", "collecting control channels for a decoder, and noise only for "
"diagnosing why nothing is being recorded.",
"skip_control":
"Trunked radio systems -- the kind police, fire and large businesses "
"use -- keep one frequency transmitting a data stream around the "
"clock. It tells the radios which channel to jump to for each "
"conversation; it carries no speech and it never goes quiet. To a "
"scanner it looks like a very strong, very interesting signal, so "
"without this setting the receiver parks on it, records the whole "
"record limit of buzzing, and finds it again on the next sweep. With "
"this on it is recognised within a second or two, named on screen, "
"and skipped. Leave it on unless you are deliberately collecting "
"control channel data to feed to a decoder.",
"control_seconds":
"Only matters for P25, DMR and NXDN. On those systems the control "
"channel and an ordinary digital conversation look identical for the "
"first few seconds, and the only thing that tells them apart is that "
"the conversation eventually pauses and the control channel never "
"does. This is how long a signal has to keep going without a break "
"before it is judged to be a control channel. Raise it if digital "
"voice calls are being skipped by mistake; lower it if you are tired "
"of waiting out control channels. It has no effect on the older "
"Motorola and EDACS systems, which are recognised immediately from "
"their symbol rate.",
"lockout_control":
"Write each control channel into the lock-out list as it is found, "
"so the scanner stops even looking at it. Skipping one already costs "
"only a second or two, so this is worth turning on mainly if you "
"scan the same band constantly and want the list built for you. With "
"'Remember lock-outs' also on, the entries are saved to your "
"settings file and survive a restart.",
"min_signal_score": "min_signal_score":
"How certain the content check must be before a recording is kept, " "How certain the content check must be before a recording is kept, "
"from 0 to 1. Lower it if real transmissions are being discarded, " "from 0 to 1. Lower it if real transmissions are being discarded, "
@ -930,6 +984,7 @@ _ARG_GROUP_TITLES = {
"Dwell and recording": "dwell behaviour", "Dwell and recording": "dwell behaviour",
"Detection": "detection", "Detection": "detection",
"What counts as a signal": "what counts as a signal", "What counts as a signal": "what counts as a signal",
"Trunking": "trunked systems",
"Receiver": "receiver", "Receiver": "receiver",
"Output": "output", "Output": "output",
"Run control": "run control", "Run control": "run control",

54
bandsaunter/simulator.py
View file

@ -177,8 +177,7 @@ class VirtualTransmitter:
out = env * np.exp(1j * self._advance(np.zeros(n), fs)) out = env * np.exp(1j * self._advance(np.zeros(n), fs))
elif m in ("fsk2", "fsk4"): elif m in ("fsk2", "fsk4"):
levels = 2 if m == "fsk2" else 4 levels = 2 if m == "fsk2" else 4
sym = self._symbols(t, fs, levels, self.baud) lv = self._shaped_levels(t, fs, levels, self.baud)
lv = (sym - (levels - 1) / 2.0) / max(1.0, (levels - 1) / 2.0)
out = np.exp(1j * self._advance(lv * self.deviation, fs)) out = np.exp(1j * self._advance(lv * self.deviation, fs))
elif m == "psk": elif m == "psk":
sym = self._symbols(t, fs, 4, self.baud) sym = self._symbols(t, fs, 4, self.baud)
@ -233,10 +232,49 @@ class VirtualTransmitter:
def _symbols(self, t: np.ndarray, fs: float, levels: int, def _symbols(self, t: np.ndarray, fs: float, levels: int,
baud: float) -> np.ndarray: baud: float) -> np.ndarray:
"""Deterministic pseudo-random symbols on an absolute-time grid.""" """Deterministic pseudo-random symbols on an absolute-time grid."""
idx = np.floor(t * baud).astype(np.int64) idx = np.floor(t * baud).astype(np.uint64)
# A cheap hash keeps the stream identical across block boundaries. # An avalanche mix, not a plain multiply. Multiplying the symbol
h = (idx * 2654435761) % 4294967296 # index by an odd constant and taking it modulo the number of levels
return (h % levels).astype(np.float64) # returns the low bits of a counter -- 0,1,0,1 for two levels -- so
# what came out was a tone at half the symbol rate rather than data.
# Every FSK feature measured from it was the feature of a square wave.
z = idx * np.uint64(0x9E3779B97F4A7C15)
z ^= z >> np.uint64(30)
z *= np.uint64(0xBF58476D1CE4E5B9)
z ^= z >> np.uint64(27)
z *= np.uint64(0x94D049BB133111EB)
z ^= z >> np.uint64(31)
return (z % np.uint64(levels)).astype(np.float64)
def _shaped_levels(self, t: np.ndarray, fs: float, levels: int,
baud: float) -> np.ndarray:
"""Symbol levels in [-1, 1] after a transmit filter.
Every real FSK transmitter shapes the symbol stream before it reaches
the modulator -- GFSK, C4FM and raised cosine all do the same job, and
an unshaped stream would be far too wide to licence. It matters here
because square-edged NRZ has a spectrum full of odd harmonics of the
symbol rate, and a cyclostationary baud estimator locks onto whichever
of those is strongest: 3600 baud came back as 18000. That is not a
fault in the estimator. It is a signal no transmitter would radiate.
The filter is applied over absolute time with symbols either side of
the block, so the stream stays continuous across block boundaries.
"""
sps = max(2.0, fs / max(1.0, baud))
std = 0.15 * sps # roughly BT = 0.5
half = int(math.ceil(3.0 * std))
n = t.size
te = t[0] + np.arange(-half, n + half) / fs
sym = self._symbols(te, fs, levels, baud)
# Divided by the half-span of the level grid, so the outer symbols
# land on +/- the requested deviation whatever the level count. The
# old form clamped that divisor at 1.0, which quietly halved the
# deviation of every two-level signal the simulator produced.
lv = (sym - (levels - 1) / 2.0) / ((levels - 1) / 2.0)
taps = np.exp(-0.5 * (np.arange(-half, half + 1) / std) ** 2)
taps /= taps.sum()
return np.convolve(lv, taps, mode="same")[half:half + n]
def _morse_envelope(self, t: np.ndarray, fs: float) -> np.ndarray: def _morse_envelope(self, t: np.ndarray, fs: float) -> np.ndarray:
key = self._morse_key(self.message, self.wpm) key = self._morse_key(self.message, self.wpm)
@ -293,6 +331,10 @@ def default_transmitters() -> list[VirtualTransmitter]:
V(446_000_000, "carrier", 0.25, 1_000, "unmodulated carrier"), V(446_000_000, "carrier", 0.25, 1_000, "unmodulated carrier"),
V(433_920_000, "ook", 0.30, 40_000, "ISM remote", baud=2000, V(433_920_000, "ook", 0.30, 40_000, "ISM remote", baud=2000,
period_seconds=9, on_seconds=1.2), period_seconds=9, on_seconds=1.2),
# Always on, because that is the whole character of a control channel
# and the reason it needs recognising rather than recording.
V(856_562_500, "fsk2", 0.42, 12_500, "SMARTNET control channel",
baud=3600, deviation=2_600),
] ]

13
bandsaunter/tui.py
View file

@ -548,7 +548,18 @@ needed; most dongles use the Q branch.
There is no filtering or gain in front of the digitiser in this mode, so an HF There is no filtering or gain in front of the digitiser in this mode, so an HF
antenna and a quiet location matter more than usual."""), antenna and a quiet location matter more than usual."""),
"8": ("Keys during a scan", """ "8": ("Trunked systems", """
Police, fire and large business radio mostly runs on trunked systems, where a
pool of channels is shared and one frequency is given over entirely to a data
stream saying which channel each conversation has been put on. That is the
control channel: loud, perfectly steady, never silent, and with nothing on it
to hear.
It is recognised by its symbol rate and its refusal to pause, named on screen,
and skipped within a second or two. Turn 'Skip trunk control channels' off
only if you are collecting them for a decoder. If digital voice calls are
being skipped by mistake, raise 'Control channel patience'."""),
"9": ("Keys during a scan", """
q stop the scan q stop the scan
p pause and resume p pause and resume
s skip the signal being recorded and carry on sweeping s skip the signal being recorded and carry on sweeping

15
bandsaunter/ui.py
View file

@ -90,6 +90,7 @@ class _RecState:
present: bool = False present: bool = False
quiet_for: float = 0.0 quiet_for: float = 0.0
active: bool = False active: bool = False
note: str = "" # e.g. "TRUNK: Motorola SMARTNET / SmartZone"
class ScanDisplay: class ScanDisplay:
@ -124,6 +125,7 @@ class ScanDisplay:
cb.on_record_start = self.on_record_start cb.on_record_start = self.on_record_start
cb.on_record_tick = self.on_record_tick cb.on_record_tick = self.on_record_tick
cb.on_record_end = self.on_record_end cb.on_record_end = self.on_record_end
cb.on_record_note = self.on_record_note
cb.on_status = self.on_status cb.on_status = self.on_status
cb.on_error = self.on_error cb.on_error = self.on_error
@ -154,6 +156,10 @@ class ScanDisplay:
self._rec.mode = rec.mode self._rec.mode = rec.mode
self._dirty = True self._dirty = True
def on_record_note(self, note: str):
self._rec.note = note
self._dirty = True
def on_record_end(self, hit: HitRecord): def on_record_end(self, hit: HitRecord):
self._rec.active = False self._rec.active = False
if hit.kept: if hit.kept:
@ -236,6 +242,15 @@ class ScanDisplay:
# being held open across a pause rather than stuck. # being held open across a pause rather than stuck.
sq = f"[yellow]gap {min(r.quiet_for, hang):4.1f}/{hang:g}s[/yellow]" sq = f"[yellow]gap {min(r.quiet_for, hang):4.1f}/{hang:g}s[/yellow]"
limit_s = f"/{limit:g}s" if limit else "" limit_s = f"/{limit:g}s" if limit else ""
if r.note:
# A control channel is not being recorded so much as identified
# and abandoned; saying "REC" while that happens is a lie.
return Panel(
Text.from_markup(
f"[bold black on yellow] {r.note} [/bold black on yellow] "
f"{fmt_hz(r.frequency)} [{r.mode}] "
f"SNR {r.snr:5.1f} dB [yellow]skipping[/yellow]"),
border_style="yellow", padding=(0, 1))
return Panel( return Panel(
Text.from_markup( Text.from_markup(
f"[bold red]REC[/bold red] {fmt_hz(r.frequency)} " f"[bold red]REC[/bold red] {fmt_hz(r.frequency)} "

99
packaging/bandsaunter.1
View file

@ -1,5 +1,5 @@
.\" Generated by packaging/make-man.py -- do not edit by hand. .\" Generated by packaging/make-man.py -- do not edit by hand.
.TH BANDSAUNTER 1 "2026-08-22" "bandsaunter 2026-08-22_01" "User Commands" .TH BANDSAUNTER 1 "2026-08-22" "bandsaunter 2026-08-22_02" "User Commands"
.SH NAME .SH NAME
bandsaunter \- scan, record and identify radio signals with an RTL-SDR bandsaunter \- scan, record and identify radio signals with an RTL-SDR
.SH SYNOPSIS .SH SYNOPSIS
@ -293,10 +293,10 @@ Record these \[em] which kinds of content are worth keeping.
.br .br
Setting name \fBaccept\fR, default \fBvoice, cw, digital\fR. Setting name \fBaccept\fR, default \fBvoice, cw, digital\fR.
.br .br
Accepts: one of: voice, cw, digital, carrier, noise. Accepts: one of: voice, cw, digital, trunk, carrier, noise.
.RS .RS
.PP .PP
Which kinds of transmission are worth keeping: speech, Morse, data, plain unmodulated carriers, and noise. Most people want voice, cw and digital. Add carrier if you are hunting beacons or interference sources, and noise only for diagnosing why nothing is being recorded. Which kinds of transmission are worth keeping: speech, Morse, data, trunking control channels, plain unmodulated carriers, and noise. Most people want voice, cw and digital. Add carrier if you are hunting beacons or interference sources, trunk if you are collecting control channels for a decoder, and noise only for diagnosing why nothing is being recorded.
.RE .RE
.TP .TP
.B --min-signal-score .B --min-signal-score
@ -343,6 +343,37 @@ Accepts: at least 0.5.
How long a recording gets to prove it contains something. If nothing recognisable has appeared by then it is abandoned and deleted, and the sweep moves on rather than sitting on an open but empty channel. How long a recording gets to prove it contains something. If nothing recognisable has appeared by then it is abandoned and deleted, and the sweep moves on rather than sitting on an open but empty channel.
.RE .RE
.PP .PP
.SS Trunking
.TP
.B --skip-control / --keep-control --no-skip-control
Skip trunk control channels \[em] spot a trunked system's control channel and move straight on.
.br
Setting name \fBskip_control\fR, default \fByes\fR.
.RS
.PP
Trunked radio systems -- the kind police, fire and large businesses use -- keep one frequency transmitting a data stream around the clock. It tells the radios which channel to jump to for each conversation; it carries no speech and it never goes quiet. To a scanner it looks like a very strong, very interesting signal, so without this setting the receiver parks on it, records the whole record limit of buzzing, and finds it again on the next sweep. With this on it is recognised within a second or two, named on screen, and skipped. Leave it on unless you are deliberately collecting control channel data to feed to a decoder.
.RE
.TP
.B --control-seconds
Control channel patience \[em] unbroken carrier a digital-voice look-alike must hold (s).
.br
Setting name \fBcontrol_seconds\fR, default \fB20 s\fR.
.br
Accepts: at least 1.
.RS
.PP
Only matters for P25, DMR and NXDN. On those systems the control channel and an ordinary digital conversation look identical for the first few seconds, and the only thing that tells them apart is that the conversation eventually pauses and the control channel never does. This is how long a signal has to keep going without a break before it is judged to be a control channel. Raise it if digital voice calls are being skipped by mistake; lower it if you are tired of waiting out control channels. It has no effect on the older Motorola and EDACS systems, which are recognised immediately from their symbol rate.
.RE
.TP
.B --lockout-control / --no-lockout-control
Lock out control channels \[em] add every control channel found to the lock-out list.
.br
Setting name \fBlockout_control\fR, default \fBno\fR.
.RS
.PP
Write each control channel into the lock-out list as it is found, so the scanner stops even looking at it. Skipping one already costs only a second or two, so this is worth turning on mainly if you scan the same band constantly and want the list built for you. With 'Remember lock-outs' also on, the entries are saved to your settings file and survive a restart.
.RE
.PP
.SS Receiver .SS Receiver
.TP .TP
.B -d --device .B -d --device
@ -755,6 +786,14 @@ bandsaunter scan \-r 144M\-148M \-\-lockout "162.55M, 450M\-455M"
A single frequency is widened by A single frequency is widened by
.BR \-\-lockout\-width ; .BR \-\-lockout\-width ;
a span is used exactly as written. a span is used exactly as written.
.PP
Two runs never write anything back.
.B \-\-no\-config
has no settings file to write to, since the point of it is to leave the saved
settings alone; and
.B \-\-simulate
is looking at an invented band, whose frequencies would be nonsense in a real
settings file. Both still lock out for the run in hand, and say so.
.SH KEYS DURING A SCAN .SH KEYS DURING A SCAN
.TP .TP
.B q .B q
@ -784,6 +823,60 @@ every transmission on one frequency is appended to a single growing file for
that frequency, with a spoken date and time before each one, so a scan can be that frequency, with a spoken date and time before each one, so a scan can be
played back as a recording of that channel rather than clicked through as played back as a recording of that channel rather than clicked through as
hundreds of fragments. hundreds of fragments.
.SH TRUNKED SYSTEMS
Police, fire and large business radio in the US mostly runs on
.IR trunked
systems. Instead of giving each department its own frequency, the system owns
a pool of channels and hands one out for each conversation as it happens. To
make that work, one frequency in the pool is given over entirely to a data
stream that runs day and night, telling every radio in the fleet where to go
next. That frequency is the
.IR "control channel" .
.PP
A control channel is the worst thing a scanner can find. It is loud, it is
perfectly steady, it never stops, and there is nothing on it to listen to \[em]
just a harsh buzz. A scanner without special handling parks on it for the
whole record limit, saves the file, and then finds it again on the next sweep,
for as long as it is left running.
.PP
bandsaunter recognises one from the shape of the signal, names the system on
screen, deletes what it captured and moves on, usually within a second or
two. What it looks for is a constant\-envelope data stream that never pauses,
at a symbol rate belonging to a known trunking standard:
.RS
.PP
3600 baud two\-level \[em] Motorola SMARTNET / SmartZone (Type I and II).
.br
9600 baud two\-level \[em] EDACS and ProVoice.
.br
1200 baud two\-level \[em] MPT\-1327.
.br
4800 baud four\-level \[em] P25 or DMR Tier III.
.br
2400 baud four\-level \[em] NXDN and NEXEDGE.
.RE
.PP
The first two are recognised at once: nothing else transmits at those rates
without pausing. The others share their shape with an ordinary digital voice
call on the same system, so they are only called a control channel once the
carrier has run unbroken for
.B \-\-control\-seconds
(20 s by default) \[em] long enough that a real conversation would have taken
a breath. Raise that figure if digital voice calls are being skipped by
mistake.
.PP
Being inside a band where trunking is common raises confidence but is never
required: trunking is licensed on business pairs all over the spectrum.
.PP
Use
.B \-\-keep\-control
to record control channels anyway, which is what you want if you are feeding
them to a decoder. Use
.B \-\-lockout\-control
to have each one written into the lock\-out list as it is found, so the
scanner stops looking at it at all; with
.B \-\-save\-lockouts
on, that list survives a restart.
.SH HF RECEPTION .SH HF RECEPTION
These receivers cannot normally tune below about 24 MHz. Below that they can These receivers cannot normally tune below about 24 MHz. Below that they can
sample the antenna directly instead, which opens up shortwave: broadcast, sample the antenna directly instead, which opens up shortwave: broadcast,

62
packaging/make-man.py
View file

@ -237,6 +237,14 @@ bandsaunter scan \-r 144M\-148M \-\-lockout "162.55M, 450M\-455M"
A single frequency is widened by A single frequency is widened by
.BR \-\-lockout\-width ; .BR \-\-lockout\-width ;
a span is used exactly as written. a span is used exactly as written.
.PP
Two runs never write anything back.
.B \-\-no\-config
has no settings file to write to, since the point of it is to leave the saved
settings alone; and
.B \-\-simulate
is looking at an invented band, whose frequencies would be nonsense in a real
settings file. Both still lock out for the run in hand, and say so.
.SH KEYS DURING A SCAN .SH KEYS DURING A SCAN
.TP .TP
.B q .B q
@ -266,6 +274,60 @@ every transmission on one frequency is appended to a single growing file for
that frequency, with a spoken date and time before each one, so a scan can be that frequency, with a spoken date and time before each one, so a scan can be
played back as a recording of that channel rather than clicked through as played back as a recording of that channel rather than clicked through as
hundreds of fragments. hundreds of fragments.
.SH TRUNKED SYSTEMS
Police, fire and large business radio in the US mostly runs on
.IR trunked
systems. Instead of giving each department its own frequency, the system owns
a pool of channels and hands one out for each conversation as it happens. To
make that work, one frequency in the pool is given over entirely to a data
stream that runs day and night, telling every radio in the fleet where to go
next. That frequency is the
.IR "control channel" .
.PP
A control channel is the worst thing a scanner can find. It is loud, it is
perfectly steady, it never stops, and there is nothing on it to listen to \[em]
just a harsh buzz. A scanner without special handling parks on it for the
whole record limit, saves the file, and then finds it again on the next sweep,
for as long as it is left running.
.PP
bandsaunter recognises one from the shape of the signal, names the system on
screen, deletes what it captured and moves on, usually within a second or
two. What it looks for is a constant\-envelope data stream that never pauses,
at a symbol rate belonging to a known trunking standard:
.RS
.PP
3600 baud two\-level \[em] Motorola SMARTNET / SmartZone (Type I and II).
.br
9600 baud two\-level \[em] EDACS and ProVoice.
.br
1200 baud two\-level \[em] MPT\-1327.
.br
4800 baud four\-level \[em] P25 or DMR Tier III.
.br
2400 baud four\-level \[em] NXDN and NEXEDGE.
.RE
.PP
The first two are recognised at once: nothing else transmits at those rates
without pausing. The others share their shape with an ordinary digital voice
call on the same system, so they are only called a control channel once the
carrier has run unbroken for
.B \-\-control\-seconds
(20 s by default) \[em] long enough that a real conversation would have taken
a breath. Raise that figure if digital voice calls are being skipped by
mistake.
.PP
Being inside a band where trunking is common raises confidence but is never
required: trunking is licensed on business pairs all over the spectrum.
.PP
Use
.B \-\-keep\-control
to record control channels anyway, which is what you want if you are feeding
them to a decoder. Use
.B \-\-lockout\-control
to have each one written into the lock\-out list as it is found, so the
scanner stops looking at it at all; with
.B \-\-save\-lockouts
on, that list survives a restart.
.SH HF RECEPTION .SH HF RECEPTION
These receivers cannot normally tune below about 24 MHz. Below that they can These receivers cannot normally tune below about 24 MHz. Below that they can
sample the antenna directly instead, which opens up shortwave: broadcast, sample the antenna directly instead, which opens up shortwave: broadcast,

303
tests/test_trunk.py Normal file
View file

@ -0,0 +1,303 @@
"""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