diff --git a/README.md b/README.md index fe72635..10f5737 100644 --- a/README.md +++ b/README.md @@ -618,6 +618,94 @@ The band name is written into each recording's sidecar too, so it travels with the capture, and `saunterbrowse` will search on it — typing `/70 cm` finds everything in the band without having to remember 420–450 MHz. +### Reading data signals + +A great deal of what a scanner finds is not speech. Doorbells, tyre-pressure +sensors, weather stations, remote controls, paging, packet radio — all of it +carries something a receiver can read, and bandsaunter reads it: + +``` +21:14:07 433.92 MHz 70 cm Amateur 3.2s SNR 45.8 dB EV1527 / PT2262-style remote (93%) + EV1527 / PT2262-style remote 24 bits 516 baud x12 B2 35 4E +21:14:31 929.6125 MHz UHF / 900 MHz Paging 4.5s SNR 49.5 dB POCSAG 1200 (97%) + [1234568D] ENGINE 4 RESPOND + [0098765A] CALL EXT 4412 +21:15:02 144.39 MHz 2 m Amateur 1.8s SNR 31.2 dB AX.25 / APRS (93%) + W1AW>APRS>WIDE1-1: !4142.45N/07243.63W-Newington CT +``` + +**Whatever the modulation, a data signal is the same shape once it has been +sliced**: a train of alternating runs whose *lengths* carry the information. +On-off keying gives that directly — the carrier is up or it is down — and +two-level FSK gives exactly the same thing from the discriminator, one tone or +the other. So both are reduced to runs, and everything after that is shared. + +What the runs mean is the line code, and it is worked out from the runs alone +rather than configured, because each code makes a different prediction about +which of the two histograms is the bimodal one: + +| Code | Pulses | Gaps | Who uses it | +|---|---|---|---| +| **PWM** | two lengths | constant, or the period is | EV1527, PT2262 and nearly every 433 MHz remote | +| **PPM** | constant | two lengths | the other half of the same market | +| **Manchester** | T and 2T only | T and 2T only | anything whose receiver recovers its own clock | +| **NRZ** | any whole number of symbols | same | what a framed protocol sits on | + +Four-level FSK — C4FM, as P25, DMR and NXDN send it — is recognised as such and +read as symbols. Slicing it down the middle also produces bits, and they mean +nothing; a capture that had been coming back as "10783 bits of NRZ at 5335 +baud" now says *4-level FSK, 5334 baud, no frame sync recognised*, which is +both true and useful. Where a frame sync word does appear, the system is named +outright. + +### Protocols that can be read in full + +Two carry their own framing and checksums, so a frame either passes or it does +not — and one that passes is not a guess: + +**POCSAG** paging, at 512, 1200 or 2400 baud. Nothing in the signal announces +which rate it is, so all three are tried and the one whose 32-bit sync word +turns up is the right one. Every codeword is checked — and a single bit error +corrected — against the BCH code the standard puts there for exactly that. The +address, function letter and message text all come out. + +**AX.25 / APRS** on 1200 baud AFSK. The frame check has to come out right +before a frame is reported at all. The sender's callsign, the digipeater path +and the payload are shown — and the callsign goes onto the map with everyone +else. + +```bash +bandsaunter analyze capture.cf32 --rate 48000 # decode a file you already have +bandsaunter scan --no-decode-data # turn it off +saunterbrowse # decoded packets sit where a transcript would +``` + +### Believing a decode + +This is the hard half. A decoder that always returns *something* is worse than +useless: noise sliced at a threshold produces runs, and runs produce bits. +Three things guard against that. + +- **The runs have to fit.** A decode whose runs do not quantise to the line + code's own grid is thrown away. +- **Most of the capture has to agree.** A data signal is data all the way + through. One lucky window in eight is a coincidence — and that is exactly + what SSB voice produced before this check existed. +- **The packet has to repeat.** Much the strongest of the three. These + transmitters send the same thing three to ten times over, and bits that come + back identical every time did not come from noise. + +A bare reading with none of that behind it — where the run lengths merely +happened to land on a grid — is reported as **nothing at all**, rather than as +a bit string with a low number beside it that somebody will read anyway. Across +27 recordings of speech, music, static, a bare carrier, Morse and PSK, the +decoder returns nothing 27 times. + +And a decode that *does* have repeats or a checksum behind it outranks the +content check. A burst of keying demodulated as FM audio is a buzz, and the +speech detector likes a buzz — but a frame whose own checksum came out right is +not a statistic. + ### Trunked systems and their control channels Police, fire and most large business radio in the US runs on *trunked* diff --git a/bandsaunter/__init__.py b/bandsaunter/__init__.py index 281e407..31e588e 100755 --- a/bandsaunter/__init__.py +++ b/bandsaunter/__init__.py @@ -9,7 +9,7 @@ and transcribing speech. # 2026-08-21_02 is the second build made on the 21st. The revision is padded # to two digits so versions sort as text. VERSION_DATE = "2026-08-28" -VERSION_REVISION = 1 +VERSION_REVISION = 2 __version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}" diff --git a/bandsaunter/browse.py b/bandsaunter/browse.py index 2bc9285..384f019 100644 --- a/bandsaunter/browse.py +++ b/bandsaunter/browse.py @@ -31,6 +31,7 @@ from rich.align import Align from rich.console import Console, Group from rich.layout import Layout from rich.live import Live +from rich.markup import escape from rich.panel import Panel from rich.table import Table from rich.text import Text @@ -85,6 +86,7 @@ class Capture: _transcript: str | None = field(default=None, init=False, repr=False) _duration: float | None = field(default=None, init=False, repr=False) _calls: list | None = field(default=None, init=False, repr=False) + _decoded: list | None = field(default=None, init=False, repr=False) # -- lazily read sidecars --------------------------------------------- @property @@ -139,6 +141,49 @@ class Capture: pass return self._duration + @property + def decoded(self) -> list[str]: + """What a data capture turned out to say, a line at a time. + + The file beside the recording wins over the sidecar for the same + reason the transcript does: it is what the most recent run wrote. + """ + if self._decoded is None: + lines: list[str] = [] + try: + text = self.path.with_name( + self.path.stem + "_data.txt").read_text() + lines = [ln for ln in text.splitlines() + if ln.strip() and not ln.startswith("#")] + except OSError: + saved = self.meta.get("data_messages") or [] + lines = [str(m) for m in saved if str(m).strip()] + for key in ("data_hex", "data_bits"): + value = str(self.meta.get(key, "")).strip() + if value and not saved: + lines.append(value) + self._decoded = lines + return self._decoded + + @property + def data_headline(self) -> str: + """What kind of data it was, never what the data said.""" + protocol = str(self.meta.get("data_protocol", "")).strip() + encoding = str(self.meta.get("data_encoding", "")).strip() + if not (protocol or encoding): + return "decoded data" if self.decoded else "" + bits = [protocol or f"{encoding} data"] + baud = self.meta.get("baud") + if baud: + bits.append(f"{float(baud):.0f} baud") + repeats = int(self.meta.get("data_repeats") or 0) + if repeats > 1: + bits.append(f"x{repeats}") + checks = self.meta.get("data_checks") or [] + if checks: + bits.append(str(checks[0])) + return " ".join(bits) + @property def callsigns(self) -> list[str]: """Callsign-shaped runs in the transcript, found once and kept.""" @@ -499,6 +544,10 @@ class Browser: # of a range, and remembering 420-450 MHz is not the point. if any(q in band.lower() for band in cap.bands): return True + # And in what a data capture said: a pager message is as searchable + # as a spoken one, and the reason for searching is the same. + if any(q in line.lower() for line in cap.decoded): + return True return q in cap.transcript.lower() @property @@ -567,7 +616,10 @@ class Browser: # they get room of their own rather than eating into it: a taller # ceiling when there are any, and never fewer than enough to show them. ceiling = max(6, min(20 if calls else 16, screen // (2 if calls else 3))) - needed = len(self._transcript_lines()) + len(calls) + 4 + cap = self.current + body = len(self._transcript_lines()) or ( + len(cap.decoded) + 2 if cap is not None and cap.decoded else 0) + needed = body + len(calls) + 4 if calls: needed += 1 # the blank line above floor = min(ceiling, len(calls) + 6) @@ -614,6 +666,22 @@ class Browser: height = self._transcript_height() if cap is None: return Panel("", border_style="bright_black", height=height) + if not cap.transcript and cap.decoded: + # A data capture has no words, but it does have content, and it + # belongs in the same place a transcript would be: at the top, + # where the reader is already looking. + body = Text() + headline = cap.data_headline + if headline: + body.append(headline + "\n\n", style="bold cyan") + room = max(1, height - 4 - (2 if headline else 0)) + shown = cap.decoded[:room] + body.append("\n".join(shown), style="bold white") + if len(cap.decoded) > len(shown): + body.append(f"\n… {len(cap.decoded) - len(shown)} more line(s)" + " — press t to read it all", style="yellow") + return Panel(body, title="decoded", title_align="left", + border_style="cyan", padding=(1, 3), height=height) if not cap.transcript: inner = Align.center(Text(self._why_no_transcript(cap), style="bright_black", justify="center"), @@ -696,6 +764,11 @@ class Browser: if cap.meta.get("morse_wpm"): line.append(f" {float(cap.meta['morse_wpm']):.0f} WPM", style="yellow") + checks = cap.meta.get("data_checks") or [] + if checks: + # A checksum that came out right is the strongest thing anyone + # can say about a decode, so it is said on the front line. + line.append(f" {checks[0]} ✓", style="green") second = Text() bands = cap.bands @@ -743,7 +816,9 @@ class Browser: base = "on grey19 " if here else "" cat = CATEGORY_STYLE.get(cap.category, "white") when = cap.when - summary = cap.transcript.replace("\n", " ") or cap.classification + summary = (cap.transcript.replace("\n", " ") + or (cap.decoded[0] if cap.decoded else "") + or cap.classification) t.add_row( Text(mark, style=base + ("bold red" if playing else "bold cyan")), @@ -770,13 +845,17 @@ class Browser: border_style="blue", padding=(0, 1)) def _footer(self) -> Text: + # Both of these carry text that did not come from this program -- what + # the user typed, a filename, a player's error -- and rich reads a + # square bracket as markup. Typing "[/" at the search prompt used to + # end the session with a MarkupError. if self.searching: return Text.from_markup( - f"[bold]search:[/bold] {self.query}[blink]_[/blink]" + f"[bold]search:[/bold] {escape(self.query)}[blink]_[/blink]" " [bright_black]enter to accept, esc to clear" "[/bright_black]") if self.message: - return Text.from_markup(f"[yellow]{self.message}[/yellow]") + return Text.from_markup(f"[yellow]{escape(self.message)}[/yellow]") if self.player.active and self.player.playing is not None: total = self.player.playing.duration done = self.player.elapsed @@ -837,6 +916,13 @@ class Browser: calls = self._callsign_lines(pad=4) room = max(3, screen - 4 - (len(calls) + 1 if calls else 0)) lines = self._transcript_lines(pad=4) + title_word = "transcript" + if not lines and cap is not None and cap.decoded: + # A long paging capture is as worth reading in full as a long + # net, and there is nowhere else to read it. + lines = ([cap.data_headline, ""] if cap.data_headline else []) \ + + list(cap.decoded) + title_word = "decoded" self.read_top = max(0, min(self.read_top, max(0, len(lines) - room))) shown = lines[self.read_top:self.read_top + room] body = Text("\n".join(shown), style="white") @@ -851,7 +937,7 @@ class Browser: body.append("\n" + line) where = (f"{self.read_top + 1}-{self.read_top + len(shown)}" f" of {len(lines)}" if len(lines) > room else "") - title = "transcript" + title = title_word if cap is not None and cap.frequency: title += f" · {fmt_hz(cap.frequency)}" when = cap.when @@ -1062,6 +1148,20 @@ def build_parser() -> argparse.ArgumentParser: return p +def _first_line(cap: Capture) -> str: + """The most informative thing about a capture, in one line.""" + if cap.transcript: + return cap.transcript.splitlines()[0] + if cap.decoded: + # A pager message speaks for itself; a packet of hex does not, so + # that one is introduced by what kind of packet it is. + first = cap.decoded[0] + headline = cap.data_headline + return first if " " in first.strip(" 0123456789ABCDEF") or not headline \ + else f"{headline} {first}" + return cap.classification + + def _write_kml(console: Console, browser: "Browser", book: CallsignBook, where: str) -> int: """Build a map from the transcripts already on disk. @@ -1171,7 +1271,7 @@ def main(argv: list[str] | None = None) -> int: line = (f"{fmt_hz(cap.frequency):>14} {shorten_band(cap.band, 20):<20} " f"{when} " f"{_dur(cap.duration):>7} {cap.category:<8} " - f"{cap.transcript.splitlines()[0] if cap.transcript else cap.classification}") + f"{_first_line(cap)}") console.print(line, highlight=False, soft_wrap=True) return 0 diff --git a/bandsaunter/cli.py b/bandsaunter/cli.py index c263bb2..720b046 100755 --- a/bandsaunter/cli.py +++ b/bandsaunter/cli.py @@ -828,6 +828,24 @@ def cmd_analyze(args) -> int: console.print(Panel(Text.from_markup("\n".join(body)), title="identification", border_style="green")) + # Whatever it is, try to read it: the whole point of pointing this at a + # file is to find out what is in it. + from .decode import decode_data + got = decode_data(iq, rate, family=cls.family, + baud_hint=cls.features.baud if cls.features else 0.0) + if got.ok: + # Printed as plain text, not markup: a decoded packet is arbitrary + # bytes from the air, and square brackets in it are common. + lines = got.report() + body = Text(lines[0], style="bold") + for line in lines[1:]: + body.append("\n" + line) + body.append(f"\n{got.confidence * 100:.0f}% confident", + style="not bold grey62") + console.print(Panel(body, title="decoded data", border_style="cyan")) + elif cls.family in ("ook", "fsk", "psk", "digital", "control"): + console.print(f"[yellow]nothing decoded: {got.note}[/yellow]") + f = cls.features if f: t = Table(box=None, header_style="bold") diff --git a/bandsaunter/config.py b/bandsaunter/config.py index 7ccf5d7..ba0cf66 100755 --- a/bandsaunter/config.py +++ b/bandsaunter/config.py @@ -108,6 +108,7 @@ class ScanConfig: iq_format: str = "cf32" # cf32 or cs16 classify: bool = True decode_morse: bool = True + decode_data: bool = True # read packets out of data signals # -- one file per frequency ------------------------------------------ combine_by_frequency: bool = False diff --git a/bandsaunter/decode.py b/bandsaunter/decode.py new file mode 100644 index 0000000..cb77a18 --- /dev/null +++ b/bandsaunter/decode.py @@ -0,0 +1,1226 @@ +"""Turning a data signal into bits, and the bits into something readable. + +Almost every data signal a scanner meets, whatever its modulation, comes down +to the same shape once it has been sliced: a train of alternating runs whose +*lengths* carry the information. On-off keying gives that directly -- the +carrier is up or it is down. Two-level FSK gives exactly the same thing from +the discriminator, one tone or the other. So both are reduced to a run-length +train here and everything after that is shared. + +What the runs mean is the line code, and there are only a handful in common +use: + +``PWM`` + The gap is constant and the pulse is one of two lengths. Nearly every + cheap 433 MHz remote, doorbell, tyre sensor and weather station. +``PPM`` + The pulse is constant and the *gap* is one of two lengths. The other + half of the same market. +``Manchester`` + Every bit is a transition in the middle of its period, so runs come in + only two lengths, T and 2T. Used where a receiver has to recover the + clock from the data. +``NRZ`` + The level is simply held for as many symbol periods as there are bits. + What a framed protocol like POCSAG sits on top of. + +Which one it is can be worked out from the runs alone, without being told, +because each makes a different prediction about which of the two histograms +-- pulses or gaps -- is the bimodal one. + +The last piece is knowing whether the answer is real. A decoder that always +returns *something* is worse than useless: noise sliced at a threshold +produces runs, and runs produce bits. Two things guard against that here. +The first is fit: the runs have to quantise to the line code's own grid, and +a decode whose runs are scattered is thrown away. The second, and much the +stronger, is repetition -- cheap transmitters send the same packet three to +ten times in a row, and bits that come back identical across several bursts +did not come from noise. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +import numpy as np + +__all__ = ["decode_data", "DataDecode", "Burst", "PulseTrain", "Encoding", + "slice_ook", "slice_fsk", "split_bursts", "decode_train", + "decode_four_level", "level_count", "nrz_bits", "bits_to_hex", + "check_crc", "ENCODINGS", "FRAME_SYNCS"] + +ENCODINGS: tuple[str, ...] = ("PWM", "PPM", "Manchester", "NRZ") + +# A run shorter than this is a slicing artefact rather than a symbol. At the +# 40-50 kHz an OOK capture is usually decimated to, four samples is around +# 100 us, well under the shortest pulse any of these devices sends. +MIN_RUN_SAMPLES = 4 + +# Bursts are separated by a gap this many times the median run. Cheap remotes +# leave a long silence between repeats -- typically ten to thirty symbol +# periods -- which is exactly the packet boundary we want. +BURST_GAP_FACTOR = 8.0 + +# Fewer than this and there is nothing to be confident about: two clusters +# fitted to six runs will always look convincing. +MIN_RUNS = 12 + +# A decode has to be at least this good a fit to the line code's grid before +# it is reported at all. +MIN_FIT = 0.55 + +# NRZ fits anything the others fit, because a run of one length and a run of +# three are also a grid of one -- so a PWM packet read as NRZ comes back as +# four times as many meaningless bits. The pulse-length codes and Manchester +# each make a much stronger claim about the data, so NRZ has to beat them +# clearly rather than by a rounding error to be preferred. +SPECIFICITY = {"PWM": 1.0, "PPM": 1.0, "Manchester": 1.0, "NRZ": 0.85} + +# A bare NRZ read -- no repeats, no checksum, no protocol -- is the weakest +# claim this module can make: it says only that the run lengths happened to +# land on a grid. Voice sliced as on-off keying will produce one such window +# in eight, so it has to be a much better fit than anything with framing +# behind it before it is reported at all. +MIN_NRZ_FIT = 0.80 +MIN_NRZ_BITS = 32 + +# What share of the bursts in a capture have to decode the same way. A data +# signal is data all the way through; one lucky window among eight is a +# coincidence, and coincidences are what a decoder has to refuse. +MIN_BURST_SHARE = 0.5 + + +# --------------------------------------------------------------------------- +# Slicing a signal into runs +# --------------------------------------------------------------------------- + +@dataclass +class PulseTrain: + """Alternating runs of a two-level signal. + + ``levels[i]`` is the state of run ``i`` and ``lengths[i]`` how many + samples it lasted, so the pair is a complete lossless description of a + sliced signal and every decoder below works on nothing else. + """ + + levels: np.ndarray # bool + lengths: np.ndarray # int samples + rate: float # samples per second + source: str = "ook" # how it was sliced: ook or fsk + contrast_db: float = 0.0 # separation of the two levels + start: int = 0 # sample offset into the original signal + + def __len__(self) -> int: + return int(self.lengths.size) + + @property + def seconds(self) -> np.ndarray: + return self.lengths / float(self.rate) + + @property + def duration(self) -> float: + return float(self.lengths.sum()) / float(self.rate) + + def pulses(self) -> np.ndarray: + """Lengths of the high runs, in samples.""" + return self.lengths[self.levels] + + def gaps(self) -> np.ndarray: + return self.lengths[~self.levels] + + def slice(self, i: int, j: int) -> "PulseTrain": + return PulseTrain(self.levels[i:j], self.lengths[i:j], self.rate, + self.source, self.contrast_db, + self.start + int(self.lengths[:i].sum())) + + +def _otsu(values: np.ndarray, bins: int = 128) -> float: + """Otsu threshold: the level that best separates two populations.""" + hist, edges = np.histogram(values, bins=bins) + hist = hist.astype(np.float64) + total = hist.sum() + if total <= 0: + return float(np.median(values)) + centres = 0.5 * (edges[1:] + edges[:-1]) + w0 = np.cumsum(hist) + w1 = total - w0 + csum = np.cumsum(hist * centres) + mu0 = csum / np.maximum(w0, 1e-12) + mu1 = (csum[-1] - csum) / np.maximum(w1, 1e-12) + between = w0 * w1 * (mu0 - mu1) ** 2 + between[~np.isfinite(between)] = 0.0 + peak = float(between.max()) + if peak <= 0: + return float(np.median(values)) + # The middle of the plateau, not its first bin. Two populations with + # nothing at all between them -- silence and full carrier, which is what + # on-off keying is -- make every threshold in the gap equally good, and + # taking the first put the threshold hard against the lower population + # where a hysteresis band around it falls outside the data entirely. + best = np.flatnonzero(between >= peak * (1.0 - 1e-9)) + return float(centres[best].mean()) + + +def _hysteresis(values: np.ndarray, low: float, high: float) -> np.ndarray: + """Slice with two thresholds, so a noisy edge makes one transition. + + A single threshold turns every wobble across it into a pair of runs one + sample long, which destroys the run-length histogram the decoders depend + on. Written as a scan over the crossings rather than sample by sample, + which for a few hundred thousand samples is the difference between + instant and unusable. + """ + above = values >= high + below = values <= low + state = np.where(above, 1, np.where(below, 0, -1)).astype(np.int8) + known = state >= 0 + if not known.any(): + return values >= 0.5 * (low + high) + # Carry each undecided sample forward from the last decided one. + idx = np.where(known, np.arange(state.size), 0) + np.maximum.accumulate(idx, out=idx) + out = state[idx] == 1 + # Anything before the first decision takes that first decision's value. + first = int(np.argmax(known)) + out[:first] = out[first] + return out + + +def _runs_of(mask: np.ndarray) -> tuple[np.ndarray, np.ndarray]: + """Split a boolean array into alternating runs.""" + if mask.size == 0: + return np.zeros(0, bool), np.zeros(0, np.int64) + edges = np.flatnonzero(np.diff(mask.astype(np.int8))) + 1 + starts = np.concatenate(([0], edges)) + ends = np.concatenate((edges, [mask.size])) + return mask[starts], (ends - starts).astype(np.int64) + + +def _despeckle(levels: np.ndarray, lengths: np.ndarray, + minimum: int) -> tuple[np.ndarray, np.ndarray]: + """Absorb runs too short to be symbols into their neighbours. + + Done after slicing rather than by filtering the signal, because a filter + wide enough to remove a one-sample glitch also rounds the edges of the + shortest real pulse and biases every length in the histogram. + """ + if lengths.size == 0: + return levels, lengths + keep_levels: list[bool] = [] + keep_lengths: list[int] = [] + for level, length in zip(levels.tolist(), lengths.tolist()): + if length < minimum and keep_lengths: + # Too short to be real: give its samples to the run before it, + # which merges the following run of the same level into it too. + keep_lengths[-1] += length + continue + if keep_levels and keep_levels[-1] == level: + keep_lengths[-1] += length + continue + keep_levels.append(level) + keep_lengths.append(length) + return (np.array(keep_levels, dtype=bool), + np.array(keep_lengths, dtype=np.int64)) + + +def slice_ook(x: np.ndarray, rate: float, + min_run: int = MIN_RUN_SAMPLES) -> PulseTrain | None: + """Slice an on-off keyed signal into runs of carrier and silence.""" + env = np.abs(np.asarray(x)).astype(np.float64) + if env.size < 64: + return None + # In the log domain, because on/off contrast is tens of dB and a linear + # threshold placed between them sits almost on the noise. Floored 80 dB + # below the peak: a real receiver never sees true silence, and one sample + # that happens to be exactly zero would otherwise stretch the range by + # two hundred decibels and drag every threshold with it. + floor = max(float(env.max()) * 1e-4, 1e-12) + db = 20.0 * np.log10(np.maximum(env, floor)) + lo, hi = np.percentile(db, (5.0, 95.0)) + if hi - lo < 6.0: + return None # nothing is being keyed + cut = _otsu(db) + margin = 0.15 * (hi - lo) + mask = _hysteresis(db, cut - margin, cut + margin) + levels, lengths = _despeckle(*_runs_of(mask), minimum=min_run) + if lengths.size < 3: + return None + on = db[mask] if mask.any() else np.array([hi]) + off = db[~mask] if (~mask).any() else np.array([lo]) + return PulseTrain(levels, lengths, float(rate), "ook", + float(np.median(on) - np.median(off))) + + +def slice_fsk(x: np.ndarray, rate: float, min_run: int = MIN_RUN_SAMPLES, + baud_hint: float = 0.0) -> PulseTrain | None: + """Slice two-level FSK into runs of one tone and the other. + + The same run-length train as OOK, so everything downstream is shared: a + 2-FSK signal is an on-off keyed signal whose "off" happens to be another + frequency rather than silence. + + The discriminator is where this gets hard. Differentiating the phase + amplifies noise, and a single sample landing the wrong side of the + threshold splits one symbol into three runs and ruins the histogram every + decoder below depends on. Two defences: a short median filter, which + removes an impulse without moving an edge the way an average would, and a + threshold placed by Otsu between the two tones rather than at the middle + of a range that noise has widened. + """ + x = np.asarray(x) + if x.size < 64: + return None + if np.iscomplexobj(x): + phase = np.unwrap(np.angle(x)) + freq = np.diff(phase) * (rate / (2.0 * math.pi)) + else: + freq = x.astype(np.float64) + if freq.size < 64: + return None + freq = _median3(freq) + # A quarter of a symbol: long enough to average the noise down, short + # enough that the shortest symbol still reaches full amplitude. + if baud_hint > 0: + win = int(max(1, min(64.0, rate / baud_hint / 4.0))) + else: + win = 3 + if win > 1: + freq = np.convolve(freq, np.ones(win) / win, mode="same") + + cut = _otsu(freq) + high = freq > cut + if not high.any() or high.all(): + return None + top = float(np.median(freq[high])) + bottom = float(np.median(freq[~high])) + spread = top - bottom + if spread <= 0: + return None + margin = 0.20 * spread + mask = _hysteresis(freq, cut - margin, cut + margin) + levels, lengths = _despeckle(*_runs_of(mask), minimum=min_run) + if lengths.size < 3: + return None + return PulseTrain(levels, lengths, float(rate), "fsk", float(spread)) + + +def _frequency_of(x: np.ndarray, rate: float, + baud_hint: float = 0.0) -> np.ndarray | None: + """The discriminator output, cleaned up enough to slice.""" + x = np.asarray(x) + if x.size < 64: + return None + if np.iscomplexobj(x): + freq = np.diff(np.unwrap(np.angle(x))) * (rate / (2.0 * math.pi)) + else: + freq = x.astype(np.float64) + if freq.size < 64: + return None + freq = _median3(freq) + if baud_hint > 0: + win = int(max(1, min(64.0, rate / baud_hint / 4.0))) + else: + win = 3 + return np.convolve(freq, np.ones(win) / win, mode="same") if win > 1 \ + else freq + + +def _steady(freq: np.ndarray, keep: float = 0.45) -> np.ndarray: + """The samples where the frequency was not on its way somewhere. + + A symbol transition sweeps through every value between the two levels, so + a histogram that includes transitions has its valleys filled in and its + peaks flattened. Keeping only the flattest samples -- those where the + frequency is barely changing -- leaves the levels standing clear. + """ + if freq.size < 4: + return freq + slope = np.abs(np.diff(freq, prepend=freq[0])) + return freq[slope <= np.percentile(slope, keep * 100.0)] + + +def level_count(freq: np.ndarray, limit: int = 6) -> int: + """How many discrete levels a discriminator output settles on. + + Slicing a four-level signal down the middle produces runs, and runs + produce bits, and the bits are nonsense -- a C4FM transmission came back + as ten thousand bits of "NRZ data" at a symbol rate that was not its own. + Counting the levels first is what stops that being reported as a decode. + """ + values = _steady(freq) + if values.size < 64: + return 0 + lo, hi = np.percentile(values, (1.0, 99.0)) + if hi <= lo: + return 1 + bins = 48 + hist, _ = np.histogram(values, bins=bins, range=(lo, hi)) + smooth = np.convolve(hist.astype(np.float64), np.ones(3) / 3.0, "same") + peak = smooth.max() + if peak <= 0: + return 1 + modes, i = 0, 0 + while i < smooth.size: + rising = i == 0 or smooth[i] >= smooth[i - 1] + falling = i == smooth.size - 1 or smooth[i] > smooth[i + 1] + if smooth[i] >= 0.25 * peak and rising and falling: + modes += 1 + i += max(2, bins // 12) # one peak, not the top of a plateau + else: + i += 1 + return min(limit, max(1, modes)) + + +def _median3(values: np.ndarray) -> np.ndarray: + """A three-point median filter: removes an impulse, keeps every edge.""" + if values.size < 3: + return values + a, b, c = values[:-2], values[1:-1], values[2:] + middle = np.maximum(np.minimum(a, b), np.minimum(np.maximum(a, b), c)) + out = values.copy() + out[1:-1] = middle + return out + + +def split_bursts(train: PulseTrain, + gap_factor: float = BURST_GAP_FACTOR) -> list[PulseTrain]: + """Cut a train at the long silences that separate packets. + + The repeats a cheap transmitter sends are the single most useful thing + about it -- the same bits arriving several times over is proof the decode + is real -- and this is what finds them. + """ + if len(train) < 3: + return [train] + if train.source != "ook": + # Only silence separates packets, and a frequency-sliced train has + # none: its "low" runs are the other tone, which is data. Cutting a + # continuous FSK stream wherever it held one tone for a while threw + # away two thirds of a thousand-bit frame. + return [train] + gaps = train.gaps() + if gaps.size == 0: + return [train] + typical = float(np.median(train.lengths)) + threshold = max(typical * gap_factor, 1.0) + out: list[PulseTrain] = [] + start = 0 + for i in range(len(train)): + if not train.levels[i] and train.lengths[i] > threshold and i > start: + out.append(train.slice(start, i)) + start = i + 1 + tail = train.slice(start, len(train)) + if len(tail): + out.append(tail) + out = [_trim(b) for b in out] + return [b for b in out if len(b) >= 3] or [train] + + +def _trim(train: PulseTrain) -> PulseTrain: + """Drop the silence at either end of a burst. + + What is left starts and ends on a pulse, which is the shape a packet + actually has: the leading gap is the tail of the previous silence and + the trailing one is the start of the next, and neither is a symbol. + """ + i, j = 0, len(train) + while i < j and not train.levels[i]: + i += 1 + while j > i and not train.levels[j - 1]: + j -= 1 + return train.slice(i, j) if (i or j != len(train)) else train + + +# --------------------------------------------------------------------------- +# Working out the line code +# --------------------------------------------------------------------------- + +@dataclass +class Encoding: + """Which line code the runs are in, and its timing.""" + + name: str = "" + short: float = 0.0 # seconds + long: float = 0.0 # seconds + fit: float = 0.0 # 0..1, how well the runs quantise + + @property + def unit(self) -> float: + return self.short or self.long + + @property + def baud(self) -> float: + if self.name in ("PWM", "PPM"): + # One bit per pulse-and-gap pair, whose average length is the + # constant part plus the mean of the two variable ones. + period = self.short + self.long + return 1.0 / period if period > 0 else 0.0 + if self.unit <= 0: + return 0.0 + # Manchester spends two cells on every bit, so the rate on the air is + # twice the rate of the data. Reporting the cell rate would have a + # 1200 baud link come back as 2400. + return 1.0 / (2.0 * self.unit) if self.name == "Manchester" \ + else 1.0 / self.unit + + +def _two_means(values: np.ndarray, rounds: int = 24 + ) -> tuple[float, float, np.ndarray]: + """Split values into a low and a high cluster. + + A plain two-cluster k-means: the run lengths of a line code really are + two tight groups, and anything more elaborate would only be fitting the + noise between them. + """ + v = np.asarray(values, dtype=np.float64) + lo, hi = float(v.min()), float(v.max()) + if hi <= lo: + return lo, hi, np.zeros(v.size, dtype=bool) + a, b = lo, hi + high = v > 0.5 * (a + b) + for _ in range(rounds): + if high.all() or not high.any(): + break + a = float(v[~high].mean()) + b = float(v[high].mean()) + moved = v > 0.5 * (a + b) + if np.array_equal(moved, high): + break + high = moved + if high.all() or not high.any(): + return float(v.mean()), float(v.mean()), high + return float(v[~high].mean()), float(v[high].mean()), high + + +def _tightness(values: np.ndarray, centre: float) -> float: + """How closely a group sits to its own mean. 1 is perfect.""" + if values.size == 0 or centre <= 0: + return 0.0 + spread = float(np.abs(values - centre).mean()) / centre + return max(0.0, 1.0 - spread * 3.0) + + +def _bimodal(values: np.ndarray) -> tuple[float, float, np.ndarray, float]: + """Fit two clusters and score how convincingly separated they are.""" + lo, hi, high = _two_means(values) + if hi <= 0 or lo <= 0 or high.all() or not high.any(): + return lo, hi, high, 0.0 + separation = (hi - lo) / (hi + lo) + if separation < 0.12: + return lo, hi, high, 0.0 + tight = 0.5 * (_tightness(values[~high], lo) + _tightness(values[high], hi)) + # Both groups have to be populated: one outlier against forty is a stuck + # sample, not a symbol. + balance = min(float(high.mean()), float((~high).mean())) / 0.5 + return lo, hi, high, float(min(1.0, separation * 2.0) * tight + * min(1.0, balance * 4.0)) + + +def _uniform(values: np.ndarray) -> float: + """How nearly constant a set of run lengths is. 1 is perfect.""" + if values.size == 0: + return 0.0 + centre = float(np.median(values)) + return _tightness(values, centre) + + +def _symbol_counts(lengths: np.ndarray, unit: float, + limit: int = 4096) -> np.ndarray: + """How many symbols each run lasts, counted along the whole train. + + Each boundary is placed on the symbol grid by its position from the + start, not by rounding its own run in isolation. Rounding runs one at a + time makes every one of them wrong by up to half a symbol independently, + and those errors accumulate: a clock a quarter of a percent out lost six + bits in six hundred. Measuring against the start cannot drift, because + every boundary is referred to the same origin. + """ + if unit <= 0 or lengths.size == 0: + return np.zeros(lengths.size, dtype=np.int64) + edges = np.cumsum(lengths.astype(np.float64)) + marks = np.round(edges / unit).astype(np.int64) + counts = np.diff(np.concatenate(([0], marks))) + return np.clip(counts, 1, limit) + + +def _quantised_fit(values: np.ndarray, unit: float, + limit: int = 8) -> tuple[float, np.ndarray]: + """How well runs sit on a grid of ``unit``, and their multiples.""" + if unit <= 0 or values.size == 0: + return 0.0, np.zeros(values.size, dtype=np.int64) + counts = np.clip(np.round(values / unit), 1, limit).astype(np.int64) + error = np.abs(values - counts * unit) / unit + return float(max(0.0, 1.0 - 2.0 * error.mean())), counts + + +def _pairs(train: PulseTrain) -> tuple[np.ndarray, np.ndarray]: + """Pulse and gap of each bit, for the pulse-length codes. + + A burst usually starts with a pulse and ends with one, so the last pulse + has no gap after it; it is dropped rather than paired with a silence that + is really the gap between repeats. + """ + levels, lengths = train.levels, train.lengths + start = 0 if (levels.size and levels[0]) else 1 + pulses, gaps = [], [] + i = start + while i + 1 < levels.size: + if levels[i] and not levels[i + 1]: + pulses.append(lengths[i]) + gaps.append(lengths[i + 1]) + i += 2 + return (np.array(pulses, dtype=np.float64), + np.array(gaps, dtype=np.float64)) + + +def _fit_pulse_length(train: PulseTrain) -> tuple[Encoding, str]: + """PWM and PPM: the two codes where a bit is one pulse-and-gap pair.""" + pulses, gaps = _pairs(train) + if pulses.size < MIN_RUNS // 2: + return Encoding(), "" + rate = train.rate + # What has to hold still for the code to be readable is either the other + # half of the pair or the whole bit period. Both shapes are sold: a + # PT2262 or EV1527 keeps the period constant and swaps the pulse and gap + # around inside it, while other remotes hold the gap constant and vary + # only the pulse. + periods = pulses + gaps + steady_period = _uniform(periods) + best = (Encoding(), "") + # Pulse width is read from every pulse in the burst, gap length from every + # gap. A burst ends on a pulse, so a pulse-width code yields one more bit + # than a gap-length one -- the last bit's gap ran into the silence before + # the next repeat and is no longer separable from it. + for name, varying, constant in (("PWM", train.pulses().astype(np.float64), + gaps), + ("PPM", train.gaps().astype(np.float64), + pulses)): + if varying.size < MIN_RUNS // 2: + continue + lo, hi, high, quality = _bimodal(varying) + if quality <= 0: + continue + steady = max(_uniform(constant), steady_period) + fit = 0.65 * quality + 0.35 * steady + if name == "PPM" and steady_period > 0.8: + # A constant period with both halves varying satisfies both + # readings, and they differ only in which way round the bits + # come out. Pulse width is how these parts are documented and + # how everyone else names them. + fit *= 0.9 + if fit <= best[0].fit: + continue + # The long form is a 1. Every one of these devices is written that + # way, and where a decode comes out inverted the checksum says so. + bits = "".join("1" if h else "0" for h in high.tolist()) + best = (Encoding(name, lo / rate, hi / rate, fit), bits) + return best + + +def _levels_at_unit(train: PulseTrain, unit_samples: float) -> np.ndarray: + """Re-sample the train onto its own symbol grid.""" + return np.repeat(train.levels, + _symbol_counts(train.lengths, unit_samples, limit=64)) + + +def _manchester(cells: np.ndarray) -> tuple[str, float]: + """Decode half-bit cells, trying both phases and both polarities. + + A Manchester stream carries no marker for where a bit begins, and the two + conventions disagree about which transition is a one, so all four + readings are tried and the one with fewest illegal cell pairs wins. + """ + best_bits, best_score = "", 0.0 + for phase in (0, 1): + cut = cells[phase:] + cut = cut[:cut.size - (cut.size % 2)] + if cut.size < 8: + continue + first, second = cut[0::2], cut[1::2] + legal = first != second + share = float(legal.mean()) + if share <= best_score: + continue + # 10 is a one under IEEE 802.3; the opposite convention shows up as a + # wholly inverted decode, which a checksum will catch. + bits = "".join("1" if a and not b else "0" + for a, b in zip(first.tolist(), second.tolist())) + best_bits, best_score = bits, share + return best_bits, best_score + + +def _fit_level_code(train: PulseTrain) -> tuple[Encoding, str]: + """Manchester and NRZ: the codes where a run is a whole number of bits.""" + lengths = train.lengths.astype(np.float64) + if lengths.size < MIN_RUNS: + return Encoding(), "" + rate = train.rate + # The unit is the shortest run that is common rather than the shortest + # that occurs: one clipped edge would otherwise set the whole grid. + unit = float(np.percentile(lengths, 10.0)) + if unit <= 0: + return Encoding(), "" + for _ in range(8): + # Each run rounded on its own, which is what makes this converge: + # every run votes independently for how many symbols it holds, and a + # unit that is a few per cent out still rounds them all correctly, so + # the next estimate is right. Counting along the cumulative grid + # here instead would be a fixed point -- a unit two per cent small + # produces two per cent more symbols and reproduces itself exactly. + # Nor may the count be clipped: capping a ten-symbol run at eight + # makes the clock read one per cent slow, which is five bits of drift + # across a six-hundred-bit frame. + _, counts = _quantised_fit(lengths, unit, limit=64) + refined = float(lengths.sum() / max(1.0, counts.sum())) + if refined <= 0 or abs(refined - unit) / unit < 1e-5: + break + unit = refined + fit, _ = _quantised_fit(lengths, unit, limit=64) + if fit < MIN_FIT: + return Encoding(), "" + counts = _symbol_counts(lengths, unit, limit=64) + + best = (Encoding(), "") + # Manchester first: its runs are only ever one or two units, which is a + # much stronger claim than NRZ's "any whole number". + if float((counts <= 2).mean()) > 0.95: + bits, share = _manchester(_levels_at_unit(train, unit)) + if bits and share > 0.9: + best = (Encoding("Manchester", unit / rate, 2 * unit / rate, + float(min(1.0, fit * share))), bits) + if not best[1]: + bits = "".join(("1" if level else "0") * int(count) + for level, count in zip(train.levels.tolist(), + counts.tolist())) + best = (Encoding("NRZ", unit / rate, unit / rate, fit), bits) + return best + + +def nrz_bits(train: PulseTrain, baud: float) -> str: + """Read a train as plain NRZ at a known symbol rate. + + For the framed protocols, which are told their own baud rate by their + specification and so have no clock to recover -- only a phase, and a run + length rounded to the nearest whole number of symbols supplies that. + """ + if baud <= 0 or len(train) == 0: + return "" + unit = train.rate / float(baud) + if unit < 1.5: + return "" # fewer than two samples a symbol + counts = _symbol_counts(train.lengths, unit) + return "".join(("1" if level else "0") * int(count) + for level, count in zip(train.levels.tolist(), + counts.tolist())) + + +# --------------------------------------------------------------------------- +# Four-level FSK +# --------------------------------------------------------------------------- + +# C4FM and its relatives put two bits on every symbol. The mapping is the +# same one P25, DMR and NXDN all use: the outer deviations are 01 and 11, the +# inner ones 00 and 10, from the top down. +C4FM_DIBITS = ("01", "00", "10", "11") + +# Frame synchronisation words, as bits. Finding one of these is decisive: +# forty-eight bits do not line up by accident, and each names the system that +# sent them without any of the rest of the frame having to be understood. +FRAME_SYNCS: tuple[tuple[str, str], ...] = ( + ("P25 Phase 1", "5575F5FF77FF"), + ("DMR (base station, voice)", "755FD7DF75F7"), + ("DMR (base station, data)", "DFF57D75DF5D"), + ("DMR (mobile, voice)", "7F7D5DD57DFD"), + ("DMR (mobile, data)", "D5D7F77FD757"), + ("DMR (direct mode, voice)", "5D577F7757FF"), +) + + +def _sync_bits(hexword: str) -> str: + return "".join(f"{int(c, 16):04b}" for c in hexword) + + +def _estimate_unit(lengths: np.ndarray) -> float: + """The symbol period a set of run lengths is built from, in samples.""" + if lengths.size == 0: + return 0.0 + unit = float(np.percentile(lengths.astype(np.float64), 10.0)) + if unit <= 0: + return 0.0 + for _ in range(8): + _, counts = _quantised_fit(lengths.astype(np.float64), unit, limit=64) + refined = float(lengths.sum() / max(1.0, counts.sum())) + if refined <= 0 or abs(refined - unit) / unit < 1e-5: + break + unit = refined + return unit + + +def _four_means(values: np.ndarray, rounds: int = 30) -> np.ndarray: + """Sort samples into four levels, returning which level each is in.""" + centres = np.percentile(values, (12.5, 37.5, 62.5, 87.5)) + which = np.zeros(values.size, dtype=np.int64) + for _ in range(rounds): + moved = np.argmin(np.abs(values[:, None] - centres[None, :]), axis=1) + if np.array_equal(moved, which): + break + which = moved + for k in range(4): + picked = values[which == k] + if picked.size: + centres[k] = float(picked.mean()) + centres.sort() + return which + + +def decode_four_level(x: np.ndarray, sample_rate: float, + baud_hint: float = 0.0) -> DataDecode | None: + """Read a four-level FSK signal as a stream of dibits. + + Not a full decode of P25 or DMR -- those go on through error coding and a + voice codec, which is a different undertaking -- but the symbols are real, + the symbol rate is real, and where a frame sync word turns up the system + can be named outright. + """ + x = np.asarray(x) + if np.iscomplexobj(x) and x.size >= 64: + # A C4FM carrier is on the whole time. Without this an on-off keyed + # burst reaches here too, and the discriminator noise in its silences + # counts as extra levels -- which would refuse a decode that the OOK + # path gets right. + env = 20.0 * np.log10(np.maximum(np.abs(x), 1e-12)) + low, high = np.percentile(env, (5.0, 95.0)) + if high - low > 10.0: + return None + freq = _frequency_of(x, sample_rate, baud_hint) + if freq is None or level_count(freq) != 4: + return None + cut = _otsu(freq) + levels, lengths = _despeckle(*_runs_of(freq > cut), + minimum=MIN_RUN_SAMPLES) + if lengths.size < MIN_RUNS: + return None + # Transitions land on symbol boundaries whatever the level, so the runs + # of a two-level slice still measure the symbol period even though their + # levels say nothing useful. + unit = _estimate_unit(lengths) + if unit < 2.0: + return None + edges = np.cumsum(lengths) + origin = float(edges[0]) if edges.size else 0.0 + first = -int(origin // unit) + last = int((freq.size - origin) // unit) + if last - first < 24: + return None + centres = origin + (np.arange(first, last) + 0.5) * unit + keep = (centres >= 0) & (centres < freq.size) + samples = freq[centres[keep].astype(np.int64)] + if samples.size < 24: + return None + + which = _four_means(samples) + # Highest deviation first, which is how the mapping above is written. + bits = "".join(C4FM_DIBITS[3 - int(k)] for k in which.tolist()) + baud = sample_rate / unit + for name, word in FRAME_SYNCS: + pattern = _sync_bits(word) + for stream, inverted in ((bits, False), (_flip(bits), True)): + found = stream.count(pattern) + if found: + note = "inverted" if inverted else "" + return DataDecode( + ok=True, protocol=name, encoding="4-level FSK", + baud=baud, bits=stream[:512], repeats=found, + agreement=1.0, checks=[f"frame sync x{found}"], + messages=[f"{name}: {found} frame sync word(s), " + f"{len(stream)} dibits at {baud:.0f} baud" + + (f" ({note})" if note else "")], + confidence=min(0.95, 0.78 + 0.04 * found)) + return DataDecode( + ok=True, encoding="4-level FSK", baud=baud, bits=bits[:512], + repeats=1, agreement=0.0, + note="four-level symbols, no frame sync recognised", + messages=[f"{len(bits)} dibits at {baud:.0f} baud, " + f"no frame sync recognised"], + confidence=0.45) + + +def _flip(bits: str) -> str: + return bits.translate(str.maketrans("01", "10")) + + +def decode_train(train: PulseTrain) -> tuple[Encoding, str]: + """Work out the line code of one burst and read its bits.""" + if len(train) < MIN_RUNS // 2: + return Encoding(), "" + candidates = [_fit_pulse_length(train), _fit_level_code(train)] + candidates = [c for c in candidates if c[1] and c[0].fit >= MIN_FIT] + if not candidates: + return Encoding(), "" + return max(candidates, key=lambda c: c[0].fit * SPECIFICITY[c[0].name]) + + +# --------------------------------------------------------------------------- +# Bits into something a person can read +# --------------------------------------------------------------------------- + +def bits_to_hex(bits: str) -> str: + """Bits as hex, MSB first, with a trailing partial byte kept.""" + if not bits: + return "" + out = [] + for i in range(0, len(bits), 8): + chunk = bits[i:i + 8] + out.append(f"{int(chunk, 2) << (8 - len(chunk)) if len(chunk) < 8 else int(chunk, 2):02X}") + return " ".join(out) + + +def _bytes_of(bits: str) -> bytes: + whole = len(bits) - len(bits) % 8 + return bytes(int(bits[i:i + 8], 2) for i in range(0, whole, 8)) + + +def _crc8(data: bytes, poly: int, init: int = 0x00) -> int: + crc = init + for byte in data: + crc ^= byte + for _ in range(8): + crc = ((crc << 1) ^ poly) & 0xFF if crc & 0x80 else (crc << 1) & 0xFF + return crc + + +def _crc16_ccitt(data: bytes, init: int = 0xFFFF) -> int: + crc = init + for byte in data: + crc ^= byte << 8 + for _ in range(8): + crc = ((crc << 1) ^ 0x1021) & 0xFFFF if crc & 0x8000 \ + else (crc << 1) & 0xFFFF + return crc + + +def check_crc(bits: str) -> list[str]: + """Which of the usual integrity checks the payload satisfies. + + Not proof on its own -- an eight-bit checksum agrees with random data one + time in 256, and these are tried several ways -- but a packet that passes + one *and* arrived several times identically is as sure as this gets. + """ + data = _bytes_of(bits) + if len(data) < 3: + return [] + body, last = data[:-1], data[-1] + found = [] + if (sum(body) & 0xFF) == last: + found.append("checksum-8") + if (sum(body) & 0xFF) == ((0x100 - last) & 0xFF): + found.append("checksum-8 (two's complement)") + for name, poly in (("CRC-8", 0x07), ("CRC-8/MAXIM", 0x31), + ("CRC-8/NRSC", 0x31)): + if _crc8(body, poly) == last: + found.append(name) + break + if len(data) >= 4: + want = (data[-2] << 8) | data[-1] + if _crc16_ccitt(data[:-2]) == want: + found.append("CRC-16/CCITT") + return found + + +def _agreement(bits: list[str]) -> tuple[str, float]: + """The bits common to several repeats, and how strongly they agree. + + Taken bit by bit rather than by demanding the strings be identical: one + repeat clipped by the squelch, or one bit read wrong at the edge of the + burst, should not throw away a packet that was received six times. + """ + usable = [b for b in bits if b] + if not usable: + return "", 0.0 + if len(usable) == 1: + return usable[0], 0.0 + # Group by length and keep the commonest. A capture that opened or + # closed part-way through a repeat has one short burst among several + # whole ones, and truncating everything to the shortest would throw away + # the end of a packet that was received perfectly well several times. + lengths: dict[int, list[str]] = {} + for candidate in usable: + lengths.setdefault(len(candidate), []).append(candidate) + width, usable = max(lengths.items(), key=lambda kv: (len(kv[1]), kv[0])) + if len(usable) == 1: + return usable[0], 0.0 + if width < 8: + return max(usable, key=len), 0.0 + grid = np.array([[c == "1" for c in b[:width]] for b in usable]) + votes = grid.mean(axis=0) + consensus = "".join("1" if v >= 0.5 else "0" for v in votes.tolist()) + # How often the bits agreed, scaled so that a coin toss reads as zero. + certainty = float(np.abs(votes - 0.5).mean()) * 2.0 + return consensus, certainty + + +def _repeat_within(bits: str, min_period: int = 8) -> tuple[str, int]: + """Find a packet sent back to back inside one burst. + + Some transmitters run their repeats together with no gap to split on, so + the same search has to be done in the bit domain. + """ + n = len(bits) + if n < 2 * min_period: + return bits, 1 + for period in range(min_period, n // 2 + 1): + copies = n // period + if copies < 2: + break + chunks = [bits[i * period:(i + 1) * period] for i in range(copies)] + if all(c == chunks[0] for c in chunks[1:]): + return chunks[0], copies + return bits, 1 + + +# --------------------------------------------------------------------------- +# Naming what came out +# --------------------------------------------------------------------------- + +def _describe_payload(bits: str, encoding: Encoding) -> str: + """Name the device class where the shape of the packet gives it away.""" + n = len(bits) + if encoding.name == "PWM" and n == 24: + # EV1527, PT2262 and the clones that fill this band: twenty address + # bits set at the factory or by solder blobs, four for the button. + return "EV1527 / PT2262-style remote" + if encoding.name in ("PWM", "PPM") and 20 <= n <= 80: + return f"{n}-bit {encoding.name} remote or sensor" + return "" + + +@dataclass +class Burst: + """One packet, as read off the air.""" + + bits: str = "" + encoding: str = "" + unit_seconds: float = 0.0 + n_runs: int = 0 + start_seconds: float = 0.0 + duration: float = 0.0 + fit: float = 0.0 + + @property + def hex(self) -> str: + return bits_to_hex(self.bits) + + +@dataclass +class DataDecode: + """What was recovered from a data signal.""" + + ok: bool = False + protocol: str = "" # a named protocol, where one was recognised + encoding: str = "" # PWM / PPM / Manchester / NRZ + baud: float = 0.0 + bits: str = "" # the consensus packet + repeats: int = 1 + agreement: float = 0.0 # how strongly the repeats agreed + checks: list[str] = field(default_factory=list) + bursts: list[Burst] = field(default_factory=list) + messages: list[str] = field(default_factory=list) # decoded text, if any + confidence: float = 0.0 + note: str = "" + + @property + def hex(self) -> str: + return bits_to_hex(self.bits) + + @property + def n_bits(self) -> int: + return len(self.bits) + + def summary(self) -> str: + """One line, for a display that has room for one line.""" + if not self.ok: + return self.note or "no data recovered" + if self.messages: + return self.messages[0] + head = self.protocol or f"{self.encoding} data" + bits = f"{self.n_bits} bits" + rate = f"{self.baud:.0f} baud" if self.baud else "" + reps = f"x{self.repeats}" if self.repeats > 1 else "" + checks = self.checks[0] if self.checks else "" + return " ".join(b for b in (head, bits, rate, reps, checks) if b) + + def report(self, limit: int = 12, width: int = 160) -> list[str]: + """Several lines, for somewhere with room to show the packet.""" + if not self.ok: + return [self.note or "no data recovered"] + out = [self.summary()] + out += self.messages[1:limit] + more = len(self.messages) - limit + if more > 0: + out.append(f"... and {more} more") + if not self.messages and self.bits: + out.append(self.hex[:width]) + out.append(self.bits[:width]) + if self.checks: + out.append("checks: " + ", ".join(self.checks)) + return out + + +def _score(decode: DataDecode, fit: float) -> float: + """How much to believe it. + + Repetition dominates: bits that arrived identically five times are not + noise, whatever the fit said. A checksum adds to that but cannot carry a + decode on its own, because there are several of them and each agrees with + random data now and then. + """ + score = 0.30 + 0.35 * max(0.0, min(1.0, fit)) + if decode.repeats >= 2: + score += 0.15 + 0.10 * min(1.0, (decode.repeats - 2) / 4.0) + score *= 0.6 + 0.4 * decode.agreement + if decode.checks: + score += 0.12 + if decode.protocol: + score += 0.08 + if decode.n_bits < 12: + score *= 0.6 + return float(max(0.0, min(0.99, score))) + + +def decode_data(x: np.ndarray, sample_rate: float, family: str = "", + baud_hint: float = 0.0) -> DataDecode: + """Recover whatever data is in ``x``. + + ``x`` is complex baseband centred on the signal. ``family`` is the + classifier's opinion of what it is, used only to decide which way to + slice it first; both ways are tried regardless, because the classifier + is working from statistics and this is working from the bits. + """ + x = np.asarray(x) + if x.size < 256: + return DataDecode(note="too short to decode") + + # Named protocols first. They know their own framing and checksums, so + # where one of them matches there is nothing to guess at. + named = _try_protocols(x, sample_rate, family, baud_hint) + if named is not None and named.ok: + return named + + # Before anything two-level. A four-level signal sliced down the middle + # yields runs, and runs yield bits, and the bits mean nothing -- so where + # four levels are found the two-level readings are not offered at all, + # even when the symbols could not be framed into anything named. + multi = decode_four_level(x, sample_rate, baud_hint) + if multi is not None: + return multi + + order = ["fsk", "ook"] if family in ("fsk", "psk", "digital", "control") \ + else ["ook", "fsk"] + best: DataDecode | None = None + for how in order: + train = slice_ook(x, sample_rate) if how == "ook" \ + else slice_fsk(x, sample_rate, baud_hint=baud_hint) + if train is None or len(train) < MIN_RUNS // 2: + continue + attempt = _decode_train_set(train) + if attempt.ok and (best is None or + attempt.confidence > best.confidence): + best = attempt + if best is not None and best.confidence > 0.8: + break + if best is not None: + return best + if named is not None: + return named + return DataDecode(note="nothing in it decoded as data") + + +def _decode_train_set(train: PulseTrain) -> DataDecode: + """Split a train into bursts, decode each, and reconcile the repeats.""" + bursts = split_bursts(train) + decoded: list[Burst] = [] + fits: list[float] = [] + for piece in bursts: + encoding, bits = decode_train(piece) + start = piece.start / piece.rate + if bits: + decoded.append(Burst(bits=bits, encoding=encoding.name, + unit_seconds=encoding.unit, + n_runs=len(piece), start_seconds=start, + duration=piece.duration, fit=encoding.fit)) + fits.append(encoding.fit) + if not decoded: + return DataDecode(note="sliced into runs, but no line code fitted") + + # The encoding the majority of the bursts agree on; a single burst read + # as something else is a misread of the same packet. + names = [b.encoding for b in decoded] + encoding_name = max(set(names), key=names.count) + same = [b for b in decoded if b.encoding == encoding_name] + consensus, agreement = _agreement([b.bits for b in same]) + # Only the bursts that carry the agreed packet count as repeats of it. + repeats = sum(1 for b in same if len(b.bits) == len(consensus)) or 1 + if repeats == 1: + consensus, repeats = _repeat_within(consensus) + agreement = 1.0 if repeats > 1 else 0.0 + + unit = float(np.median([b.unit_seconds for b in same if b.unit_seconds]) + or 0.0) + encoding = Encoding(encoding_name, unit, unit, + float(np.mean([b.fit for b in same]))) + if encoding_name in ("PWM", "PPM"): + # The pair length, which is what a bit actually costs on the air. + spans = [b.duration / max(1, len(b.bits)) for b in same if b.bits] + baud = 1.0 / float(np.median(spans)) if spans else 0.0 + else: + baud = encoding.baud + + out = DataDecode( + ok=True, encoding=encoding_name, baud=baud, bits=consensus, + repeats=repeats, agreement=agreement, bursts=decoded, + checks=check_crc(consensus)) + out.protocol = _describe_payload(consensus, encoding) + fit = float(np.mean(fits)) + refused = _refuse(out, fit, share=len(decoded) / max(1, len(bursts))) + if refused: + return DataDecode(note=refused, bursts=decoded) + out.confidence = _score(out, fit) + return out + + +def _refuse(out: DataDecode, fit: float, share: float) -> str: + """Say why a decode is not worth reporting, or "" to report it. + + Written as a gate rather than folded into the confidence score because + the honest answer to a weak decode is "nothing decoded", not a bit string + with a low number beside it that somebody will read anyway. + """ + if out.repeats >= 2 and out.agreement >= 0.9: + return "" # the same bits several times over + if out.checks and out.n_bits >= 24: + return "" # framed and checked + if share < MIN_BURST_SHARE: + return (f"only {share*100:.0f}% of the bursts decoded the same way " + f"-- not a data signal") + if out.encoding == "NRZ" and (fit < MIN_NRZ_FIT + or out.n_bits < MIN_NRZ_BITS): + return "run lengths fit a grid, but nothing frames them as data" + return "" + + +def _try_protocols(x: np.ndarray, sample_rate: float, family: str, + baud_hint: float) -> DataDecode | None: + """Hand the signal to each framed protocol that might own it.""" + from .protocols import PROTOCOLS + + best: DataDecode | None = None + for protocol in PROTOCOLS: + try: + got = protocol(x, sample_rate, baud_hint) + except Exception: + continue # one broken decoder must not stop the rest + if got is not None and got.ok and \ + (best is None or got.confidence > best.confidence): + best = got + return best diff --git a/bandsaunter/protocols.py b/bandsaunter/protocols.py new file mode 100644 index 0000000..fa98c61 --- /dev/null +++ b/bandsaunter/protocols.py @@ -0,0 +1,453 @@ +"""Data protocols that carry their own framing, and so can be read outright. + +The generic decoder in :mod:`bandsaunter.decode` recovers bits and has to +argue about whether they are real. The protocols here do not have that +problem: each has a sync word to find and a checksum to verify, so a frame +either passes or it does not, and one that passes is not a guess. + +Two are implemented, because between them they cover most of what an ordinary +receiver actually hears carrying words rather than measurements: + +``POCSAG`` + Paging. Still in daily use by hospitals, fire services and industrial + plant long after the consumer pagers went away, and it carries plain + text. 512, 1200 or 2400 baud two-level FSK. + +``AX.25 / APRS`` + Amateur packet. 1200 baud AFSK on VHF, and the payload is position + reports, weather and messages -- with the sender's callsign in the + header, which the map already knows what to do with. +""" + +from __future__ import annotations + +import math + +import numpy as np + +from .decode import DataDecode, PulseTrain, nrz_bits, slice_fsk + +__all__ = ["decode_pocsag", "decode_ax25", "PROTOCOLS", "AX25Frame", + "POCSAG_BAUDS", "SYNC_WORD", "MAX_ADDRESS", "pocsag_bits", + "pocsag_codeword"] + + +# --------------------------------------------------------------------------- +# Shared +# --------------------------------------------------------------------------- + +def _find(bits: str, pattern: str, start: int = 0) -> int: + return bits.find(pattern, start) + + +def _invert(bits: str) -> str: + return bits.translate(str.maketrans("01", "10")) + + +# --------------------------------------------------------------------------- +# POCSAG +# --------------------------------------------------------------------------- + +# The frame synchronisation codeword, sent before each batch of sixteen. +SYNC_WORD = 0x7CD215D8 +SYNC_BITS = f"{SYNC_WORD:032b}" + +# The three rates in the standard. Which one a transmitter uses is not +# announced anywhere in the signal, so all three are tried and the one whose +# sync word appears is the right one -- a 32-bit pattern does not turn up in +# the wrong reading by accident. +POCSAG_BAUDS: tuple[float, ...] = (1200.0, 512.0, 2400.0) + +# BCH(31,21) with an even parity bit, which is what protects each codeword. +_BCH_POLY = 0b11101101001 # x^10 + x^9 + x^8 + x^6 + x^5 + x^3 + 1 + +_IDLE = 0x7A89C197 + +# The 20 data bits of a message codeword are packed end to end and then read +# out either as 7-bit ASCII or as 4-bit digits, depending on the pager. +_NUMERIC = "0123456789*U -)(" + + +def _bch_syndrome(word: int) -> int: + """Zero when the codeword's BCH check passes.""" + remainder = word >> 1 # drop the parity bit + for shift in range(30, 9, -1): + if remainder & (1 << shift): + remainder ^= _BCH_POLY << (shift - 10) + return remainder & 0x3FF + + +def _parity_ok(word: int) -> bool: + return bin(word).count("1") % 2 == 0 + + +def _correct(word: int) -> tuple[int, bool]: + """Check a codeword, correcting a single bit error if there is one. + + Worth doing rather than discarding: one bad bit in thirty-two is exactly + what a fading paging signal delivers, and the BCH code was put there to + survive it. + """ + if _bch_syndrome(word) == 0 and _parity_ok(word): + return word, True + for bit in range(32): + trial = word ^ (1 << bit) + if _bch_syndrome(trial) == 0 and _parity_ok(trial): + return trial, True + return word, False + + +def _pocsag_text(payload: str) -> tuple[str, str]: + """Read a run of message bits as text and as digits. + + Both, because nothing in the message says which it is: an alphanumeric + pager sends 7-bit ASCII least significant bit first, a numeric one sends + 4-bit digits, and the only way to tell is to look at what comes out. + """ + letters = [] + for i in range(0, len(payload) - 6, 7): + chunk = payload[i:i + 7] + code = int(chunk[::-1], 2) # LSB first on the air + letters.append(chr(code) if 32 <= code < 127 else + ("\n" if code in (10, 13) else ".")) + digits = [] + for i in range(0, len(payload) - 3, 4): + code = int(payload[i:i + 4][::-1], 2) + digits.append(_NUMERIC[code]) + return "".join(letters).rstrip(". \n"), "".join(digits).rstrip(" ") + + +def _readable(text: str) -> float: + """What share of a string is characters a message would really contain.""" + if not text: + return 0.0 + good = sum(1 for c in text if c.isalnum() or c in " .,:;/-+()@#'\"!?\n") + return good / len(text) + + +def _pocsag_batches(bits: str) -> list[tuple[int, str, str]]: + """Every address and message in a bit stream. + + Returns ``(address, function, text)``. Frames are read from each sync + word independently, so a stream that loses lock partway through still + yields everything before and after it. + """ + out: list[tuple[int, str, str]] = [] + pos = 0 + pending_address: int | None = None + pending_function = 0 + pending_bits: list[str] = [] + + def flush(): + if pending_address is None: + return + payload = "".join(pending_bits) + text, digits = _pocsag_text(payload) + # Whichever reading looks more like a message someone would send. + chosen = text if _readable(text) >= 0.75 and len(text) >= 2 else digits + # Function bits 00..11 are the pager's four addresses, written A to + # D by everyone who documents them. + out.append((pending_address, "ABCD"[pending_function & 3], + chosen.strip())) + + while True: + at = _find(bits, SYNC_BITS, pos) + if at < 0: + break + pos = at + 32 + for frame in range(8): + for half in range(2): + start = pos + (frame * 2 + half) * 32 + if start + 32 > len(bits): + pos = len(bits) + break + word = int(bits[start:start + 32], 2) + if word == _IDLE: + flush() + pending_address, pending_bits = None, [] + continue + word, valid = _correct(word) + if not valid: + continue + if word & 0x80000000: + # A message codeword: twenty bits of payload. + if pending_address is not None: + pending_bits.append(f"{word:032b}"[1:21]) + continue + flush() + # An address codeword. The low three bits of the address are + # not sent: they are which of the eight frames it arrived in. + pending_address = ((word >> 13) & 0x3FFFF) << 3 | frame + pending_function = (word >> 11) & 0x3 + pending_bits = [] + else: + continue + break + pos += 16 * 32 + flush() + return out + + +def decode_pocsag(x: np.ndarray, sample_rate: float, + baud_hint: float = 0.0) -> DataDecode | None: + """Read POCSAG paging out of a two-level FSK signal.""" + train = slice_fsk(x, sample_rate) + if train is None or len(train) < 16: + return None + order = list(POCSAG_BAUDS) + if baud_hint: + order.sort(key=lambda b: abs(b - baud_hint)) + for baud in order: + raw = nrz_bits(train, baud) + if len(raw) < 64: + continue + for bits in (raw, _invert(raw)): + if SYNC_BITS not in bits: + continue + pages = _pocsag_batches(bits) + if not pages: + continue + # A capture usually spans several batches and a transmitter + # repeats its queue, so the same page arrives more than once. + # Reported once, in the order it first appeared. + messages: list[str] = [] + for address, function, text in pages: + line = f"[{address:07d}{function}]" + line = f"{line} {text}" if text else line + if line not in messages: + messages.append(line) + syncs = bits.count(SYNC_BITS) + return DataDecode( + ok=True, protocol=f"POCSAG {baud:.0f}", encoding="NRZ", + baud=baud, bits=bits[:256], repeats=syncs, + agreement=1.0 if syncs > 1 else 0.0, + checks=["BCH(31,21)"], messages=messages, + confidence=min(0.97, 0.72 + 0.05 * len(pages) + + 0.05 * min(3, syncs))) + return None + + +# --------------------------------------------------------------------------- +# AX.25 over Bell 202 AFSK -- APRS and amateur packet +# --------------------------------------------------------------------------- + +MARK_HZ = 1200.0 +SPACE_HZ = 2200.0 +AFSK_BAUD = 1200.0 +_FLAG = "01111110" + + +class AX25Frame: + """One AX.25 frame whose frame-check sequence was correct.""" + + def __init__(self, raw: bytes): + self.raw = raw + self.source = "" + self.destination = "" + self.path: list[str] = [] + self.info = "" + self._parse() + + def _parse(self) -> None: + addresses = [] + i = 0 + while i + 7 <= len(self.raw) and len(addresses) < 10: + field = self.raw[i:i + 7] + call = "".join(chr(b >> 1) for b in field[:6]).strip() + ssid = (field[6] >> 1) & 0x0F + addresses.append(f"{call}-{ssid}" if ssid else call) + i += 7 + if field[6] & 0x01: # the end-of-address bit + break + if len(addresses) >= 2: + self.destination, self.source = addresses[0], addresses[1] + self.path = addresses[2:] + # Skip the control and protocol-identifier bytes. + body = self.raw[i + 2:] if len(self.raw) > i + 2 else b"" + self.info = body.decode("ascii", "replace").rstrip("\r\n") + + def describe(self) -> str: + route = ">".join([self.source or "?", self.destination or "?"] + + self.path) + return f"{route}: {self.info}" if self.info else route + + +def _fcs(data: bytes) -> int: + """The AX.25 frame check: CRC-16/X.25, reflected, inverted at the end.""" + crc = 0xFFFF + for byte in data: + crc ^= byte + for _ in range(8): + crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1 + return crc ^ 0xFFFF + + +def _afsk_symbols(x: np.ndarray, sample_rate: float) -> np.ndarray | None: + """Turn Bell 202 audio into a two-level signal, mark against space. + + A correlator rather than a discriminator: the two tones are close enough + together, and radio audio distorted enough, that measuring which of the + two a bit-length window contains more of works where measuring the + instantaneous frequency does not. + """ + x = np.asarray(x) + if np.iscomplexobj(x): + # FM first: AFSK is audio, and on VHF it arrives inside an FM carrier. + phase = np.unwrap(np.angle(x)) + audio = np.diff(phase) + else: + audio = x.astype(np.float64) + if audio.size < int(sample_rate / 100.0): + return None + audio = audio - float(audio.mean()) + window = int(round(sample_rate / AFSK_BAUD)) + if window < 4: + return None + n = np.arange(window) + out = [] + for tone in (MARK_HZ, SPACE_HZ): + arg = 2.0 * math.pi * tone * n / sample_rate + i = np.convolve(audio, np.cos(arg)[::-1], mode="same") + q = np.convolve(audio, np.sin(arg)[::-1], mode="same") + out.append(np.hypot(i, q)) + return out[0] - out[1] + + +def _hdlc_frames(bits: str) -> list[bytes]: + """Split a bit stream at HDLC flags and undo the bit stuffing.""" + frames: list[bytes] = [] + pos = bits.find(_FLAG) + if pos < 0: + return frames + while True: + # Flags repeat back to back between frames; skip past all of them. + while bits.startswith(_FLAG, pos): + pos += 8 + end = bits.find(_FLAG, pos) + if end < 0: + break + body = bits[pos:end] + pos = end + if len(body) < 8 * 17: # shorter than an empty AX.25 frame + continue + # Undo stuffing: a zero inserted after every five ones. + out, ones = [], 0 + for bit in body: + if ones == 5: + ones = 0 + if bit == "0": + continue # the stuffed bit + out.append(bit) + ones = ones + 1 if bit == "1" else 0 + clean = "".join(out) + whole = len(clean) - len(clean) % 8 + # Least significant bit first on the air. + frames.append(bytes(int(clean[i:i + 8][::-1], 2) + for i in range(0, whole, 8))) + return frames + + +def decode_ax25(x: np.ndarray, sample_rate: float, + baud_hint: float = 0.0) -> DataDecode | None: + """Read AX.25 packet, as APRS uses it, from 1200 baud AFSK.""" + soft = _afsk_symbols(x, sample_rate) + if soft is None: + return None + levels = soft > 0 + edges = np.flatnonzero(np.diff(levels.astype(np.int8))) + 1 + if edges.size < 16: + return None + starts = np.concatenate(([0], edges)) + ends = np.concatenate((edges, [levels.size])) + train = PulseTrain(levels[starts], (ends - starts).astype(np.int64), + float(sample_rate), "fsk") + raw = nrz_bits(train, AFSK_BAUD) + if len(raw) < 200: + return None + + frames: list[AX25Frame] = [] + for stream in (raw, _invert(raw)): + # NRZI: the data is in whether the level changed, not what it is. + decoded = ["1" if a == b else "0" + for a, b in zip(stream, stream[1:])] + for frame in _hdlc_frames("".join(decoded)): + if len(frame) < 17 or _fcs(frame[:-2]) != \ + (frame[-1] << 8 | frame[-2]): + continue + frames.append(AX25Frame(frame[:-2])) + if frames: + break + if not frames: + return None + return DataDecode( + ok=True, protocol="AX.25 / APRS", encoding="NRZI", + baud=AFSK_BAUD, bits=raw[:256], repeats=len(frames), + agreement=1.0, checks=["FCS (CRC-16/X.25)"], + messages=[f.describe() for f in frames], + confidence=min(0.98, 0.85 + 0.04 * len(frames))) + + +# Tried in order; each returns None when the signal is not its own. +PROTOCOLS = (decode_pocsag, decode_ax25) + + +# --------------------------------------------------------------------------- +# Encoding POCSAG, for the simulator and the tests +# --------------------------------------------------------------------------- +# +# Kept next to the decoder rather than in the test helpers so the two cannot +# drift apart: a bug shared by an encoder and its decoder is invisible, and +# the way to avoid one is to have exactly one copy of the polynomial, the +# sync word and the idle word for both to use. + +def pocsag_codeword(payload: int) -> int: + """Add the BCH check bits and the parity bit to 21 bits of payload.""" + word = (payload & 0x1FFFFF) << 10 + remainder = word + for shift in range(30, 9, -1): + if remainder & (1 << shift): + remainder ^= _BCH_POLY << (shift - 10) + word = (word | (remainder & 0x3FF)) << 1 + return word | (1 if bin(word).count("1") % 2 else 0) + + +# Eighteen bits go out on the air and three more are implied by which frame +# the codeword arrived in, so this is the whole address space. +MAX_ADDRESS = (1 << 21) - 1 + + +def pocsag_bits(pages, preamble: int = 600) -> str: + """A complete POCSAG transmission: preamble, then batches of sixteen. + + Raises on an address that will not fit rather than truncating it: a + stream built around a silently mangled address decodes to a different + pager, which is worse than not building one. + """ + for address, _, _ in pages: + if not 0 <= address <= MAX_ADDRESS: + raise ValueError( + f"POCSAG address {address} is outside 0-{MAX_ADDRESS}") + out = ["10" * (preamble // 2)] + batch: list[int] = [] + + def flush() -> None: + if not batch: + return + while len(batch) < 16: + batch.append(_IDLE) + out.append(SYNC_BITS + "".join(f"{w:032b}" for w in batch[:16])) + batch.clear() + + for address, function, text in pages: + frame = address & 0x7 + while len(batch) < frame * 2: + batch.append(_IDLE) + batch.append(pocsag_codeword(((address >> 3) << 2) | (function & 3))) + payload = "".join(f"{ord(c):07b}"[::-1] for c in text) + payload += "0" * (-len(payload) % 20) + for i in range(0, len(payload), 20): + batch.append(pocsag_codeword((1 << 20) | int(payload[i:i + 20], 2))) + if len(batch) >= 16: + flush() + flush() + return "".join(out) diff --git a/bandsaunter/recorder.py b/bandsaunter/recorder.py index 9235094..168a5a0 100755 --- a/bandsaunter/recorder.py +++ b/bandsaunter/recorder.py @@ -68,6 +68,19 @@ class HitRecord: ctcss_hz: float = 0.0 baud: float = 0.0 + # What a data signal turned out to say. Kept apart from the transcript + # because it is not speech and nothing that reads transcripts should + # have to know about it. + data_protocol: str = "" # POCSAG, AX.25, EV1527-style, ... + data_encoding: str = "" # PWM / PPM / Manchester / NRZ / 4-level FSK + data_bits: str = "" + data_hex: str = "" + data_repeats: int = 0 + data_checks: list[str] = field(default_factory=list) + data_messages: list[str] = field(default_factory=list) + data_confidence: float = 0.0 + data_path: str = "" + category: str = "" # voice / cw / digital / carrier / noise signal_score: float = 0.0 voice_score: float = 0.0 @@ -85,6 +98,34 @@ class HitRecord: kept: bool = True features: dict = field(default_factory=dict) + def data_headline(self) -> str: + """What kind of data it was, never what the data said.""" + if not self.data_encoding: + return "" + bits = [self.data_protocol or f"{self.data_encoding} data"] + if self.data_messages: + n = len(self.data_messages) + bits.append(f"{n} message{'' if n == 1 else 's'}") + elif self.data_bits: + bits.append(f"{len(self.data_bits)} bits") + if self.baud: + bits.append(f"{self.baud:.0f} baud") + if self.data_repeats > 1: + bits.append(f"x{self.data_repeats}") + if self.data_checks: + bits.append(self.data_checks[0]) + return " ".join(bits) + + def data_summary(self) -> str: + """One line for a display, or "" if nothing was decoded. + + The message where there is one, because that is what somebody wants + to read; the description of the packet where there is not. + """ + if self.data_messages: + return self.data_messages[0] + return self.data_headline() + def to_dict(self) -> dict: return asdict(self) @@ -95,6 +136,8 @@ class HitRecord: self.classification or "unclassified"] if self.morse_text: bits.append(f'CW "{self.morse_text.strip()[:40]}"') + elif self.data_messages: + bits.append(self.data_messages[0][:48]) elif self.ctcss_hz: bits.append(f"CTCSS {self.ctcss_hz:.1f}") return " ".join(bits) diff --git a/bandsaunter/scanner.py b/bandsaunter/scanner.py index 4341837..66515ff 100755 --- a/bandsaunter/scanner.py +++ b/bandsaunter/scanner.py @@ -24,6 +24,7 @@ from .bandplan import fmt_hz, label_for, presets_covering from .callsign import CallsignBook, find_callsigns from .classify import classify, ssb_alignment from .config import ScanConfig, remember_lockouts +from .decode import decode_data from .demod import make_demodulator from .device import RtlSdrDevice, RtlSdrError from .kml import KmlLog @@ -879,7 +880,11 @@ class Scanner: 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 \ + and not hit.data_messages: + # A decoded packet outranks the content check. That test works + # from statistics -- how noise-like, how speech-like -- and a + # frame whose own checksum came out right is not a statistic. hit.kept = False hit.stop_reason = f"no signal content: {verdict.reason}" self._reject(rec, hit) @@ -891,6 +896,7 @@ class Scanner: hit.filename = rec.stem hit.audio_path = str(rec.audio_path) if cfg.save_audio else "" hit.iq_path = str(rec.iq_path) if cfg.save_iq else "" + self._write_data_file(rec, hit) # Add this transmission to the running file for its frequency, after # a spoken timestamp, so one channel plays back as one recording. @@ -1060,6 +1066,93 @@ class Scanner: meta_path=Path(hit.meta_path) if hit.meta_path else None, recording=rec.audio_path.name) + # Families whose signals carry bits. Voice and Morse are excluded not to + # save the work -- it is a fraction of a second -- but because a decoder + # run over speech will eventually find a window that fits, and a scanner + # that occasionally reports a doorbell code from a conversation is worse + # than one that never looks. + DATA_FAMILIES = ("ook", "fsk", "psk", "digital", "control", "data") + + def _decode_payload(self, rec: Recording, hit: HitRecord, demod, + cls) -> None: + """Read the bits out of a data capture and attach them to the hit.""" + if not self.cfg.decode_data or cls.family not in self.DATA_FAMILIES: + return + iq = rec.classification_iq(limit=0) + if iq.size < 2048: + return + try: + got = decode_data(iq, demod.if_rate, family=cls.family, + baud_hint=hit.baud) + except Exception as exc: + self._error(exc) + return + if not got.ok: + return + hit.data_protocol = got.protocol + hit.data_encoding = got.encoding + hit.data_bits = got.bits[:512] + hit.data_hex = got.hex[:512] + hit.data_repeats = got.repeats + hit.data_checks = list(got.checks) + hit.data_messages = list(got.messages) + hit.data_confidence = round(got.confidence, 3) + if got.baud: + hit.baud = got.baud + if got.protocol: + # A named protocol beats the modulation label: "2-FSK" is true + # and useless where "POCSAG 1200" says what it is. + hit.classification = got.protocol + hit.confidence = max(hit.confidence, got.confidence) + hit.reasons.insert(0, f"decoded: {got.summary()}") + self._announce_data(hit, got) + + def _write_data_file(self, rec: Recording, hit: HitRecord) -> None: + """Put what was decoded beside the recording, as text. + + A separate file from the transcript: this is not speech, and the + browser, the callsign search and anything else that reads transcripts + should not have to sort one kind from the other. Written only once + the capture has been renamed and kept, so it cannot be left orphaned + beside a recording that was thrown away. + """ + if not (hit.data_messages or hit.data_bits): + return + path = rec.dir / f"{rec.stem}_data.txt" + lines = [f"# {fmt_hz(hit.frequency)} {hit.started_iso}", + f"# {hit.data_headline()}"] + lines += hit.data_messages + if hit.data_bits and not hit.data_messages: + lines += [hit.data_hex, hit.data_bits] + try: + path.write_text("\n".join(lines) + "\n") + hit.data_path = str(path) + except OSError as exc: + self._error(exc) + + def _announce_data(self, hit: HitRecord, got) -> None: + """Say on the display what was just read off the air.""" + self._status(f"decoded {fmt_hz(hit.frequency)}: {got.summary()}") + # APRS carries callsigns, and the map already knows what to do with + # one. Nothing else in a data packet is a person. + book = self.callsigns + if book is None or "AX.25" not in got.protocol: + return + changed = False + for message in got.messages: + call = message.split(">", 1)[0].split("-", 1)[0].strip() + if not call or not call.isalnum(): + continue + entry = book.get(call) + self.heard[entry.call] = self.heard.get(entry.call, 0) + 1 + if self.kml is not None: + changed |= self.kml.add(entry, hit.frequency, hit.started_at, + hit.band, hit.filename) + if changed and self.kml is not None: + # Saved as we go, not only at the end: a scan left running + # overnight and stopped with a signal should still have its map. + self.kml.save() + def _on_transcript(self, path, result, job) -> None: """Pull callsigns out of a finished transcript and map them. @@ -1141,6 +1234,10 @@ class Scanner: if k != "extras" and not k.startswith("_") } + # After the features, so the decoder is handed the classifier's own + # symbol-rate estimate rather than a zero. + self._decode_payload(rec, hit, demod, cls) + if morse is not None and morse.is_morse: hit.morse_text = morse.text hit.morse_wpm = round(morse.wpm, 1) @@ -1152,7 +1249,13 @@ class Scanner: hit.reasons.append( "keyed carrier but the Morse timing did not resolve") + # A decode that has repeats or a checksum behind it outranks the + # audio content check below. A burst of on-off keying demodulated as + # FM audio is a buzz, and the speech detector likes a buzz; but bits + # that arrived identically twelve times are not a conversation. + decoded_firmly = hit.data_confidence >= 0.7 if verdict is not None and verdict.category == "voice" and \ + not decoded_firmly and \ hit.family not in ("nfm", "wfm", "am", "ssb"): # The content check found speech, so the modulation label was # wrong. Speech on a quiet FM channel is easy to mistake for @@ -1174,6 +1277,15 @@ class Scanner: hit.classification = label hit.family = fam + if decoded_firmly and verdict is not None and \ + verdict.category != "digital": + # It carries data: that is what "content" means for a signal + # nobody speaks on, and the packet is better evidence of it than + # any statistic the content check could compute. + verdict.category = "digital" + verdict.accept = True + verdict.reason = f"decoded: {hit.data_summary()}" + if verdict is not None: hit.category = verdict.category hit.signal_score = round(verdict.score, 3) diff --git a/bandsaunter/settings.py b/bandsaunter/settings.py index 7ca9d8b..c7681cc 100644 --- a/bandsaunter/settings.py +++ b/bandsaunter/settings.py @@ -390,6 +390,14 @@ _TABLE: tuple[Setting, ...] = ( unit="s", minimum=0.0, flags=("--transcribe-min",), metavar="SEC"), # -- callsigns --------------------------------------------------------- + S("decode_data", "Decode data signals", "Output", "bool", + "read the packets out of anything carrying data", + "On-off keyed remotes and sensors, two-level FSK, POCSAG paging and " + "APRS packet are all read down to their bits, and named where the " + "framing gives them away. Costs a fraction of a second per capture " + "and only runs on captures the classifier called data.", + flags=("--decode-data",), off_flags=("--no-decode-data",)), + S("callsign_lookup", "Look callsigns up", "Callsigns", "bool", "ask the licence database who a callsign belongs to", "Callsigns heard in a transcript are looked up in the FCC's published " @@ -763,6 +771,18 @@ _GUIDANCE: dict[str, str] = { "Do not bother transcribing captures shorter than this. Very short " "clips rarely contain a whole word and mostly produce noise or " "nothing, while still costing the processing.", + "decode_data": + "Read what a data signal actually says. A great deal of what a " + "scanner finds is not speech: doorbells, tyre-pressure sensors, " + "weather stations, remote controls, paging, packet radio. Each one " + "is sliced into its pulses, the line code worked out from the " + "pulse lengths alone, and the bits reported -- with the packet " + "named where its framing says what it is, and the message printed " + "in full where the protocol carries one. The check that keeps it " + "honest is repetition: these transmitters send the same packet " + "several times over, and bits that come back identical every time " + "did not come from noise. Turn it off to save a little processing " + "on a busy band.", "callsign_lookup": "When someone gives their callsign, look it up and say who they " "are. The data is the FCC's own published licence register, which " diff --git a/bandsaunter/simulator.py b/bandsaunter/simulator.py index 5e34c7c..11f914d 100755 --- a/bandsaunter/simulator.py +++ b/bandsaunter/simulator.py @@ -107,7 +107,8 @@ class VirtualTransmitter: """One synthetic signal on the air.""" frequency: float - mode: str = "nfm" # nfm wfm am usb cw fsk2 fsk4 psk carrier ook + mode: str = "nfm" # nfm wfm am usb cw fsk2 fsk4 psk + # carrier ook packet pocsag power: float = 0.35 # linear amplitude bandwidth: float = 12_500.0 label: str = "" @@ -121,6 +122,12 @@ class VirtualTransmitter: wpm: float = 18.0 baud: float = 4800.0 deviation: float = 2_500.0 + # For the packet modes: the payload, and how many times a transmitter + # repeats it. Cheap remotes send everything three to ten times over, + # which is what makes a decode checkable. + payload: str = "101100100011010101001110" + repeats: int = 5 + pages: tuple = () # (address, function, text) for POCSAG pitch_hz: float = 120.0 # synthetic talker's voice pitch _phase: float = field(default=0.0, init=False, repr=False) @@ -175,6 +182,19 @@ class VirtualTransmitter: elif m == "ook": env = (self._symbols(t, fs, 2, self.baud) > 0).astype(np.float64) out = env * np.exp(1j * self._advance(np.zeros(n), fs)) + elif m == "pocsag": + # A real paging batch, not random keying: preamble, sync word, + # addressed message codewords with their BCH check bits. A + # decoder has to have something correct to find. + lv = self._pocsag_levels(t, fs) + out = np.exp(1j * self._advance(lv * self.deviation, fs)) + elif m == "packet": + # A real remote, not a random bit stream: a pulse-width coded + # payload sent several times over with a silence between the + # repeats. Random keying exercises the classifier but there is + # nothing in it for a decoder to get right. + env = self._packet_envelope(t, fs) + out = env * np.exp(1j * self._advance(np.zeros(n), fs)) elif m in ("fsk2", "fsk4"): levels = 2 if m == "fsk2" else 4 lv = self._shaped_levels(t, fs, levels, self.baud) @@ -188,6 +208,33 @@ class VirtualTransmitter: return (self.power * out).astype(np.complex64) + def _pocsag_levels(self, t: np.ndarray, fs: float) -> np.ndarray: + """The POCSAG bit stream as +/-1, read out against absolute time.""" + from .protocols import pocsag_bits + pages = tuple(self.pages) or ((1234568, 3, "TEST PAGE"),) + bits = pocsag_bits(pages) + table = np.where(np.frombuffer(bits.encode(), dtype=np.uint8) == + ord("1"), 1.0, -1.0) + unit = 1.0 / max(1.0, self.baud) + pos = np.floor((t % (table.size * unit)) / unit).astype(np.int64) + return table[np.clip(pos, 0, table.size - 1)] + + def _packet_envelope(self, t: np.ndarray, fs: float) -> np.ndarray: + """Pulse-width keying of ``payload``, repeated, on absolute time. + + Read out against absolute time like everything else here, so a block + boundary falls in the middle of a packet without disturbing it. + """ + unit = 1.0 / max(1.0, self.baud) # the short pulse + frame = [] + for bit in self.payload: + frame += [1] * (3 if bit == "1" else 1) + frame += [0] * (1 if bit == "1" else 3) + gap = [0] * (24 * 4) # silence between repeats + cycle = np.array((frame + gap) * max(1, self.repeats), dtype=np.float64) + pos = np.floor((t % (cycle.size * unit)) / unit).astype(np.int64) + return cycle[np.clip(pos, 0, cycle.size - 1)] + def _advance(self, inst_freq: np.ndarray, fs: float) -> np.ndarray: """Integrate an instantaneous-frequency series, keeping phase continuous.""" ph = self._phase + np.cumsum(2.0 * np.pi * inst_freq / fs) @@ -326,11 +373,13 @@ def default_transmitters() -> list[VirtualTransmitter]: V(97_500_000, "wfm", 0.45, 180_000, "FM broadcast"), V(460_025_000, "fsk4", 0.30, 12_500, "P25-style digital voice", baud=4800, deviation=1_800, period_seconds=13, on_seconds=4), - V(929_612_500, "fsk2", 0.28, 12_500, "POCSAG pager", baud=1200, - deviation=2_400, period_seconds=11, on_seconds=2, phase_offset=5), + V(929_612_500, "pocsag", 0.28, 12_500, "POCSAG pager", baud=1200, + deviation=4_500, period_seconds=11, on_seconds=2.5, phase_offset=5, + pages=((1234568, 3, "ENGINE 4 RESPOND"), (98765, 0, "CALL EXT 4412"))), V(446_000_000, "carrier", 0.25, 1_000, "unmodulated carrier"), - V(433_920_000, "ook", 0.30, 40_000, "ISM remote", baud=2000, - period_seconds=9, on_seconds=1.2), + V(433_920_000, "packet", 0.30, 40_000, "ISM remote", baud=2000, + period_seconds=9, on_seconds=1.2, + payload="101100100011010101001110", repeats=5), # 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", diff --git a/bandsaunter/tui.py b/bandsaunter/tui.py index 0f5234e..9466e11 100644 --- a/bandsaunter/tui.py +++ b/bandsaunter/tui.py @@ -559,7 +559,26 @@ 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": ("Callsigns and the map", """ + "9": ("Reading data signals", """ +Much of what a scanner finds is not speech: doorbells, tyre-pressure sensors, +weather stations, remote controls, paging, packet radio. All of it is read. + +Whatever the modulation, a data signal comes down to the same shape once it has +been sliced — a train of runs whose lengths carry the information — and which +line code it is (PWM, PPM, Manchester, NRZ) is worked out from the runs alone +rather than configured. Four-level FSK, as P25 and DMR use it, is recognised +as such and read as symbols; where a frame sync word appears the system is +named. + +POCSAG paging and APRS packet carry their own checksums, so those are read in +full: the address and message text of a page, the callsign and position of an +APRS beacon. A decoded callsign goes onto the map with the rest. + +What keeps it honest is repetition. Noise sliced at a threshold produces runs, +and runs produce bits, so a reading with nothing behind it is reported as +nothing at all. These transmitters send the same packet three to ten times +over, and bits that come back identical every time did not come from noise."""), + "10": ("Callsigns and the map", """ Anyone who identifies themselves in a transcript is picked out and looked up in the FCC's published licence register: the name, the town, the class of licence. The callsign is the only thing sent, and each one is asked about once and then @@ -575,7 +594,7 @@ can reach. Turn 'Look callsigns up' off to keep the scan entirely offline; callsigns are still found, and named by country from their prefix. Clear 'Map file' to stop writing the map. US licence records are public, and include addresses."""), - "10": ("Keys during a scan", """ + "11": ("Keys during a scan", """ q stop the scan p pause and resume s skip the signal being recorded and carry on sweeping diff --git a/bandsaunter/ui.py b/bandsaunter/ui.py index 191cb5f..4c4182b 100755 --- a/bandsaunter/ui.py +++ b/bandsaunter/ui.py @@ -12,6 +12,7 @@ from dataclasses import dataclass import numpy as np from rich.console import Console, Group +from rich.markup import escape from rich.panel import Panel from rich.table import Table from rich.text import Text @@ -175,7 +176,13 @@ class ScanDisplay: self._dirty = True def on_status(self, msg: str): - self.messages.appendleft(msg) + # Escaped on the way in. A status line can carry text straight off + # the air -- a decoded pager message, a callsign, a Morse decode -- + # and rich reads square brackets as markup: "[/x]" in a message is + # not a style, it is an exception in the middle of the display. + # Errors keep their markup, which is written here rather than by + # whatever raised. + self.messages.appendleft(escape(msg)) self._dirty = True def on_error(self, exc: Exception): @@ -258,7 +265,7 @@ class ScanDisplay: # 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"[bold black on yellow] {escape(r.note)} [/bold black on yellow] " f"{fmt_hz(r.frequency)}{where} [{r.mode}] " f"SNR {r.snr:5.1f} dB [yellow]skipping[/yellow]"), border_style="yellow", padding=(0, 1)) @@ -350,7 +357,14 @@ class ScanDisplay: for h in shown: extra = "" if h.morse_text: - extra = f' [yellow]"{h.morse_text.strip()[:32]}"[/yellow]' + extra = f' [yellow]"{escape(h.morse_text.strip()[:32])}"[/yellow]' + elif h.data_messages: + # What was decoded is the most interesting thing about a data + # capture, and far more so than its baud rate. + extra = (" [bright_cyan]" + + escape(h.data_messages[0][:40]) + "[/bright_cyan]") + elif h.data_encoding: + extra = f" [cyan]{escape(h.data_headline()[:40])}[/cyan]" elif h.ctcss_hz: extra = f" [grey62]CTCSS {h.ctcss_hz:.1f}[/grey62]" elif h.baud: @@ -429,10 +443,18 @@ def print_hit(console: Console, hit: HitRecord) -> None: indent = 26 + _PLAIN_BAND + 2 if hit.morse_text: console.print(f'{"":>{indent}}[yellow]Morse @ {hit.morse_wpm:.0f} WPM: ' - f'"{hit.morse_text.strip()}"[/yellow]', highlight=False) - if hit.reasons: - console.print(f'{"":>{indent}}[grey54]{hit.reasons[0]}[/grey54]', + f'"{escape(hit.morse_text.strip())}"[/yellow]', highlight=False) + for line in hit.data_messages[:4]: + console.print(f'{"":>{indent}}[bright_cyan]{escape(line)}' + f'[/bright_cyan]', highlight=False) + if hit.data_hex and not hit.data_messages: + console.print(f'{"":>{indent}}[cyan]{escape(hit.data_headline())}' + f'[/cyan] [grey54]{hit.data_hex[:48]}[/grey54]', + highlight=False) + if hit.reasons: + console.print(f'{"":>{indent}}[grey54]{escape(hit.reasons[0])}' + f'[/grey54]', highlight=False) def print_band_table(console: Console, presets, title: str = "band plan") -> None: diff --git a/packaging/bandsaunter.1 b/packaging/bandsaunter.1 index 3ccf408..64ae95b 100644 --- a/packaging/bandsaunter.1 +++ b/packaging/bandsaunter.1 @@ -1,5 +1,5 @@ .\" Generated by packaging/make-man.py -- do not edit by hand. -.TH BANDSAUNTER 1 "2026-08-28" "bandsaunter 2026-08-28_01" "User Commands" +.TH BANDSAUNTER 1 "2026-08-28" "bandsaunter 2026-08-28_02" "User Commands" .SH NAME bandsaunter \- scan, record and identify radio signals with an RTL-SDR .SH SYNOPSIS @@ -543,6 +543,15 @@ Setting name \fBdecode_morse\fR, default \fByes\fR. Turn keyed carriers into readable text, with the sending speed. Morse is still in daily use by amateurs and by beacons, and this saves you learning to read it by ear. It costs almost nothing when there is no Morse about. .RE .TP +.B --decode-data / --no-decode-data +Decode data signals \[em] read the packets out of anything carrying data. +.br +Setting name \fBdecode_data\fR, default \fByes\fR. +.RS +.PP +Read what a data signal actually says. A great deal of what a scanner finds is not speech: doorbells, tyre-pressure sensors, weather stations, remote controls, paging, packet radio. Each one is sliced into its pulses, the line code worked out from the pulse lengths alone, and the bits reported -- with the packet named where its framing says what it is, and the message printed in full where the protocol carries one. The check that keeps it honest is repetition: these transmitters send the same packet several times over, and bits that come back identical every time did not come from noise. Turn it off to save a little processing on a busy band. +.RE +.TP .B --log-file Log file \[em] name of the run log inside the output directory. .br @@ -934,6 +943,78 @@ than a directory of placeholders. .BR saunterbrowse (1) reads these back, and lists any callsigns it finds in them with the licence they belong to. +.SH DECODING DATA +A great deal of what a scanner finds is not speech. Doorbells, tyre\-pressure +sensors, weather stations, remote controls, paging and packet radio all carry +words or numbers that a receiver can read, and +.B bandsaunter +reads them. +.PP +Whatever the modulation, a data signal comes down to the same shape once it +has been sliced: a train of alternating runs whose lengths carry the +information. On\-off keying gives that directly \[em] the carrier is up or it is +down \[em] and two\-level FSK gives the same thing from the discriminator, one +tone or the other. So both are reduced to runs and everything after that is +shared. +.PP +What the runs mean is the line code, and it is worked out from the runs alone +rather than being configured: +.TP +.B PWM +The pulse carries the bit and the gap or the period holds still. Nearly every +cheap 433 MHz remote, and everything built on an EV1527 or PT2262. +.TP +.B PPM +The pulse holds still and the gap carries the bit. The other half of the same +market. +.TP +.B Manchester +Every bit is a transition in the middle of its own period, so runs come in +only two lengths. +.TP +.B NRZ +The level is held for as many symbol periods as there are bits. What a framed +protocol sits on top of. +.PP +Four\-level FSK \[em] C4FM, as P25, DMR and NXDN use it \[em] is recognised as +such and read as symbols rather than being sliced down the middle, which would +give bits that mean nothing. Where a frame sync word appears the system is +named outright. +.SH PROTOCOLS THAT CAN BE READ IN FULL +Two carry their own framing and checksums, so a frame either passes or it does +not, and one that passes is not a guess. +.TP +.B POCSAG +Paging, at 512, 1200 or 2400 baud. The rate is not announced anywhere in the +signal, so all three are tried and the one whose sync word appears is the +right one. Each codeword is checked, and a single bit error is corrected, +against the BCH code the standard puts there for the purpose. The address, the +function letter and the message text are all reported. +.TP +.B "AX.25 / APRS" +Amateur packet on 1200 baud AFSK. The frame check has to come out right before +a frame is reported at all. The sender's callsign, the digipeater path and the +payload are shown \[em] and the callsign goes onto the map with the rest. +.SH BELIEVING A DECODE +A decoder that always returns something is worse than useless: noise sliced at +a threshold produces runs, and runs produce bits. Three things guard against +that. +.PP +The runs have to quantise to the line code's own grid, and a decode whose runs +are scattered is thrown away. Most of the bursts in a capture have to decode +the same way, because a data signal is data all the way through and one lucky +window among eight is a coincidence. And, much the strongest, the packet has +to repeat \[em] these transmitters send the same thing three to ten times over, +and bits that come back identical every time did not come from noise. +.PP +A bare reading with none of that behind it, where the runs merely happened to +land on a grid, is reported as nothing at all rather than as a bit string with +a low number beside it that somebody will read anyway. +.PP +A decode that does have repeats or a checksum behind it outranks the content +check: a burst of keying demodulated as FM audio is a buzz, and the speech +detector likes a buzz, but a frame whose own checksum came out right is not a +statistic. .SH THE MAP A callsign heard in a transcript is looked up in the FCC's published licence data, which gives the licensee, the town, and coordinates. Those go into a @@ -995,6 +1076,9 @@ Where recordings, transcripts and logs are written, unless .B \-\-output says otherwise. Chosen on first run. .TP +.IR ... _data.txt +What a data capture said, where anything was decoded. +.TP .I ~/bandsaunter/callsigns.kml The map of stations heard, added to as scans run. .TP diff --git a/packaging/make-browse-man.py b/packaging/make-browse-man.py index 12d92b2..b69fe65 100755 --- a/packaging/make-browse-man.py +++ b/packaging/make-browse-man.py @@ -225,6 +225,21 @@ KML is the format Google Earth uses. .BR marble (1) and OsmAnd open it too, and it is XML, so a scan interrupted halfway through leaves a file that still opens. +.SH DECODED DATA +Where a capture carried data rather than speech, what was decoded takes the +place of the transcript at the top of the screen: the kind of packet, and then +the message. A pager's text, an APRS position report, or the bits and hex of a +remote control. It is searchable with +.B / +like anything else, so "which page mentioned engine 4" is a question that can +be asked here. +.PP +The text comes from the +.I _data.txt +beside the recording, or from the sidecar where there is none. +.BR bandsaunter (1) +describes how it is decoded and what has to be true before a decode is +believed. .SH TRANSCRIPTS A transcript appears only where a recogniser produced one, which means the capture was judged to be speech and @@ -252,6 +267,9 @@ Its measurements and identification. .TP .IR ... _transcription.txt What was said, where a recogniser heard speech. +.TP +.IR ... _data.txt +What was decoded, where the capture carried data. .SH ENVIRONMENT .TP .B BANDSAUNTER_OUTPUT diff --git a/packaging/make-man.py b/packaging/make-man.py index c340ea6..d096e3e 100755 --- a/packaging/make-man.py +++ b/packaging/make-man.py @@ -365,6 +365,78 @@ than a directory of placeholders. .BR saunterbrowse (1) reads these back, and lists any callsigns it finds in them with the licence they belong to. +.SH DECODING DATA +A great deal of what a scanner finds is not speech. Doorbells, tyre\-pressure +sensors, weather stations, remote controls, paging and packet radio all carry +words or numbers that a receiver can read, and +.B bandsaunter +reads them. +.PP +Whatever the modulation, a data signal comes down to the same shape once it +has been sliced: a train of alternating runs whose lengths carry the +information. On\-off keying gives that directly \[em] the carrier is up or it is +down \[em] and two\-level FSK gives the same thing from the discriminator, one +tone or the other. So both are reduced to runs and everything after that is +shared. +.PP +What the runs mean is the line code, and it is worked out from the runs alone +rather than being configured: +.TP +.B PWM +The pulse carries the bit and the gap or the period holds still. Nearly every +cheap 433 MHz remote, and everything built on an EV1527 or PT2262. +.TP +.B PPM +The pulse holds still and the gap carries the bit. The other half of the same +market. +.TP +.B Manchester +Every bit is a transition in the middle of its own period, so runs come in +only two lengths. +.TP +.B NRZ +The level is held for as many symbol periods as there are bits. What a framed +protocol sits on top of. +.PP +Four\-level FSK \[em] C4FM, as P25, DMR and NXDN use it \[em] is recognised as +such and read as symbols rather than being sliced down the middle, which would +give bits that mean nothing. Where a frame sync word appears the system is +named outright. +.SH PROTOCOLS THAT CAN BE READ IN FULL +Two carry their own framing and checksums, so a frame either passes or it does +not, and one that passes is not a guess. +.TP +.B POCSAG +Paging, at 512, 1200 or 2400 baud. The rate is not announced anywhere in the +signal, so all three are tried and the one whose sync word appears is the +right one. Each codeword is checked, and a single bit error is corrected, +against the BCH code the standard puts there for the purpose. The address, the +function letter and the message text are all reported. +.TP +.B "AX.25 / APRS" +Amateur packet on 1200 baud AFSK. The frame check has to come out right before +a frame is reported at all. The sender's callsign, the digipeater path and the +payload are shown \[em] and the callsign goes onto the map with the rest. +.SH BELIEVING A DECODE +A decoder that always returns something is worse than useless: noise sliced at +a threshold produces runs, and runs produce bits. Three things guard against +that. +.PP +The runs have to quantise to the line code's own grid, and a decode whose runs +are scattered is thrown away. Most of the bursts in a capture have to decode +the same way, because a data signal is data all the way through and one lucky +window among eight is a coincidence. And, much the strongest, the packet has +to repeat \[em] these transmitters send the same thing three to ten times over, +and bits that come back identical every time did not come from noise. +.PP +A bare reading with none of that behind it, where the runs merely happened to +land on a grid, is reported as nothing at all rather than as a bit string with +a low number beside it that somebody will read anyway. +.PP +A decode that does have repeats or a checksum behind it outranks the content +check: a burst of keying demodulated as FM audio is a buzz, and the speech +detector likes a buzz, but a frame whose own checksum came out right is not a +statistic. .SH THE MAP A callsign heard in a transcript is looked up in the FCC's published licence data, which gives the licensee, the town, and coordinates. Those go into a @@ -426,6 +498,9 @@ Where recordings, transcripts and logs are written, unless .B \-\-output says otherwise. Chosen on first run. .TP +.IR ... _data.txt +What a data capture said, where anything was decoded. +.TP .I ~/bandsaunter/callsigns.kml The map of stations heard, added to as scans run. .TP diff --git a/packaging/saunterbrowse.1 b/packaging/saunterbrowse.1 index a084641..bc716d8 100644 --- a/packaging/saunterbrowse.1 +++ b/packaging/saunterbrowse.1 @@ -1,5 +1,5 @@ .\" Generated by packaging/make-browse-man.py -- do not edit by hand. -.TH SAUNTERBROWSE 1 "2026-08-28" "bandsaunter 2026-08-28_01" "User Commands" +.TH SAUNTERBROWSE 1 "2026-08-28" "bandsaunter 2026-08-28_02" "User Commands" .SH NAME saunterbrowse \- read and listen to what a bandsaunter scan collected .SH SYNOPSIS @@ -207,6 +207,21 @@ KML is the format Google Earth uses. .BR marble (1) and OsmAnd open it too, and it is XML, so a scan interrupted halfway through leaves a file that still opens. +.SH DECODED DATA +Where a capture carried data rather than speech, what was decoded takes the +place of the transcript at the top of the screen: the kind of packet, and then +the message. A pager's text, an APRS position report, or the bits and hex of a +remote control. It is searchable with +.B / +like anything else, so "which page mentioned engine 4" is a question that can +be asked here. +.PP +The text comes from the +.I _data.txt +beside the recording, or from the sidecar where there is none. +.BR bandsaunter (1) +describes how it is decoded and what has to be true before a decode is +believed. .SH TRANSCRIPTS A transcript appears only where a recogniser produced one, which means the capture was judged to be speech and @@ -234,6 +249,9 @@ Its measurements and identification. .TP .IR ... _transcription.txt What was said, where a recogniser heard speech. +.TP +.IR ... _data.txt +What was decoded, where the capture carried data. .SH ENVIRONMENT .TP .B BANDSAUNTER_OUTPUT diff --git a/tests/signals.py b/tests/signals.py index afca258..bef80bf 100644 --- a/tests/signals.py +++ b/tests/signals.py @@ -74,3 +74,154 @@ def make(kind, n=64000, fs=FS, snr_db=30.0, seed=3): else: raise ValueError(kind) return _noise(np.asarray(x, dtype=np.complex128), snr_db, rng) + + +# --------------------------------------------------------------------------- +# Packets. What the decoder is for: signals that carry something, rather +# than random keying that only exercises the classifier. +# --------------------------------------------------------------------------- + +def _rf(env, fs, snr_db=30.0, seed=1): + """An envelope on a carrier, with noise.""" + x = np.asarray(env, dtype=np.float64).astype(np.complex128) + rng = np.random.default_rng(seed) + p = float(np.mean(np.abs(x) ** 2)) or 1e-9 + n = np.sqrt(p / (2 * 10 ** (snr_db / 10.0))) + return (x + n * (rng.standard_normal(x.size) + + 1j * rng.standard_normal(x.size))).astype(np.complex64) + + +def ook_pwm(bits, fs=50_000.0, alpha=350e-6, repeats=4, gap=10e-3, + sync=True, snr_db=30.0, seed=1, fixed_gap=None): + """Pulse-width keying, as an EV1527 or PT2262 remote sends it. + + With ``fixed_gap`` the gap is constant and only the pulse varies; without + it the two are complementary and the bit period is constant, which is + what the common parts actually do. + """ + a = int(round(alpha * fs)) + env = [] + for _ in range(repeats): + if sync: + env += [1] * a + [0] * (31 * a) # the sync pulse and its gap + for b in bits: + if fixed_gap is None: + high, low = (3 * a, a) if b == "1" else (a, 3 * a) + else: + high = 3 * a if b == "1" else a + low = int(round(fixed_gap * fs)) + env += [1] * high + [0] * low + env += [0] * int(gap * fs) + return _rf(env, fs, snr_db, seed) + + +def ook_ppm(bits, fs=50_000.0, alpha=400e-6, repeats=4, gap=10e-3, + snr_db=30.0, seed=2): + """Pulse-distance keying: the pulse is constant, the gap carries the bit.""" + a = int(round(alpha * fs)) + env = [] + for _ in range(repeats): + for b in bits: + env += [1] * a + [0] * (3 * a if b == "1" else a) + env += [0] * int(gap * fs) + return _rf(env, fs, snr_db, seed) + + +def ook_manchester(bits, fs=50_000.0, baud=2000.0, repeats=3, gap=10e-3, + snr_db=30.0, seed=3, preamble="10101010"): + half = int(round(fs / baud / 2)) + env = [] + for _ in range(repeats): + for b in preamble + bits: + first, second = (1, 0) if b == "1" else (0, 1) # IEEE 802.3 + env += [first] * half + [second] * half + env += [0] * int(gap * fs) + return _rf(env, fs, snr_db, seed) + + +def fsk_nrz(bits, fs=48_000.0, baud=1200.0, dev=4500.0, snr_db=30.0, seed=4): + """Two-level FSK holding each bit for one symbol period.""" + sp = fs / baud + n = int(round(len(bits) * sp)) + idx = np.clip((np.arange(n) / sp).astype(int), 0, len(bits) - 1) + level = np.array([1.0 if c == "1" else -1.0 for c in bits])[idx] + rng = np.random.default_rng(seed) + x = np.exp(1j * np.cumsum(2 * np.pi * level * dev / fs)) + nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) + return (x + nz * (rng.standard_normal(n) + + 1j * rng.standard_normal(n))).astype(np.complex64) + + +_C4FM = {"01": 3, "00": 1, "10": -1, "11": -3} + + +def c4fm(bits, fs=48_000.0, baud=4800.0, dev=1800.0, snr_db=30.0, seed=8): + """Four-level FSK, mapped the way P25 and DMR map it.""" + bits = bits[:len(bits) - len(bits) % 2] + level = np.array([_C4FM[bits[i:i + 2]] / 3.0 + for i in range(0, len(bits), 2)]) + sp = fs / baud + n = int(level.size * sp) + idx = np.clip((np.arange(n) / sp).astype(int), 0, level.size - 1) + rng = np.random.default_rng(seed) + x = np.exp(1j * np.cumsum(2 * np.pi * level[idx] * dev / fs)) + nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) + return (x + nz * (rng.standard_normal(n) + + 1j * rng.standard_normal(n))).astype(np.complex64) + + +def ax25_address(call, ssid, last=False): + call = (call.upper() + " ")[:6] + return bytes((ord(c) << 1) & 0xFE for c in call) + \ + bytes([0x60 | ((ssid & 0x0F) << 1) | (1 if last else 0)]) + + +def ax25_frame(source, dest, info, path=()): + """A complete AX.25 frame, with its frame-check sequence.""" + body = ax25_address(*dest) + ax25_address(*source, last=not path) + for i, hop in enumerate(path): + body += ax25_address(*hop, last=(i == len(path) - 1)) + body += bytes([0x03, 0xF0]) + info.encode() + crc = 0xFFFF + for byte in body: + crc ^= byte + for _ in range(8): + crc = (crc >> 1) ^ 0x8408 if crc & 1 else crc >> 1 + crc ^= 0xFFFF + return body + bytes([crc & 0xFF, (crc >> 8) & 0xFF]) + + +def aprs_afsk(frames, fs=48_000.0, baud=1200.0, snr_db=30.0, seed=6, + mark=1200.0, space=2200.0, dev=3000.0, flags=8): + """AX.25 over Bell 202 AFSK inside an FM carrier, as APRS is sent.""" + stream = "" + for frame in frames: + body = "01111110" * flags + ones = 0 + for byte in frame: + for k in range(8): # least significant first + bit = (byte >> k) & 1 + body += "1" if bit else "0" + if bit: + ones += 1 + if ones == 5: # stuff a zero after five + body += "0" + ones = 0 + else: + ones = 0 + stream += body + "01111110" * flags + level, nrzi = 1, [] + for bit in stream: + if bit == "0": + level ^= 1 # NRZI: a zero is a change + nrzi.append(level) + sp = fs / baud + n = int(len(nrzi) * sp) + idx = np.clip((np.arange(n) / sp).astype(int), 0, len(nrzi) - 1) + tone = np.where(np.array(nrzi)[idx] == 1, mark, space) + audio = np.sin(np.cumsum(2 * np.pi * tone / fs)) + rng = np.random.default_rng(seed) + x = np.exp(1j * np.cumsum(2 * np.pi * dev * audio / fs)) + nz = np.sqrt(1.0 / (2 * 10 ** (snr_db / 10.0))) + return (x + nz * (rng.standard_normal(n) + + 1j * rng.standard_normal(n))).astype(np.complex64) diff --git a/tests/test_decode.py b/tests/test_decode.py new file mode 100644 index 0000000..eb3a017 --- /dev/null +++ b/tests/test_decode.py @@ -0,0 +1,656 @@ +"""Reading the data out of a data signal. + +Two things have to be true of a decoder and they pull against each other: it +has to read a real packet correctly, and it has to refuse a signal that is not +a packet. The second is the harder one -- noise sliced at a threshold makes +runs, and runs make bits -- so about half of what is here is signals that must +come back with nothing. +""" +import sys +from pathlib import Path + +import numpy as np +import pytest + +sys.path.insert(0, str(Path(__file__).parent)) +import signals # noqa: E402 +from bandsaunter import decode as D # noqa: E402 +from bandsaunter.decode import (bits_to_hex, check_crc, decode_data, + decode_four_level, level_count, nrz_bits, + slice_fsk, slice_ook, split_bursts) +from bandsaunter.protocols import (POCSAG_BAUDS, SYNC_BITS, decode_ax25, + decode_pocsag, pocsag_bits, + pocsag_codeword) + +PAYLOAD = "101100100011010101001110" # 24 bits, the usual remote + + +def _random_bits(n, seed=7): + return "".join(np.random.default_rng(seed).choice(list("01"), n)) + + +def _contains(truth: str, got: str, window: int = 120) -> bool: + """True if a long stretch of the truth is in the decode, either polarity. + + Either polarity because nothing in an unframed stream says which level is + a one; a protocol's own sync word settles it, and these have none. + """ + flipped = got.translate(str.maketrans("01", "10")) + middle = truth[len(truth) // 4:len(truth) // 4 + window] + return middle in got or middle in flipped + + +# --------------------------------------------------------------------------- +# Slicing +# --------------------------------------------------------------------------- + +def test_on_off_keying_slices_into_runs(): + train = slice_ook(signals.ook_pwm(PAYLOAD), 50_000.0) + assert train is not None + assert len(train) > 50 + assert train.contrast_db > 10.0 + + +def test_a_signal_that_is_never_keyed_has_no_runs(): + """A steady carrier is not on-off keyed, whatever a threshold would do.""" + assert slice_ook(signals.make("carrier", n=32000), 32_000.0) is None + + +def test_two_level_fsk_slices_the_same_way_as_keying(): + """The point of the design: after slicing, FSK and OOK are one problem.""" + train = slice_fsk(signals.fsk_nrz(_random_bits(400)), 48_000.0, + baud_hint=1200.0) + assert train is not None + assert train.source == "fsk" + assert len(train) > 100 + + +def test_a_glitch_shorter_than_a_symbol_is_absorbed(): + """One sample the wrong side of the threshold must not become two runs.""" + levels = np.array([True, False, True, False, True], dtype=bool) + lengths = np.array([40, 1, 39, 80, 40], dtype=np.int64) + out_levels, out_lengths = D._despeckle(levels, lengths, minimum=4) + assert out_lengths.tolist() == [80, 80, 40] + assert out_levels.tolist() == [True, False, True] + + +def test_bursts_are_cut_at_the_silence_between_repeats(): + train = slice_ook(signals.ook_pwm(PAYLOAD, repeats=4), 50_000.0) + bursts = split_bursts(train) + assert len(bursts) == 4 + for burst in bursts: + # Trimmed to start and end on a pulse: the silence either side + # belongs to the gap between packets, not to the packet. + assert burst.levels[0] and burst.levels[-1] + + +def test_a_continuous_stream_is_never_cut_into_bursts(): + """Long runs of one tone are data, not the silence between packets.""" + train = slice_fsk(signals.fsk_nrz(_random_bits(600)), 48_000.0, + baud_hint=1200.0) + assert len(split_bursts(train)) == 1 + + +# --------------------------------------------------------------------------- +# The line codes +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("build,encoding,exact", [ + (lambda: signals.ook_pwm(PAYLOAD), "PWM", True), + (lambda: signals.ook_pwm(PAYLOAD, fixed_gap=350e-6), "PWM", True), + (lambda: signals.ook_ppm(PAYLOAD), "PPM", False), + (lambda: signals.ook_manchester(PAYLOAD), "Manchester", True), +]) +def test_each_line_code_is_recognised_and_read(build, encoding, exact): + got = decode_data(build(), 50_000.0, family="ook") + assert got.ok, got.note + assert got.encoding == encoding + if exact: + assert PAYLOAD in got.bits + else: + # A gap-length code loses its final bit: that gap ran into the + # silence before the next repeat and is no longer separable from it. + assert PAYLOAD[:-1] in got.bits + + +def test_a_pulse_width_code_keeps_its_last_bit(): + got = decode_data(signals.ook_pwm(PAYLOAD), 50_000.0, family="ook") + assert got.bits == PAYLOAD + + +def test_the_symbol_rate_is_measured_not_guessed(): + # 350 us units, four to a bit: 714 bits per second. + got = decode_data(signals.ook_pwm(PAYLOAD), 50_000.0, family="ook") + assert got.baud == pytest.approx(714, rel=0.05) + + +def test_manchester_reports_the_data_rate_not_the_cell_rate(): + """Two cells go out for every bit; the link is not twice as fast.""" + got = decode_data(signals.ook_manchester(PAYLOAD, baud=2000.0), + 50_000.0, family="ook") + assert got.baud == pytest.approx(2000, rel=0.06) + + +@pytest.mark.parametrize("baud,fs", [(512.0, 32_000.0), (1200.0, 48_000.0), + (2400.0, 48_000.0), (4800.0, 96_000.0)]) +def test_plain_nrz_comes_back_bit_for_bit(baud, fs): + truth = _random_bits(400, seed=int(baud) % 97) + got = decode_data(signals.fsk_nrz(truth, fs=fs, baud=baud), fs, + family="fsk", baud_hint=baud) + assert got.ok, got.note + assert got.baud == pytest.approx(baud, rel=0.02) + assert _contains(truth, got.bits), "the bits drifted" + + +def test_a_clock_a_shade_out_does_not_drift_across_a_long_frame(): + """Six hundred bits is where a quarter of a per cent of error shows up.""" + truth = _random_bits(600, seed=13) + got = decode_data(signals.fsk_nrz(truth, fs=50_000.0, baud=1200.0), + 50_000.0, family="fsk") + assert got.n_bits == pytest.approx(600, abs=2) + assert _contains(truth, got.bits, window=400) + + +# --------------------------------------------------------------------------- +# Repeats, which are what make a decode believable +# --------------------------------------------------------------------------- + +def test_repeats_are_counted_and_have_to_agree(): + got = decode_data(signals.ook_pwm(PAYLOAD, repeats=6), 50_000.0, + family="ook") + assert got.repeats >= 4 + assert got.agreement > 0.95 + assert got.confidence > 0.85 + + +def test_one_lonely_packet_is_believed_less_than_six(): + once = decode_data(signals.ook_pwm(PAYLOAD, repeats=1), 50_000.0, + family="ook") + often = decode_data(signals.ook_pwm(PAYLOAD, repeats=6), 50_000.0, + family="ook") + assert often.confidence > once.confidence + + +def test_a_packet_repeated_with_no_gap_is_still_found(): + """Some transmitters run their repeats together with nothing between.""" + bits, repeats = D._repeat_within(PAYLOAD * 4) + assert bits == PAYLOAD and repeats == 4 + + +def test_repeats_that_disagree_are_voted_on(): + consensus, certainty = D._agreement(["10110010", "10110010", "10110011"]) + assert consensus == "10110010" + assert 0.5 < certainty < 1.0 + + +# --------------------------------------------------------------------------- +# Refusing what is not data +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("kind", ["usb", "lsb", "nfm", "wfm", "am", "noise", + "carrier", "cw", "psk4"]) +@pytest.mark.parametrize("seed", [1, 3, 5]) +def test_signals_that_are_not_data_decode_to_nothing(kind, seed): + """The hard half: runs exist in anything, and bits follow from runs.""" + x = signals.make(kind, n=64000, fs=32_000.0, snr_db=25.0, seed=seed) + got = decode_data(x, 32_000.0) + assert not got.ok, f"{kind}: invented {got.summary()}" + assert got.note + + +def test_a_lucky_window_in_speech_is_not_a_packet(): + """One burst in eight fitting a grid is a coincidence, not a signal.""" + got = decode_data(signals.make("usb", n=64000, fs=32_000.0), 32_000.0) + assert not got.ok + assert "bursts" in got.note or "frames" in got.note or "nothing" in got.note + + +def test_a_bare_grid_fit_is_refused_without_framing(): + out = D.DataDecode(ok=True, encoding="NRZ", bits="1" * 40, repeats=1) + assert D._refuse(out, fit=0.6, share=1.0) + assert not D._refuse(out, fit=0.95, share=1.0) + + +def test_repetition_carries_a_decode_that_fit_alone_would_not(): + out = D.DataDecode(ok=True, encoding="NRZ", bits="1" * 40, repeats=5, + agreement=1.0) + assert not D._refuse(out, fit=0.2, share=0.1) + + +# --------------------------------------------------------------------------- +# Four levels +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("kind,levels", [("fsk2", 2), ("fsk4", 4), + ("nfm", 1), ("carrier", 1), + ("noise", 1)]) +def test_the_number_of_levels_is_counted_correctly(kind, levels): + x = signals.make(kind, n=64000, fs=32_000.0, snr_db=25.0) + assert level_count(D._frequency_of(x, 32_000.0)) == levels + + +def test_a_four_level_signal_is_never_read_as_two(): + """Slicing C4FM down the middle gives bits, and they mean nothing.""" + x = signals.make("fsk4", n=64000, fs=32_000.0, snr_db=25.0) + got = decode_data(x, 32_000.0) + assert got.encoding == "4-level FSK" + assert "no frame sync" in got.summary() + assert got.confidence < 0.6 + + +def test_a_frame_sync_word_names_the_system_that_sent_it(): + sync = "".join(f"{int(c, 16):04b}" for c in "5575F5FF77FF") + payload = "".join(sync + _random_bits(300, seed=i) for i in range(6)) + got = decode_four_level(signals.c4fm(payload), 48_000.0, 4800.0) + assert got is not None and got.ok + assert got.protocol == "P25 Phase 1" + assert got.baud == pytest.approx(4800, rel=0.02) + assert got.confidence > 0.85 + + +def test_an_on_off_keyed_burst_never_takes_the_four_level_path(): + """Discriminator noise in the silences would count as extra levels.""" + assert decode_four_level(signals.ook_pwm(PAYLOAD), 50_000.0) is None + + +# --------------------------------------------------------------------------- +# Checks and rendering +# --------------------------------------------------------------------------- + +def test_a_checksum_that_comes_out_right_is_reported(): + body = bytes([0x12, 0x34, 0x56]) + bits = "".join(f"{b:08b}" for b in body + bytes([sum(body) & 0xFF])) + assert "checksum-8" in check_crc(bits) + + +def test_a_crc16_that_comes_out_right_is_reported(): + from bandsaunter.decode import _crc16_ccitt + body = bytes([0xDE, 0xAD, 0xBE, 0xEF]) + crc = _crc16_ccitt(body) + bits = "".join(f"{b:08b}" for b in body + bytes([crc >> 8, crc & 0xFF])) + assert "CRC-16/CCITT" in check_crc(bits) + + +def test_a_packet_with_no_valid_check_claims_none(): + assert check_crc("0" * 32) == [] or "checksum" in check_crc("0" * 32)[0] + + +@pytest.mark.parametrize("bits,expected", [ + ("10110010", "B2"), ("1011001000110101", "B2 35"), ("", ""), + ("1011", "B0"), # a partial byte, padded on the right +]) +def test_bits_render_as_hex(bits, expected): + assert bits_to_hex(bits) == expected + + +def test_the_summary_says_what_matters_first(): + got = decode_data(signals.ook_pwm(PAYLOAD), 50_000.0, family="ook") + assert "EV1527" in got.summary() + assert "24 bits" in got.summary() + + +def test_a_twenty_four_bit_pulse_width_packet_is_named(): + got = decode_data(signals.ook_pwm(PAYLOAD), 50_000.0, family="ook") + assert got.protocol == "EV1527 / PT2262-style remote" + + +# --------------------------------------------------------------------------- +# POCSAG +# --------------------------------------------------------------------------- + +def test_a_pocsag_codeword_checks_out(): + word = pocsag_codeword(0x0ABCD) + assert D and word & 1 in (0, 1) + from bandsaunter.protocols import _bch_syndrome, _parity_ok + assert _bch_syndrome(word) == 0 and _parity_ok(word) + + +def test_a_single_bit_error_in_a_codeword_is_corrected(): + from bandsaunter.protocols import _correct + word = pocsag_codeword(0x15555) + broken = word ^ (1 << 17) + fixed, ok = _correct(broken) + assert ok and fixed == word + + +def test_a_pocsag_transmission_reads_back_as_the_pages_that_went_in(): + pages = [(1234568, 3, "ENGINE 4 RESPOND"), (98765, 0, "CALL EXT 4412")] + bits = pocsag_bits(pages) + assert SYNC_BITS in bits + got = decode_pocsag(signals.fsk_nrz(bits, fs=48_000.0, baud=1200.0), + 48_000.0) + assert got is not None and got.ok + assert got.protocol == "POCSAG 1200" + assert got.messages[0] == "[1234568D] ENGINE 4 RESPOND" + assert got.messages[1] == "[0098765A] CALL EXT 4412" + assert got.checks == ["BCH(31,21)"] + + +@pytest.mark.parametrize("baud", POCSAG_BAUDS) +def test_every_pocsag_rate_is_found_without_being_told(baud): + """Nothing in the signal announces the rate, so all three are tried.""" + bits = pocsag_bits([(2097151, 0, "HELLO")]) # the top address + fs = max(32_000.0, baud * 20) + got = decode_pocsag(signals.fsk_nrz(bits, fs=fs, baud=baud), fs) + assert got is not None and got.ok + assert got.protocol == f"POCSAG {baud:.0f}" + assert "HELLO" in got.messages[0] + + +def test_pocsag_survives_being_received_upside_down(): + """Which tone is a one is a property of the receiver, not the standard.""" + bits = pocsag_bits([(1234568, 3, "INVERTED")]) + flipped = bits.translate(str.maketrans("01", "10")) + got = decode_pocsag(signals.fsk_nrz(flipped, fs=48_000.0, baud=1200.0), + 48_000.0) + assert got is not None and "INVERTED" in got.messages[0] + + +def test_the_same_page_arriving_twice_is_reported_once(): + bits = pocsag_bits([(1234568, 3, "ONCE ONLY")]) * 3 + got = decode_pocsag(signals.fsk_nrz(bits, fs=48_000.0, baud=1200.0), + 48_000.0) + assert got is not None + assert len(got.messages) == 1 + + +def test_a_numeric_page_is_read_as_digits(): + bits = pocsag_bits([(1234568, 0, "5551234")]) + got = decode_pocsag(signals.fsk_nrz(bits, fs=48_000.0, baud=1200.0), + 48_000.0) + assert got is not None and got.ok + assert "5551234" in got.messages[0] + + +def test_an_address_that_will_not_fit_is_refused_rather_than_truncated(): + """A silently mangled address is a page delivered to somebody else.""" + from bandsaunter.protocols import MAX_ADDRESS + with pytest.raises(ValueError): + pocsag_bits([(MAX_ADDRESS + 1, 0, "NOPE")]) + + +def test_something_that_is_not_pocsag_is_refused(): + assert decode_pocsag(signals.fsk_nrz(_random_bits(600), fs=48_000.0), + 48_000.0) is None + + +# --------------------------------------------------------------------------- +# AX.25 and APRS +# --------------------------------------------------------------------------- + +def test_an_aprs_frame_reads_back_with_its_callsign_and_payload(): + frame = signals.ax25_frame(("W1AW", 0), ("APRS", 0), + "!4142.45N/07243.63W-Newington") + got = decode_ax25(signals.aprs_afsk([frame]), 48_000.0) + assert got is not None and got.ok + assert got.protocol == "AX.25 / APRS" + assert got.messages[0].startswith("W1AW>APRS") + assert "Newington" in got.messages[0] + assert got.checks == ["FCS (CRC-16/X.25)"] + + +def test_a_digipeater_path_is_kept(): + frame = signals.ax25_frame(("KU0W", 9), ("APZ001", 0), ">testing", + path=[("WIDE1", 1), ("WIDE2", 2)]) + got = decode_ax25(signals.aprs_afsk([frame]), 48_000.0) + assert got is not None + assert "KU0W-9" in got.messages[0] + assert "WIDE1-1" in got.messages[0] and "WIDE2-2" in got.messages[0] + + +def test_several_frames_in_one_capture_all_come_back(): + frames = [signals.ax25_frame(("W1AW", 0), ("APRS", 0), "first"), + signals.ax25_frame(("KU0W", 0), ("APRS", 0), "second")] + got = decode_ax25(signals.aprs_afsk(frames), 48_000.0) + assert got is not None and len(got.messages) == 2 + + +def test_a_frame_whose_checksum_is_wrong_is_thrown_away(): + """The frame check is the whole reason to believe an AX.25 decode.""" + frame = bytearray(signals.ax25_frame(("W1AW", 0), ("APRS", 0), "corrupt")) + frame[-1] ^= 0xFF + got = decode_ax25(signals.aprs_afsk([bytes(frame)]), 48_000.0) + assert got is None + + +def test_noise_is_not_a_packet_frame(): + assert decode_ax25(signals.make("noise", n=64000, fs=48_000.0), + 48_000.0) is None + + +# --------------------------------------------------------------------------- +# Robustness +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("snr_db", [30.0, 20.0, 12.0]) +def test_a_remote_still_decodes_as_the_signal_weakens(snr_db): + got = decode_data(signals.ook_pwm(PAYLOAD, snr_db=snr_db, repeats=6), + 50_000.0, family="ook") + assert got.ok, f"{snr_db} dB: {got.note}" + assert PAYLOAD in got.bits + + +@pytest.mark.parametrize("snr_db", [30.0, 18.0]) +def test_paging_still_decodes_as_the_signal_weakens(snr_db): + bits = pocsag_bits([(1234568, 3, "WEAK SIGNAL")]) + got = decode_pocsag(signals.fsk_nrz(bits, fs=48_000.0, baud=1200.0, + snr_db=snr_db), 48_000.0) + assert got is not None and "WEAK" in got.messages[0] + + +def test_a_capture_too_short_to_hold_a_packet_says_so(): + got = decode_data(np.zeros(100, dtype=np.complex64), 48_000.0) + assert not got.ok and "short" in got.note + + +def test_nrz_bits_refuses_a_rate_it_cannot_resolve(): + """Fewer than two samples a symbol is not a sampling problem to solve.""" + train = slice_ook(signals.ook_pwm(PAYLOAD), 50_000.0) + assert nrz_bits(train, 40_000.0) == "" + assert nrz_bits(train, 0.0) == "" + + +# --------------------------------------------------------------------------- +# A scan that finds one +# --------------------------------------------------------------------------- + +def _scan(tmp_path, transmitters, **over): + from bandsaunter.config import ScanConfig + from bandsaunter.ranges import parse_range_list + from bandsaunter.scanner import Scanner, ScannerCallbacks + from bandsaunter.simulator import SimulatedDevice + + cfg = ScanConfig(ranges=parse_range_list(over.pop("ranges", "433.9M-433.95M")), + output_dir=str(tmp_path), record_seconds=3.0, + hang_seconds=1.0, threshold_db=12, dwell_seconds=0.05, + max_cycles=2, revisit_seconds=0.2) + on_status = over.pop("_on_status", None) + for key, value in over.items(): + setattr(cfg, key, value) + hits = [] + scanner = Scanner(cfg, device=SimulatedDevice( + transmitters=transmitters).open(), + callbacks=ScannerCallbacks(on_record_end=hits.append, + on_status=on_status)) + scanner.prepare() + scanner.run() + return scanner, [h for h in hits if h.kept] + + +def _remote(): + from bandsaunter.simulator import VirtualTransmitter as V + return [V(433_920_000, "packet", 0.45, 40_000, "remote", baud=2000, + payload=PAYLOAD, repeats=6)] + + +def _pager(): + from bandsaunter.simulator import VirtualTransmitter as V + return [V(929_612_500, "pocsag", 0.45, 12_500, "pager", baud=1200, + deviation=4_500, + pages=((1234568, 3, "ENGINE 4 RESPOND"),))] + + +def test_a_scan_reads_the_packet_off_a_remote(tmp_path): + _, hits = _scan(tmp_path, _remote()) + assert hits, "the remote was never captured" + assert all(h.data_encoding == "PWM" for h in hits) + read = [h for h in hits if h.data_bits == PAYLOAD] + assert read, [h.data_bits for h in hits] + assert read[0].data_repeats > 1 + assert read[0].classification == "EV1527 / PT2262-style remote" + + +def test_what_was_decoded_is_written_beside_the_recording(tmp_path): + _, hits = _scan(tmp_path, _remote()) + assert hits + written = list(tmp_path.glob("*_data.txt")) + assert written, "nothing was written" + bodies = [path.read_text() for path in written] + assert any(PAYLOAD in body for body in bodies), bodies + assert any("EV1527" in body for body in bodies) + # and every record points at the file it wrote + for hit in hits: + assert Path(hit.data_path).exists() + + +def test_a_scan_reads_a_page_and_prints_the_message(tmp_path): + _, hits = _scan(tmp_path, _pager(), ranges="929.55M-929.7M", + record_seconds=4.0) + assert hits, "the pager was never captured" + hit = hits[0] + assert hit.data_protocol == "POCSAG 1200" + assert "ENGINE 4 RESPOND" in hit.data_messages[0] + assert hit.classification == "POCSAG 1200" + assert "BCH(31,21)" in hit.data_checks + + +def test_the_display_is_told_what_was_decoded(tmp_path): + said = [] + _scan(tmp_path, _remote(), _on_status=said.append) + assert any("decoded" in m and PAYLOAD[:8] not in m for m in said), said + + +def test_decoding_can_be_turned_off(tmp_path): + _, hits = _scan(tmp_path, _remote(), decode_data=False) + assert hits + assert not hits[0].data_encoding + assert not list(tmp_path.glob("*_data.txt")) + + +def test_no_data_file_is_left_beside_a_recording_that_was_thrown_away(tmp_path): + """The capture is renamed after it is kept, and orphans confuse a reader.""" + _, hits = _scan(tmp_path, _remote()) + for path in tmp_path.glob("*_data.txt"): + stem = path.name[:-len("_data.txt")] + assert (tmp_path / f"{stem}.wav").exists(), f"orphan: {path.name}" + + +def test_a_decoded_packet_is_kept_even_when_the_content_check_says_no(tmp_path): + """A frame whose own checksum came out right is not a statistic.""" + _, hits = _scan(tmp_path, _pager(), ranges="929.55M-929.7M", + record_seconds=4.0, accept=["voice"]) + assert hits, "a decoded page was discarded as contentless" + assert hits[0].category == "digital" + + +def test_the_browser_shows_what_was_decoded(tmp_path): + from rich.console import Console + from bandsaunter.browse import Browser, Player + _scan(tmp_path, _pager(), ranges="929.55M-929.7M", record_seconds=4.0) + console = Console(width=100, height=30, force_terminal=True) + browser = Browser(tmp_path, console=console, player=Player([])) + assert browser.captures + cap = browser.captures[0] + assert cap.decoded and "ENGINE 4 RESPOND" in cap.decoded[0] + assert "POCSAG" in cap.data_headline + with console.capture() as frame: + console.print(browser.render()) + text = frame.get() + assert "ENGINE 4 RESPOND" in text + assert "decoded" in text + + +def test_a_pager_message_can_be_searched_for(tmp_path): + from rich.console import Console + from bandsaunter.browse import Browser, Player + _scan(tmp_path, _pager(), ranges="929.55M-929.7M", record_seconds=4.0) + browser = Browser(tmp_path, + console=Console(width=100, height=30, + force_terminal=True), + player=Player([])) + browser.query = "engine 4" + browser.apply() + assert browser.view, "searching what a data capture said found nothing" + + +# --------------------------------------------------------------------------- +# Text that came off the air +# --------------------------------------------------------------------------- + +HOSTILE = "[1234568D] ALERT [/red] see [bold] the thing" + + +def test_a_decoded_message_cannot_break_the_live_display(): + """Rich reads square brackets as markup, and a page is arbitrary text.""" + import os + import tempfile + import time + from rich.console import Console + from bandsaunter.config import ScanConfig + from bandsaunter.ranges import parse_range_list + from bandsaunter.recorder import HitRecord + from bandsaunter.scanner import Scanner + from bandsaunter.simulator import SimulatedDevice + from bandsaunter.ui import ScanDisplay + + console = Console(width=120, height=30, record=True, + file=open(os.devnull, "w")) + cfg = ScanConfig(ranges=parse_range_list("144M-148M"), + output_dir=tempfile.mkdtemp()) + scanner = Scanner(cfg, device=SimulatedDevice().open()) + scanner.prepare() + display = ScanDisplay(scanner, console=console) + hit = HitRecord(frequency=929.6e6, started_at=time.time(), duration=4.5, + snr_db=49.5, classification="POCSAG 1200") + hit.data_messages = [HOSTILE] + hit.data_protocol = "POCSAG 1200" + display.hits.appendleft(hit) + display.on_status(f"decoded 929.6 MHz: {HOSTILE}") + console.print(display._hits_table()) + console.print(display._footer()) + text = console.export_text() + assert "ALERT" in text + + +def test_a_decoded_message_cannot_break_the_line_per_hit_output(): + import os + from rich.console import Console + from bandsaunter.recorder import HitRecord + from bandsaunter.ui import print_hit + + console = Console(width=140, record=True, file=open(os.devnull, "w")) + hit = HitRecord(frequency=929.6e6, started_at=0, duration=4.5, + snr_db=49.5, classification="POCSAG 1200") + hit.kept = True + hit.data_messages = [HOSTILE] + print_hit(console, hit) + assert "ALERT" in console.export_text() + + +@pytest.mark.parametrize("typed", ["[/x]", "[bold", "]]]", "[/]"]) +def test_what_is_typed_at_the_search_prompt_cannot_break_the_browser(typed, + tmp_path): + """Typing "[/" used to end the session with a MarkupError.""" + from rich.console import Console + from bandsaunter.browse import Browser, Player + + browser = Browser(tmp_path, + console=Console(width=100, height=30, + force_terminal=True), + player=Player([])) + browser.searching = True + browser.query = typed + browser._footer() + browser.searching = False + browser.message = f"nothing matches {typed}" + browser._footer()