From 2b653c2c3ec105f84e736f7d4e09c97954f88ef3 Mon Sep 17 00:00:00 2001 From: The Dust Council Date: Sun, 20 Sep 2026 19:36:30 -0700 Subject: [PATCH] Read the APRS channel: who is out there, and what they said A third section, alongside the aircraft and the weather sensors, and for the same reason as both: a scan stops on a signal, records it and moves on, while APRS is a two-second transmission every few minutes from a hundred stations sharing one frequency. A sweep catches whichever happened to key up as it passed. `bandsaunter aprs` parks on the channel and catches all of them; `bandsaunter packets` reads a log back. Four layers, three of them new. The link layer was already here, opportunistically, in the generic decoder -- a correlator, NRZI, HDLC and a checksum, run on whatever a scan happened to record. It is now a receiver. What had to change is the state that survives a block boundary: the tail of the audio so the correlators see no edge, the phase of the sampling loop so a bit is not lost where one block meets the next, the tone the line was last at, and the bits themselves. A packet is most of a second and a block is about one, so frames straddling the boundary are not an edge case, they are most of them. Above that, the APRS information field, which is not one format but about twenty, chosen by the first character and accreted over thirty years. Positions uncompressed and compressed; Mic-E, which every Kenwood and Yaesu mobile sends and which hides the latitude inside the destination callsign because in 1995 those six bytes were carrying the word "APRS" and nothing else; weather with a position and without; messages, acknowledgements, rejections and bulletins; objects and items; status; telemetry; third-party traffic, credited to whoever originally sent it rather than to the gateway. Course and speed, altitude, power and antenna height, range and the precision extension, all of which ride in the comment. Every one has a writer beside its reader, so a packet goes in and the same packet comes out. Above that the section: a registry of who is out there and what each last said of each kind, distances and bearings from --at, a log keeping the whole frame under whatever was made of it, a spreadsheet, a map, and a channel full of stations that are not there for --simulate. One rule is worth naming because it is the difference between a decoder and a liar. A packet whose format does not match what its first character promised comes back as unparsed with its text intact. Thirteen characters of a *malformed* uncompressed position are perfectly good base-91, so trying one format and falling back to the other does not fail on a bad packet -- it succeeds, as a confident and completely different place, usually a thousand miles away. The specification makes the two unambiguous, a leading digit always meaning uncompressed, and the rule is read rather than guessed at. Two faults found by building it, both by measurement rather than by reading the code again. The framer handed back frames it had already reported, because it trimmed its buffer to before them rather than after -- every packet counted twice, for ever, which only shows up once the same signal is read across more than one block. And the invented channel truncated a transmission at the end of the block it began in rather than carrying the remainder over, which was invisible for as long as the simulated clock advanced in exact seconds and put every transmission at a block boundary; the moment it was paced against a real clock, nothing decoded at all. 204 new tests against seven deliberately broken builds, one of which survived until a test was written for the case it actually breaks. Full suite 2597 passed. Built as 2026-09-20_02. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg --- INSTALL.md | 16 +- README.md | 225 +++++++- bandsaunter/__init__.py | 2 +- bandsaunter/aprs.py | 1078 ++++++++++++++++++++++++++++++++++ bandsaunter/aprslog.py | 345 +++++++++++ bandsaunter/ax25.py | 524 +++++++++++++++++ bandsaunter/cli.py | 187 ++++++ bandsaunter/packets.py | 1213 +++++++++++++++++++++++++++++++++++++++ bandsaunter/tui.py | 180 +++++- bandsaunter/ui.py | 130 ++++- packaging/bandsaunter.1 | 248 +++++++- packaging/make-man.py | 140 +++++ tests/test_aprs.py | 629 ++++++++++++++++++++ tests/test_ax25.py | 260 +++++++++ tests/test_manpage.py | 18 +- tests/test_packets.py | 365 ++++++++++++ 16 files changed, 5543 insertions(+), 17 deletions(-) create mode 100644 bandsaunter/aprs.py create mode 100644 bandsaunter/aprslog.py create mode 100644 bandsaunter/ax25.py create mode 100644 bandsaunter/packets.py create mode 100644 tests/test_aprs.py create mode 100644 tests/test_ax25.py create mode 100644 tests/test_packets.py diff --git a/INSTALL.md b/INSTALL.md index e3b047a..dc4f69f 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -10,9 +10,10 @@ If you are on Debian, Ubuntu or Mint, [build the package](#a-debian-ubuntu-mint- **You can try the whole program before buying or plugging in a receiver.** `--simulate` runs the scanner against a synthetic band, `bandsaunter adsb ---simulate` flies imaginary aircraft past an imaginary receiver, and -`bandsaunter weather --simulate` puts six weather sensors on a fence that does -not exist. None of the three needs +--simulate` flies imaginary aircraft past an imaginary receiver, `bandsaunter +weather --simulate` puts six weather sensors on a fence that does not exist, +and `bandsaunter aprs --simulate` fills a channel with amateur stations that +are not there. None of the four needs hardware or a network. --- @@ -120,6 +121,8 @@ bandsaunter adsb --simulate --seconds 30 bandsaunter flights # draws what the last command heard bandsaunter weather --simulate --seconds 60 bandsaunter readings --csv # turns what it heard into a spreadsheet +bandsaunter aprs --simulate --seconds 120 +bandsaunter packets --kml # turns what it heard into a map ``` With a receiver plugged in: @@ -229,6 +232,13 @@ read the output. `--save-iq FILE` keeps the raw samples (2 MB a second, so bound it with `--seconds 60`) and `--from-iq FILE` reads one back, so a recording made where the aerial is can be worked on anywhere. +**APRS needs nothing extra either** — no network, no key, no package beyond +the table above. The only thing that decides whether you hear it is the aerial +and whether a digipeater is in range: a quarter-wave whip for 144 MHz is 49 cm, +which is longer than the one most dongles ship with. Check `--region` before +anything else, because on the wrong channel there is silence rather than a bad +signal. `bandsaunter aprs --simulate` runs the whole thing without an aerial. + **Aircraft and callsign lookups need no installation**, only a network. They ask public registers about a callsign or a 24-bit address and cache the answers for a month; `--no-lookup` turns them off, and what the address and diff --git a/README.md b/README.md index 8b370ca..8abe96d 100644 --- a/README.md +++ b/README.md @@ -7,10 +7,11 @@ built-in US band plan — and it sweeps them, stops on anything above the noise floor, records it, and works out what kind of signal it was. CW/Morse is decoded to text. -Two things get sections of their own, because neither fits through a scan: -[the aircraft overhead](#aircraft) on 1090 MHz, drawn on a moving map, and +Three things get sections of their own, because none of them fits through a +scan: [the aircraft overhead](#aircraft) on 1090 MHz, drawn on a moving map; [the weather sensors](#weather-sensors-on-433-mhz) on 433 MHz, which you can -give names to as they arrive. +give names to as they arrive; and [APRS](#aprs-on-144-mhz) on 144 MHz, where +amateur stations report where they are and talk to each other. Two programs: `bandsaunter` scans, and [`saunterbrowse`](#browsing-what-you-recorded) reads back what it collected — @@ -247,6 +248,7 @@ bandsaunter scan -r 144M-148M -r 420M-450M # your own ranges bandsaunter scan -b 2m --simulate # try it without hardware bandsaunter adsb --window # aircraft overhead, on a map bandsaunter weather # the weather sensors on 433 MHz +bandsaunter aprs # amateur packet on 144.39 MHz ``` ## Two ways to drive it @@ -262,15 +264,19 @@ table of settings, so neither can offer something the other cannot. 4 Saved settings and profiles 5 Aircraft (ADS-B) listen on 1090 MHz, draw where they went 6 Weather sensors listen on 433 MHz, name what is out there + 7 APRS (144 MHz packet) positions, weather and messages from amateurs h Help s Start scanning q Quit ``` -Items 5 and 6 are sections of their own, with their own options and their own -menus, because neither fits through a scan: ADS-B is a megabit a second and a -scan channel is 12.5 kHz wide, and a weather sensor message is a burst of a -carrier switched on and off that a scan would record as clicks. +Items 5, 6 and 7 are sections of their own, with their own options and their +own menus, because none of them fits through a scan: ADS-B is a megabit a +second and a scan channel is 12.5 kHz wide; a weather sensor message is a burst +of a carrier switched on and off that a scan would record as clicks; and APRS +is a two-second transmission every few minutes from a hundred stations sharing +one frequency, of which a sweep would catch whichever happened to key up as it +passed. Settings are grouped, show their current value against the built-in default, and carry their own help: @@ -2250,6 +2256,211 @@ decoder problem apart: a capture that yields nothing on replay yields nothing for anybody, and a capture that yields readings on replay but not on the air is a setting. +## APRS on 144 MHz + +One channel, one frequency, everybody. 144.390 MHz across North America and a +different number in every other region, carrying position reports, weather, +messages, objects and telemetry from every amateur station within earshot — +and from every hilltop digipeater repeating them onward, which is most of what +you will actually hear. + +```bash +bandsaunter aprs listen, and show who is out there +bandsaunter aprs --region europe ...on 144.800 instead +bandsaunter aprs --simulate a channel full of stations that are not there +bandsaunter packets --kml turn a log into something Google Earth opens +``` + +It is a section of its own for the same reason the other two are: a scan stops +on a signal, records it and moves on, and APRS is a two-second transmission +every few minutes from a hundred stations sharing one frequency. A sweep +catches whichever one happened to key up while the sweep was pointed there. + +Unlike the others this is a conversation rather than a broadcast. Stations +address each other, acknowledge each other and relay for each other, so what +is worth showing is not only who is out there but what was said. And nothing +here has to be named: a weather sensor broadcasts a number out of a hat, but +an APRS station broadcasts a callsign issued by a government, which is already +the name. + +``` +╭──────────────────────────────────────────────────────────────────────────────────────────────╮ +│ 144.39 MHz 6 stations 12 seen 148 packets 9 messages 41:02 aprs_2026-09-20.jsonl │ +╰──────────────────────────────────────────────────────────────────────────────────────────────╯ +station what said away signal pkts ago +KU0W-9 car 47.5570N 122.2783W ENE 54 km/h mobile 2 km ENE 38 dB 41 0:08 +KB1XYZ weather station 47.5900N 122.2700W temperature 15.0 C 5 km NNE 34 dB 12 0:22 + humidity 62% pressure 1013.2 hPa +W1AW-1 digipeater 47.6062N 122.3322W Seattle wide digi 7 km NNW 37 dB 4 1:14 +EVENT1 fire 47.5900N 122.3300W 5 km NNW 36 dB 1 3:40 +``` + +### Which channel + +The frequency is agreed between amateurs rather than allocated, so it differs +by region and **there is no way to discover it from the air**: on the wrong one +you hear silence, not a bad signal. `--region` covers the common ones — +`north-america` (144.390), `europe` (144.800), `australia` (145.175), `japan`, +`brazil`, `thailand` — and `--frequency` takes a number for anything else. + +### What it reads + +The information field of an APRS packet is not one format. It is about twenty, +chosen by the first character, accreted over thirty years, and several of them +exist only because a particular radio shipped with them. + +| what | how it arrives | +|---|---| +| Position | uncompressed, or compressed into thirteen characters of base-91 | +| Mic-E | every Kenwood and Yaesu mobile, with half the position hidden in the destination callsign | +| Weather | attached to a position, or positionless | +| Messages | to a station, acknowledged, rejected, or broadcast as a bulletin | +| Objects and items | something placed on the map that is not the station placing it | +| Status | what the operator typed | +| Telemetry | five analogue channels and eight bits | +| Third-party traffic | a packet relayed in from another network, credited to whoever originally sent it | + +Riding in the comment: course and speed, altitude, transmitter power and +antenna height, pre-computed range, direction-finding reports, and the +precision extension. + +**Mic-E deserves a note**, because it is a quarter of everything on the channel +and it is the least readable thing in amateur radio. In 1995 the destination +address of an APRS frame carried nothing but the word "APRS", and somebody +noticed that six bytes is exactly enough for a latitude. So a Mic-E packet puts +the latitude, the north/south bit, the east/west bit, a hundred degrees of +longitude and a three-bit status message into *the callsign it is addressed +to*, and the rest into the information field as characters chosen so the whole +thing survives being typed into a logbook. It is also the reason APRS fits in a +two-second transmission. + +### Refusing to guess + +A packet whose format does not match what its first character promised comes +back as unparsed, with its text kept, rather than as a position. + +That sounds like a nicety and is not. Thirteen characters of a *malformed* +uncompressed position are perfectly good base-91, so a decoder that tries one +format and falls back to the other does not fail on a bad packet — it succeeds, +as a confident and completely different place, usually a thousand miles away. +The specification makes the two unambiguous (a leading digit always means +uncompressed, which is why the compressed format writes a numeric overlay as a +letter), so the rule is read rather than guessed at. + +`--unparsed` is on by default: a packet that defeats every parser here still +arrived, still came from a real station, and showing it is how the next format +gets added. + +### Off the air + +Four things happen between the aerial and a frame. + +**The tones become a soft symbol.** APRS is Bell 202 — 1200 Hz for a mark, +2200 Hz for a space, twelve hundred a second, inside an ordinary FM +transmission. Two correlators, one at each tone, and the difference between +them. A correlator rather than a frequency discriminator because the tones are +less than an octave apart and radio audio is distorted enough that +instantaneous frequency wanders badly. + +De-emphasis is turned off, which matters and is easy to miss: voice FM lifts +the high frequencies on transmit and drops them again on receive, and packet +radio takes its audio from the discriminator *before* that happens. Dropping +the highs would take four decibels off the 2200 Hz tone and leave the 1200 Hz +one alone, which is exactly the difference being measured. + +**The soft symbol becomes bits**, sampled once a bit at an instant held in the +middle of the bit by a loop nudged at every zero crossing. The loop carries its +phase from one block of audio to the next, which is what makes this a receiver +rather than a decoder of recordings — a packet is most of a second and a block +is about one, so frames straddling the boundary are not an edge case, they are +most of them. + +**The bits become a frame.** NRZI first, where a zero is a change of tone and a +one is no change — which makes the whole thing immune to being wired up +backwards, since inverting the stream decodes to identical data. Then HDLC: +frames delimited by 01111110, with a zero stuffed after every five ones so the +flag cannot occur inside one. + +**The frame is believed or it is not.** Sixteen bits of CRC, and nothing +without a correct one is reported. That is what makes it safe to leave running +for hours with the squelch open: a frame either checks out or it never existed. +A frame whose callsigns are unprintable is refused as well — on a band this +busy, sixteen bits is strong but not infinite. + +It decodes down to about 8 dB of signal-to-noise and finds nothing at all in +pure noise. + +### Afterwards + +Three tables, because they answer different questions. **Stations heard** is +about the band and the aerial: where each was, how far off, how many packets, +how many of those arrived *directly* rather than through a digipeater, and how +strongly. **What they said** is the weather, the speeds and the status lines. +**What passed between them** is the messages, in order — the only part of APRS +that is a conversation. + +`--at LAT,LON` turns on the distance and bearing columns. `--direct-only` +leaves out anything that reached you through a digipeater, which is a much +shorter list and is the honest measure of what your aerial can actually reach. + +`--csv` writes a row per packet with position, speed and weather in their own +columns. `--kml` writes a pin where each station was last heard and a line for +anything that moved. Both can be made later from a log with `bandsaunter +packets --csv --kml`, and `--station CALL` narrows either to one callsign. + +The log keeps the whole AX.25 frame in hexadecimal under whatever was made of +it. That is not a hypothetical precaution: the list of APRS formats is still +growing, so a packet this version cannot read is on the disk in full for a +version that can. + +### Every APRS option + +| option | flag | default | what it does | +|---|---|---|---| +| Receiver | `--device` | 0 | which receiver, when more than one is plugged in | +| Gain | `--gain` | auto | tuner gain in dB, or automatic | +| Sample rate | `--rate` | 240 kS/s | 96 kS/s is the least that holds the channel | +| Region | `--region` | north-america | which APRS channel to listen on | +| Listen on | `--frequency`, `--freq` | 144.390 MHz | the exact frequency, if the region's is not what you want | +| Invent a channel | `--simulate` / `--no-simulate` | no | stations that are not there | +| Listen for | `--seconds` | until stopped | how long before stopping | +| Write a log | `--log` / `--no-log` | yes | one line of JSON per packet | +| Print every packet | `--packets` / `--no-packets` | no | a stream of lines instead of a table | +| Keep on screen for | `--hold` | 3600 s | how long a station stays after its last packet | +| Receiver at | `--at` | blank | where the aerial is, for distances | +| Show readings in | `--units` | metric | metric or imperial, for the display and the export | +| Show what cannot be read | `--unparsed` / `--no-unparsed` | yes | list packets in unknown formats | +| Include relayed packets | `--digipeated` / `--direct-only` | yes | count what reached you through a digipeater | +| Report at the end | `--report` / `--no-report` | yes | print what was heard | +| Also write a spreadsheet | `--csv` / `--no-csv` | no | CSV beside the log | +| Also write a map | `--kml` / `--no-kml` | no | KML beside the log | + +Saved in `aprs.yaml` beside the other settings, from the menu's **s** or by +hand. + +### Without an aerial + +`--simulate` puts a dozen stations on a channel that does not exist — cars +moving, a weather station, a digipeater, objects, messages passing between +them — keyed as real Bell 202 audio on a real FM carrier, through the real +demodulator and the real parsers. Nothing touches the receiver. + +### If nothing is heard + +**Check the region first.** It is the one fault that looks like a dead aerial +and is not: on the wrong channel there is silence rather than a bad signal. + +Then give it time. A fixed station beacons every twenty or thirty minutes and a +mobile every minute or two, so five minutes of an ordinary suburb might be +three packets and an hour is a fair picture. If you are somewhere without a +digipeater in range you may genuinely hear nothing — APRS coverage is built +from volunteers' hilltops, and it has holes. + +A quarter-wave whip for 144 MHz is 49 cm, which is longer than the aerial most +dongles ship with; the stock telescopic one extended properly does well. +`--packets` shows each frame as it arrives, which is the thing to watch while +moving an aerial about. + ## Meters on 900 MHz A scan of 902–928 MHz that turns up a burst gets it named rather than reported diff --git a/bandsaunter/__init__.py b/bandsaunter/__init__.py index 20d2ead..03490ae 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-09-20" -VERSION_REVISION = 1 +VERSION_REVISION = 2 __version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}" diff --git a/bandsaunter/aprs.py b/bandsaunter/aprs.py new file mode 100644 index 0000000..932ac2c --- /dev/null +++ b/bandsaunter/aprs.py @@ -0,0 +1,1078 @@ +"""Listening to APRS, from either front end. + +One channel, one frequency, everybody: 144.390 MHz in North America and a +different number in every other region, carrying position reports, weather, +messages, objects and telemetry from every amateur station within earshot and +every digipeater repeating them onward. Unlike the aircraft and the weather +sensors, this is a conversation rather than a broadcast -- stations address +each other, acknowledge each other, and relay for each other -- so what is +worth showing is not only who is out there but what has been said. + +It has a channel to itself for the same reason the other two modes do: a scan +stops on a signal, records it and moves on, and APRS is a two-second +transmission every few minutes from a hundred different stations. A sweep +catches whichever one happened to key up while the sweep was pointed there. +Parking on the channel catches all of them. + +Nothing here has to be named. A weather sensor broadcasts a number out of a +hat and needs a person to say which shed it is on; an APRS station broadcasts +its callsign, which is issued by a government and is already the name. That +is the one way this section is simpler than the last one. +""" + +from __future__ import annotations + +import math +import time +from dataclasses import asdict, dataclass, field +from datetime import datetime +from pathlib import Path + +import yaml + +from . import ax25, packets +from .settings import Setting, format_value + +__all__ = ["AprsOptions", "OPTIONS", "OPTION_GROUPS", "defaults", "in_group", + "by_key", "format_option", "describe", "summarise", "Station", + "Net", "Heard", "listen", "open_device", "open_log", "pump", + "finish", "report", "load_options", "save_options", "options_path", + "logs_in", "channel_named", "distance_km", "bearing_deg"] + + +# --------------------------------------------------------------------------- +# The options +# --------------------------------------------------------------------------- + +@dataclass +class AprsOptions: + """Everything the APRS mode can be told, in one place.""" + + # -- receiver ------------------------------------------------------- + device: int = 0 + gain: str = "auto" + rate: float = 240_000.0 + frequency: float = ax25.APRS_HZ + region: str = "north-america" + simulate: bool = False + + # -- listening ------------------------------------------------------ + seconds: float = 0.0 + log: bool = True + packets_seen: bool = False # a line per packet rather than a table + hold: float = 3600.0 + location: str = "" # where the aerial is, for distances + + # -- what to show --------------------------------------------------- + units: str = "metric" + unparsed: bool = True + digipeated: bool = True # count frames that reached here relayed + + # -- afterwards ----------------------------------------------------- + report: bool = True + csv: bool = False + kml: bool = False + + @property + def imperial(self) -> bool: + return str(self.units).lower().startswith("imp") + + def validate(self) -> list[str]: + out = [] + if self.rate < 96_000: + out.append("the sample rate must be at least 96 kS/s to hold a " + "16 kHz channel with room to filter it") + if self.seconds < 0: + out.append("times cannot be negative") + if self.hold <= 0: + out.append("a station must stay on the display for some time") + if not 1e6 < self.frequency < 2e9: + out.append("that is not a frequency a receiver can be tuned to") + return out + + def to_dict(self) -> dict: + return asdict(self) + + +def defaults() -> AprsOptions: + return AprsOptions() + + +def channel_named(name: str) -> float: + """The APRS channel for a region, or the default. + + A number per region by agreement rather than regulation, which is why + this is a list rather than a constant and why getting it wrong means + hearing nothing at all rather than hearing it badly. + """ + wanted = (name or "").strip().lower() + for key, hz, _where in ax25.APRS_CHANNELS: + if wanted == key: + return hz + return ax25.APRS_HZ + + +O = Setting +_REGIONS = tuple(key for key, _hz, _where in ax25.APRS_CHANNELS) + +OPTIONS: tuple[Setting, ...] = ( + O("device", "Receiver", "Receiver", "int", + "which receiver to use, when more than one is plugged in", + "The index shown by `bandsaunter devices`. Zero unless you have " + "several dongles.", + minimum=0, flags=("--device",), example="0"), + O("gain", "Gain", "Receiver", "gain", + "tuner gain in dB, or automatic", + "APRS is a two-second burst from a handheld that may be a street away " + "or a hilltop digipeater thirty miles off, and the automatic gain " + "control copes with both. A fixed gain makes the signal figures " + "comparable between one run and the next, which matters if you are " + "using them to judge an aerial.", + flags=("--gain",), example="auto"), + O("rate", "Sample rate", "Receiver", "float", + "how fast to sample; 96 kS/s is the least that holds the channel", + "A 16 kHz FM channel needs a few times that to filter cleanly. The " + "default leaves room and costs little, since everything after the " + "filter runs at audio rates.", + unit="Hz", minimum=96_000.0, flags=("--rate",), example="240000"), + O("region", "Region", "Receiver", "choice", + "which APRS channel to listen on", + "The frequency is agreed between amateurs rather than allocated, so " + "it differs by region and there is no way to discover it from the " + "air: on the wrong one you hear silence, not a bad signal. North " + "America uses 144.390, Europe 144.800, Australia 145.175.", + choices=_REGIONS, flags=("--region",), example="north-america"), + O("frequency", "Listen on", "Receiver", "float", + "the exact frequency, if the region's channel is not what you want", + "Set by the region unless given here. Worth setting by hand for a " + "local packet network on another channel, or for the 9600 baud links " + "some areas run alongside the main one.", + unit="Hz", minimum=1e6, maximum=2e9, flags=("--frequency", "--freq"), + example="144390000"), + O("simulate", "Invent a channel", "Receiver", "bool", + "put imaginary stations on an imaginary band", + "A dozen stations that are not there -- cars moving, a weather " + "station, a digipeater, objects, messages passing between them -- " + "keyed as real AFSK through the real demodulator and the real " + "parsers. Nothing touches the receiver.", + flags=("--simulate",), off_flags=("--no-simulate",), + guidance="Turn this on to see what the whole thing does without an " + "aerial. Turn it off to hear real stations."), + + # -- listening ------------------------------------------------------ + O("seconds", "Listen for", "Listening", "float", + "how long to listen before stopping (0 = until interrupted)", + "A station beacons every few minutes, and a digipeater relays " + "everything it hears, so an hour gives a fair picture of a region and " + "a minute gives almost none. Everything heard is on the disk as it " + "arrives, so stopping never loses anything.", + unit="s", minimum=0.0, flags=("--seconds",), example="3600"), + O("log", "Write a log", "Listening", "bool", + "write every packet down as it arrives", + "One line of JSON per packet, holding the raw frame as well as what " + "was made of it. It is what the report, the spreadsheet and the map " + "are made from afterwards, and what lets a later version of this " + "program read a format this one cannot.", + flags=("--log",), off_flags=("--no-log",)), + O("packets_seen", "Print every packet", "Listening", "bool", + "one line per packet instead of a table that updates in place", + "The table is easier to watch and useless in a pipe or a file. Turn " + "this on for a plain stream of lines that can be piped somewhere, or " + "to watch individual packets arrive.", + flags=("--packets",), off_flags=("--no-packets",)), + O("hold", "Keep on screen for", "Listening", "float", + "how long a station stays on the display after its last packet", + "An hour by default. Stations beacon every few minutes when moving " + "and every half hour when not, so a short hold makes a fixed station " + "flicker in and out while a long one keeps a car on the screen after " + "it has driven out of range.", + unit="s", minimum=1.0, flags=("--hold",), example="3600"), + O("location", "Receiver at", "Listening", "text", + "where the aerial is, as latitude,longitude (blank = no distances)", + "Only used to work out how far away each station is and in which " + "direction. Left blank the positions are still recorded and drawn; " + "there is simply nothing to measure them from.", + flags=("--at",), example="47.55,-122.30", metavar="LAT,LON"), + + # -- what to show --------------------------------------------------- + O("units", "Show readings in", "Showing", "choice", + "metric or imperial, for the display and the export", + "Only what is shown. APRS is imperial almost throughout -- knots, " + "feet, miles, Fahrenheit -- and all of it is converted once on the " + "way in, so the log holds one kind of number and this can be changed " + "afterwards on an old log.", + choices=("metric", "imperial"), flags=("--units",), example="metric"), + O("unparsed", "Show what cannot be read", "Showing", "bool", + "list packets whose format this does not understand", + "APRS has about twenty formats and a long tail of things one radio " + "manufacturer did once. A packet that defeats every parser here still " + "arrived, still came from a real station, and still has its text; " + "showing it is how the next format gets added.", + flags=("--unparsed",), off_flags=("--no-unparsed",)), + O("digipeated", "Include relayed packets", "Showing", "bool", + "count packets that reached here through a digipeater", + "Most of what a receiver hears is not the original transmission but a " + "hilltop repeating it, which is the whole design. Turning this off " + "leaves only what was heard directly, which is a much shorter list " + "and is the honest measure of what the aerial can reach.", + flags=("--digipeated",), off_flags=("--direct-only",), + guidance="Leave it on for a picture of the network; turn it off to " + "find out what you can actually hear."), + + # -- afterwards ----------------------------------------------------- + O("report", "Report at the end", "Afterwards", "bool", + "print what each station said when the listening stops", + "A table of stations heard with where they were, how far off, how " + "often they spoke and how well they came in, then the messages that " + "passed between them.", + flags=("--report",), off_flags=("--no-report",)), + O("csv", "Also write a spreadsheet", "Afterwards", "bool", + "write the packets as CSV beside the log", + "A row per packet with the position, the speed and the weather in " + "their own columns, which is the shape anything that draws a graph " + "or a track wants.", + flags=("--csv",), off_flags=("--no-csv",)), + O("kml", "Also write a map", "Afterwards", "bool", + "write the stations and their tracks for Google Earth", + "A placemark per station where it was last heard, and a line for " + "anything that moved. The same writer the callsign map uses.", + flags=("--kml",), off_flags=("--no-kml",)), +) + +OPTION_GROUPS = ("Receiver", "Listening", "Showing", "Afterwards") + + +def in_group(group: str) -> list[Setting]: + return [o for o in OPTIONS if o.group == group] + + +def by_key(key: str) -> Setting | None: + return next((o for o in OPTIONS if o.key == key), None) + + +_ZERO_MEANS = {"seconds": "until stopped"} + + +def format_option(option: Setting, value) -> str: + if not value and option.key in _ZERO_MEANS: + return _ZERO_MEANS[option.key] + return format_value(option, value) + + +def describe(options: AprsOptions) -> str: + how_long = ("until stopped" if not options.seconds + else f"{options.seconds:g} s") + where = "simulated" if options.simulate \ + else f"{options.frequency / 1e6:g} MHz" + return (f"{where}, {how_long}, " + f"{'everything' if options.digipeated else 'direct only'}, " + f"{options.units}") + + +def summarise(options: AprsOptions) -> str: + return describe(options) + + +def options_path(directory=None) -> Path: + from .config import DEFAULT_CONFIG_DIR + + return Path(directory or DEFAULT_CONFIG_DIR) / "aprs.yaml" + + +def load_options(directory=None) -> AprsOptions: + """The saved options, or the defaults. A broken file is not an error.""" + options = AprsOptions() + try: + body = yaml.safe_load(options_path(directory).read_text()) or {} + except (OSError, ValueError, yaml.YAMLError): + return options + if not isinstance(body, dict): + return options + known = set(options.__dict__) + for key, value in body.items(): + if key in known and value is not None: + try: + setattr(options, key, type(getattr(options, key))(value)) + except (TypeError, ValueError): + pass + return options + + +def save_options(options: AprsOptions, directory=None) -> Path: + path = options_path(directory) + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as fh: + yaml.safe_dump(options.to_dict(), fh, sort_keys=False, + default_flow_style=False) + return path + + +def logs_in(directory) -> list[Path]: + """Every APRS log in a directory, newest first.""" + try: + found = list(Path(directory).expanduser().glob("aprs_*.jsonl")) + except OSError: + return [] + return sorted(found, key=lambda p: p.stat().st_mtime, reverse=True) + + +def coordinates(text: str) -> tuple[float, float] | None: + try: + lat, lon = (float(x) for x in str(text).split(",", 1)) + except (TypeError, ValueError): + return None + return (lat, lon) if abs(lat) <= 90 and abs(lon) <= 180 else None + + +# --------------------------------------------------------------------------- +# How far off, and which way +# --------------------------------------------------------------------------- + +EARTH_KM = 6371.0088 + + +def distance_km(from_: tuple, to: tuple) -> float: + """Great-circle distance, by the haversine, in kilometres.""" + lat1, lon1 = math.radians(from_[0]), math.radians(from_[1]) + lat2, lon2 = math.radians(to[0]), math.radians(to[1]) + a = (math.sin((lat2 - lat1) / 2) ** 2 + + math.cos(lat1) * math.cos(lat2) * math.sin((lon2 - lon1) / 2) ** 2) + return 2.0 * EARTH_KM * math.asin(min(1.0, math.sqrt(a))) + + +def bearing_deg(from_: tuple, to: tuple) -> float: + """Initial bearing from one place to another, in degrees true.""" + lat1, lon1 = math.radians(from_[0]), math.radians(from_[1]) + lat2, lon2 = math.radians(to[0]), math.radians(to[1]) + east = math.sin(lon2 - lon1) * math.cos(lat2) + north = (math.cos(lat1) * math.sin(lat2) + - math.sin(lat1) * math.cos(lat2) * math.cos(lon2 - lon1)) + return math.degrees(math.atan2(east, north)) % 360.0 + + +# --------------------------------------------------------------------------- +# Who is out there, and what they said +# --------------------------------------------------------------------------- + +# How much of a moving station's track to keep in memory. A car beaconing +# every twenty seconds for an evening is a few hundred points; this is a +# ceiling rather than a target, and the log has every one of them regardless. +TRACK_POINTS = 600 + + +@dataclass +class Station: + """One callsign, and the latest of everything it has said. + + The latest of each *kind* of thing, not the latest packet: a station may + send its position, then a status, then a weather report, and a display + showing only the newest packet would lose two of the three every time. + """ + + call: str = "" + kind: str = "" # what it mostly sends + symbol: str = "" + position: object = None # a packets.Position + course: float | None = None + speed: float | None = None + altitude: float | None = None + weather: dict = field(default_factory=dict) + status: str = "" + comment: str = "" + path: tuple = () + first: float = 0.0 + last: float = 0.0 + packets: int = 0 + direct: int = 0 # heard without a digipeater in between + snr: float = 0.0 + best_snr: float = 0.0 + kinds: set = field(default_factory=set) + track: list = field(default_factory=list) # (when, lat, lon) + object_of: str = "" # the station that placed this, if any + + @property + def moving(self) -> bool: + return bool(self.speed) and self.speed > 1.0 + + @property + def gap(self) -> float: + """The average wait between packets, in seconds, or zero.""" + return (self.last - self.first) / (self.packets - 1) \ + if self.packets > 1 else 0.0 + + def add(self, packet) -> None: + self.kind = packet.kind if packet.kind != "unparsed" else self.kind + self.kinds.add(packet.kind) + self.first = self.first or packet.at + self.last = max(self.last, packet.at) + self.packets += 1 + if not any("*" in hop for hop in packet.path): + self.direct += 1 + if packet.snr: + self.snr = packet.snr + self.best_snr = max(self.best_snr, packet.snr) + if packet.path: + self.path = packet.path + if packet.position is not None: + self.position = packet.position + self.symbol = packet.position.symbol or self.symbol + here = (packet.at, packet.position.latitude, + packet.position.longitude) + if not self.track or self.track[-1][1:] != here[1:]: + self.track.append(here) + del self.track[:-TRACK_POINTS] + for name in ("course", "speed", "altitude"): + value = getattr(packet, name) + if value is not None: + setattr(self, name, value) + if packet.weather: + self.weather.update(packet.weather) + if packet.status: + self.status = packet.status + if packet.comment: + self.comment = packet.comment + + def away(self, home) -> tuple[float, float] | None: + """How far off and in which direction, if there is a home to measure + from and a position to measure to.""" + if home is None or self.position is None: + return None + there = (self.position.latitude, self.position.longitude) + return distance_km(home, there), bearing_deg(home, there) + + +class Net: + """Every station heard, and every message that passed between them. + + The messages are kept apart from the stations on purpose. A message is + an event between two callsigns rather than a property of either, and a + conversation read in order is the thing somebody actually wants to see. + """ + + def __init__(self): + self.stations: dict[str, Station] = {} + self.messages: list = [] + self.packets = 0 + self.unparsed = 0 + + def __len__(self) -> int: + return len(self.stations) + + def add(self, packet) -> Station: + name = packet.station + station = self.stations.get(name) + if station is None: + station = self.stations[name] = Station(call=name) + if packet.name: + station.object_of = packet.source + station.add(packet) + self.packets += 1 + if packet.kind == "unparsed": + self.unparsed += 1 + if packet.message is not None: + self.messages.append(packet) + return station + + def showing(self, hold: float, now: float | None = None) -> list[Station]: + now = time.time() if now is None else now + alive = [s for s in self.stations.values() if now - s.last <= hold] + return sorted(alive, key=lambda s: (-s.last, s.call)) + + def all(self) -> list[Station]: + return sorted(self.stations.values(), key=lambda s: (s.first, s.call)) + + @classmethod + def of(cls, heard) -> "Net": + net = cls() + for packet in heard: + net.add(packet) + return net + + +# --------------------------------------------------------------------------- +# Listening +# --------------------------------------------------------------------------- + +@dataclass +class Heard: + """What one listening session came to.""" + + packets: int = 0 + net: object = None + log_path: Path | None = None + csv_path: Path | None = None + kml_path: Path | None = None + + @property + def stations(self) -> int: + return len(self.net) if self.net is not None else 0 + + +def open_device(console, options: AprsOptions): + """The receiver, or an invented channel, or None if neither can be had.""" + from rich.panel import Panel + from rich.text import Text + + from .device import RtlSdrDevice, RtlSdrError + + if options.simulate: + console.print("[yellow]simulated: these stations are not there." + "[/yellow]") + return SimulatedChannel(sample_rate=options.rate, realtime=True).open() + try: + device = RtlSdrDevice(index=options.device, + sample_rate=int(options.rate), + gain=options.gain, + agc=options.gain == "auto") + device.open() + except RtlSdrError as exc: + console.print(Panel(Text(str(exc)), + title="[red]cannot open the receiver", + border_style="red")) + return None + return device + + +def open_log(console, options: AprsOptions, output_dir: str, + started: float, log_path=None): + """The packet log, or None if it was not wanted or cannot be written.""" + from .aprslog import AprsLog + + if not options.log: + return None + stamp = datetime.fromtimestamp(started).strftime("%Y-%m-%d_%H_%M_%S") + where = Path(log_path).expanduser() if log_path else \ + Path(output_dir).expanduser() / f"aprs_{stamp}.jsonl" + try: + return AprsLog(where, frequency=options.frequency, + sample_rate=options.rate, + receiver="simulated" if options.simulate else + f"device {options.device}", started=started) + except OSError as exc: + console.print(f"[red]cannot write {where}: {exc}[/red]") + return None + + +def make_receiver(options: AprsOptions): + """The FM demodulator and the packet receiver, wired together. + + De-emphasis is turned off, which matters and is easy to miss. Voice FM + lifts the high frequencies on transmit and drops them again on receive; + packet radio takes its audio from the discriminator *before* that + happens, because dropping the highs would take four decibels off the + 2200 Hz tone and leave the 1200 Hz one alone -- which is exactly the + difference the demodulator here is measuring. + """ + from .demod import make_demodulator + + demod = make_demodulator("nfm", int(options.rate), bandwidth=16_000.0, + audio_rate=22_050, deemphasis=None) + return demod, ax25.Receiver(float(demod.audio_rate)) + + +def pump(device, options: AprsOptions, net: Net, log, started: float, + on_block=None, on_packet=None, on_samples=None, + stopping=None) -> int: + """Read the receiver until it stops, or until told to. + + The one loop both front ends are driven from, so that what is written to + the log cannot depend on which one you happened to be looking at. + """ + demod, receiver = make_receiver(options) + total = 0 + device.tune(options.frequency) + block = int(options.rate) # a second at a time + while stopping is None or not stopping(): + at = time.time() + samples = device.read_samples(block) + if samples is None or samples.size == 0: + break + if on_samples is not None: + on_samples(samples, at) + audio, iq = demod.step(samples) + if audio.size == 0: + continue + for frame in receiver.feed(audio, when=at, snr=_level(iq)): + packet = packets.parse(frame, now=at) + if packet.kind == "unparsed" and not options.unparsed: + continue + if not options.digipeated and frame.heard_through: + continue + net.add(packet) + total += 1 + if log is not None: + log.append(packet, frame) + if on_packet is not None: + on_packet(packet, frame) + if on_block is not None: + on_block(total) + if options.seconds and time.time() - started >= options.seconds: + break + return total + + +def _level(iq) -> float: + """How far the channel stood above its own quiet level, in decibels. + + A ratio off one receiver in one second, and not a power at the aerial -- + a dongle has no reference level and, on automatic gain, no fixed gain + either. Enough to compare two stations at the same moment and to point + an aerial, which is what it is for. + """ + import numpy as np + + if iq is None or iq.size < 64: + return 0.0 + magnitude = np.abs(iq) + quiet = float(np.percentile(magnitude, 20)) + loud = float(np.percentile(magnitude, 95)) + if quiet <= 0.0 or loud <= quiet: + return 0.0 + return 20.0 * math.log10(loud / quiet) + + +# --------------------------------------------------------------------------- +# A channel that is not there +# --------------------------------------------------------------------------- + +@dataclass +class VirtualStation: + """An amateur station that does not exist, doing what one does. + + It has a callsign, a place to be, possibly a speed to get somewhere at, + and a reason to transmit. What it sends goes out through the real + encoders, is keyed as real Bell 202 audio, is modulated onto a real FM + carrier and comes back through the real demodulator and the real + parsers -- so nothing downstream can tell the difference, which is the + point, because everything downstream is what is being exercised. + """ + + call: str = "N0CALL" + kind: str = "beacon" # beacon, car, weather, object, message + latitude: float = 47.55 + longitude: float = -122.30 + course: float = 90.0 + speed: float = 0.0 # km/h + symbol: str = "/-" + comment: str = "" + every: float = 120.0 + phase: float = 0.0 + strength: float = 1.0 + path: tuple = ("WIDE1-1", "WIDE2-1") + talks_to: str = "" + turn: int = 0 + + def frame(self, at: float) -> bytes: + """One transmission, as the bytes that go on the air.""" + self.turn += 1 + stamp = time.strftime("%d%H%Mz", time.gmtime(at)) + if self.kind == "weather": + info = packets.weather_report( + self.latitude, self.longitude, timestamp=stamp, + wind_from=(self.turn * 37) % 360, wind=11.0 + self.turn % 7, + gust=18.0, temperature=14.5 + (self.turn % 5) * 0.4, + rain_hour=0.0, humidity=62, pressure=1013.2) + return ax25.frame_bytes(self.call, "APRS", info, self.path) + if self.kind == "message" and self.talks_to: + info = packets.message_text(self.talks_to, + f"testing {self.turn}", + number=str(self.turn)) + return ax25.frame_bytes(self.call, "APRS", info, self.path) + if self.kind == "object": + info = packets.object_report(f"EVENT{self.turn % 3}", + self.latitude + 0.02, + self.longitude - 0.02, + symbol="/:", timestamp=stamp) + return ax25.frame_bytes(self.call, "APRS", info, self.path) + if self.kind == "car": + # A mobile, which in the real world means Mic-E. + destination, info = packets.mic_e_report( + self.latitude, self.longitude, course=self.course, + speed=self.speed, symbol=self.symbol, status="en route", + comment=self.comment) + return ax25.frame_bytes(self.call, destination, info, self.path) + info = packets.position_report(self.latitude, self.longitude, + self.symbol, self.comment, + timestamp=stamp) + return ax25.frame_bytes(self.call, "APRS", info, self.path) + + def advance(self, seconds: float) -> None: + if self.speed <= 0.0: + return + north = math.cos(math.radians(self.course)) * self.speed * seconds / 3600.0 + east = math.sin(math.radians(self.course)) * self.speed * seconds / 3600.0 + self.latitude += north / 111.32 + self.longitude += east / (111.32 * max(0.2, math.cos( + math.radians(self.latitude)))) + self.course = (self.course + math.sin(self.turn * 0.7) * 3.0) % 360.0 + + +def default_stations() -> list[VirtualStation]: + """A channel's worth of them, in a region that does not exist.""" + return [ + VirtualStation(call="W1AW-1", kind="beacon", latitude=47.6062, + longitude=-122.3321, symbol="/#", every=600.0, + phase=5.0, comment="digipeater", path=()), + VirtualStation(call="KU0W-9", kind="car", latitude=47.5480, + longitude=-122.2960, course=70.0, speed=54.0, + symbol="/>", every=45.0, phase=12.0, + comment="mobile"), + VirtualStation(call="N0ABC-7", kind="car", latitude=47.6200, + longitude=-122.3500, course=200.0, speed=32.0, + symbol="/b", every=60.0, phase=27.0, strength=0.5), + VirtualStation(call="KB1XYZ", kind="weather", latitude=47.5900, + longitude=-122.2700, every=300.0, phase=40.0), + VirtualStation(call="W7ABC-10", kind="object", latitude=47.5700, + longitude=-122.3100, every=420.0, phase=70.0), + VirtualStation(call="KU0W-9", kind="message", talks_to="W1AW-1", + latitude=47.5480, longitude=-122.2960, every=240.0, + phase=95.0), + VirtualStation(call="VE3XYZ-5", kind="beacon", latitude=47.4900, + longitude=-122.4100, symbol="\\k", every=900.0, + phase=130.0, strength=0.35, comment="portable"), + ] + + +class SimulatedChannel: + """A receiver-shaped source of an amateur band that is not busy. + + Answers ``read_samples`` the way the dongle does, with real AFSK on a + real FM carrier, so the whole section -- the demodulator, the framer, + every parser, the log, the report and the map -- runs through without an + aerial or a licence. + """ + + def __init__(self, stations=None, sample_rate: float = 240_000.0, + noise: float = 0.05, seed: int = 0, realtime: bool = False, + deviation: float = 3_000.0): + import numpy as np + + self.stations = list(stations) if stations is not None \ + else default_stations() + self.sample_rate = float(sample_rate) + self.noise = noise + self.deviation = deviation + self.rng = np.random.default_rng(seed) + self.frequency = ax25.APRS_HZ + self.clock = 0.0 + self.realtime = realtime + self._last = None + # Whatever of a transmission did not fit in the block it began in. + # A packet is most of a second and a block is about one, so a + # transmission straddling the boundary is not an edge case -- it is + # most of them, and dropping the remainder means the receiver sees + # the first tenth of a frame and nothing decodes at all. This was + # invisible for a while because a simulated clock that advances in + # exact seconds puts every transmission at the start of a block. + self._pending = np.zeros(0, dtype=np.complex64) + + def open(self): + return self + + def close(self) -> None: + return None + + def tune(self, hz: float, settle: bool = True) -> int: + self.frequency = hz + return int(hz) + + def read_samples(self, count: int, flush: bool = False): + import numpy as np + + seconds = count / self.sample_rate + if self.realtime: + now = time.monotonic() + if self._last is not None: + behind = seconds - (now - self._last) + if behind > 0: + time.sleep(behind) + now = time.monotonic() + seconds = max(0.0, now - self._last) + self._last = now + block = np.zeros(count, dtype=np.complex64) + if self._pending.size: + take = min(self._pending.size, count) + block[:take] += self._pending[:take] + self._pending = self._pending[take:] + began, ended = self.clock, self.clock + seconds + for station in self.stations: + for at in _due(station, began, ended): + burst = self._keyed(station) + start = int((at - began) * self.sample_rate) + if start >= count: + self._hold(burst, start - count) + continue + room = min(burst.size, count - start) + block[start:start + room] += burst[:room] + if burst.size > room: + self._hold(burst[room:], 0) + station.advance(seconds) + self.clock = ended + if self.noise: + block = block + self.noise * ( + self.rng.standard_normal(count) + + 1j * self.rng.standard_normal(count)).astype(np.complex64) + return block.astype(np.complex64) + + def _hold(self, samples, offset: int) -> None: + """Keep the rest of a transmission for the block it runs into.""" + import numpy as np + + need = offset + samples.size + if self._pending.size < need: + self._pending = np.concatenate( + (self._pending, + np.zeros(need - self._pending.size, dtype=np.complex64))) + self._pending[offset:need] += samples + + def _keyed(self, station: VirtualStation): + """One transmission as a receiver would see it: AFSK inside FM.""" + import numpy as np + + audio = ax25.modulate(station.frame(self.clock), self.sample_rate, + amplitude=0.7, quiet_ms=10.0) + turn = 2.0 * math.pi * self.deviation / self.sample_rate + phase = np.cumsum(audio.astype(np.float64)) * turn + return (station.strength * np.exp(1j * phase)).astype(np.complex64) + + def read_seconds(self, seconds: float, flush: bool = False): + return self.read_samples(int(self.sample_rate * seconds)) + + +def _due(station: VirtualStation, began: float, ended: float) -> list[float]: + """Every moment in a block at which one station is due to transmit.""" + if station.every <= 0 or ended <= began: + return [] + first = math.ceil((began - station.phase) / station.every) + out, at = [], station.phase + first * station.every + while at < ended: + if at >= began: + out.append(at) + at += station.every + return out + + +# --------------------------------------------------------------------------- +# A listening session, start to finish +# --------------------------------------------------------------------------- + +def listen(console, options: AprsOptions, output_dir: str, + log_path=None, device=None) -> Heard: + """Park on the APRS channel and write down everything that passes. + + Everything heard goes into the log as it arrives, because a packet is a + two-second transmission and then gone: the screen is for the person + watching, and the log is for the report, the map and everything + afterwards. + """ + heard = Heard() + device = open_device(console, options) if device is None else device + if device is None: + return heard + + started = time.time() + log = open_log(console, options, output_dir, started, log_path) + net = Net() + heard.net = net + console.print(f"[grey62]listening on {options.frequency / 1e6:g} MHz at " + f"{options.rate / 1e6:g} MS/s — control-C to stop" + "[/grey62]") + if log is not None: + console.print(f"[grey62]writing {log.path}[/grey62]") + + display, live = _open_display(console, options, started) + + def on_block(total: int) -> None: + if live is not None: + display.update(net, total, + log.path if log is not None else None) + live.update(display.render()) + + def on_packet(packet, frame) -> None: + if options.packets_seen or live is None: + console.print(f"[cyan]{packet.station}[/cyan] " + f"{packet.describe(options.imperial)}", + highlight=False) + + total = 0 + try: + total = pump(device, options, net, log, started, + on_block=on_block, on_packet=on_packet) + except KeyboardInterrupt: + pass + finally: + if live is not None: + live.stop() + console.print("[grey62]stopped listening[/grey62]") + device.close() + if log is not None: + log.close() + heard.log_path = log.path + heard.packets = total + return finish(console, options, output_dir, heard) + + +def _open_display(console, options: AprsOptions, started: float): + """A live table where there is a terminal to draw it on, else nothing.""" + if options.packets_seen or not getattr(console, "is_terminal", False): + return None, None + from rich.live import Live + + from .ui import AprsDisplay + + display = AprsDisplay(console, hold=options.hold, + imperial=options.imperial, + frequency=options.frequency, + home=coordinates(options.location)) + display.started = started + live = Live(display.render(), console=console, refresh_per_second=2, + screen=False, transient=False, vertical_overflow="crop") + live.start() + return display, live + + +def finish(console, options: AprsOptions, output_dir: str, + heard: Heard) -> Heard: + """Say what was heard, and leave the files behind.""" + net = heard.net + if net is None or not len(net): + console.print("[yellow]nothing heard. APRS is a two-second " + "transmission every few minutes, so give it a while " + "— and check the region: on the wrong channel there " + "is silence rather than a bad signal.[/yellow]") + return heard + + if options.report: + report(console, net, options) + if heard.log_path is not None: + from .aprslog import read_logs, write_csv, write_kml + + if options.csv: + heard.csv_path = _write(console, write_csv, + heard.log_path.with_suffix(".csv"), + read_logs([heard.log_path]), options) + if options.kml: + heard.kml_path = _write(console, write_kml, + heard.log_path.with_suffix(".kml"), + read_logs([heard.log_path]), options) + console.print(f"[grey62]read it back: bandsaunter packets " + f"{heard.log_path}[/grey62]") + return heard + + +def _write(console, writer, where, heard, options): + try: + written = writer(where, heard, options.imperial) + except OSError as exc: + console.print(f"[red]cannot write {where}: {exc}[/red]") + return None + console.print(f"[green]wrote {written}[/green]") + return written + + +# --------------------------------------------------------------------------- +# What the evening came to +# --------------------------------------------------------------------------- + +SIGNAL_FAIR = 8.0 +SIGNAL_GOOD = 16.0 + + +def signal_text(level: float) -> str: + """One station's strength, coloured so an aerial can be aimed by it.""" + if not level: + return "" + colour = ("red" if level < SIGNAL_FAIR else + "yellow" if level < SIGNAL_GOOD else "green") + return f"[{colour}]{level:.0f} dB[/{colour}]" + + +def report(console, net: Net, options: AprsOptions) -> None: + """Three tables: who was heard, what they said, and what passed between. + + Split because they answer different questions. The first is about the + band and the aerial; the second is about the stations; the third is the + only part of APRS that is a conversation rather than a broadcast, and + reading it in order is the point of it. + """ + from rich.table import Table + + from .packets import height_text + from .acurite import Measure, format_measure + + home = coordinates(options.location) + stations = sorted(net.all(), key=lambda s: (-s.packets, s.call)) + + t = Table(box=None, header_style="bold", pad_edge=False, + title="[bold]stations heard[/bold]", title_justify="left") + t.add_column("station", overflow="fold") + t.add_column("what", style="grey62", overflow="fold") + t.add_column("position", style="grey62", no_wrap=True) + if home is not None: + t.add_column("away", justify="right", no_wrap=True) + t.add_column("pkts", justify="right") + t.add_column("direct", justify="right", style="grey62") + t.add_column("signal", justify="right") + t.add_column("last heard", style="grey62", no_wrap=True) + for station in stations: + row = [station.call + (" (object)" if station.object_of else ""), + station.symbol or station.kind, + station.position.describe() if station.position else ""] + if home is not None: + away = station.away(home) + row.append("" if away is None else + f"{format_measure(Measure('', away[0], 'km'), options.imperial)}" + f" {_point(away[1])}") + row += [f"{station.packets:,}", f"{station.direct:,}", + signal_text(station.snr), + datetime.fromtimestamp(station.last).strftime("%H:%M:%S") + if station.last else ""] + t.add_row(*row) + console.print(t) + + said = [s for s in stations if s.status or s.comment or s.weather + or s.speed or s.altitude is not None] + if said: + w = Table(box=None, header_style="bold", pad_edge=False, + title="\n[bold]what they said[/bold]", title_justify="left") + w.add_column("station", overflow="fold") + w.add_column("moving", style="grey62", no_wrap=True) + w.add_column("and", overflow="fold") + for station in said: + moving = "" + if station.speed: + moving = (f"{_point(station.course or 0.0)} " + + format_measure(Measure("", station.speed, "km/h"), + options.imperial)) + if station.altitude is not None: + moving += (" " if moving else "") + height_text( + station.altitude, options.imperial) + words = " ".join( + [f"{name} {format_measure(m, options.imperial)}" + for name, m in station.weather.items()] + + [x for x in (station.status, station.comment) if x]) + w.add_row(station.call, moving, words) + console.print(w) + + if net.messages: + m = Table(box=None, header_style="bold", pad_edge=False, + title="\n[bold]what passed between them[/bold]", + title_justify="left") + m.add_column("when", style="grey62", no_wrap=True) + m.add_column("from", no_wrap=True) + m.add_column("message", overflow="fold") + for packet in net.messages[-40:]: + m.add_row(datetime.fromtimestamp(packet.at).strftime("%H:%M:%S"), + packet.source, packet.message.describe()) + console.print(m) + + if net.unparsed: + console.print(f"[grey62]{net.unparsed:,} packet" + f"{'s' if net.unparsed != 1 else ''} in a format this " + f"cannot read; their text is in the log[/grey62]") + + +def _point(degrees: float) -> str: + from .acurite import compass + + return compass(degrees) diff --git a/bandsaunter/aprslog.py b/bandsaunter/aprslog.py new file mode 100644 index 0000000..5a55428 --- /dev/null +++ b/bandsaunter/aprslog.py @@ -0,0 +1,345 @@ +"""Writing down what came over the channel, and reading it back. + +One line of JSON per packet, written the moment it arrives. Flushed after +every one, for the reason every log in this program is: a listening session +ends when the operator gets bored and presses control-C, and a log that only +reached the disk on a clean shutdown would be empty exactly when it was most +wanted. + +Each line keeps the whole AX.25 frame in hexadecimal alongside whatever was +made of it. APRS has a long tail of formats -- about twenty in the +specification and more that one manufacturer invented once -- so a packet +this version cannot read is still on the disk in full, and a later version +that can read it can go back through old logs and do so. That is not a +hypothetical: the reason the frame is kept is that the list of formats is +still growing. + +Out of it come two other things. A spreadsheet, because a track and a +temperature are both columns of numbers people want to plot; and a KML file, +because the natural thing to do with a hundred positions is to look at them +on a globe. +""" + +from __future__ import annotations + +import csv +import json +import time +from datetime import datetime +from pathlib import Path + +from .acurite import Measure +from .packets import Message, Packet, Position, Telemetry + +__all__ = ["AprsLog", "read_logs", "write_csv", "write_kml", "logs_in", + "LOG_VERSION"] + +LOG_VERSION = 1 + + +class AprsLog: + """A JSON Lines record of every packet heard, written as it arrives.""" + + def __init__(self, path, receiver: str = "", frequency: float = 0.0, + sample_rate: float = 0.0, started: float = 0.0): + self.path = Path(path) + self.packets = 0 + self.started = started or time.time() + self.path.parent.mkdir(parents=True, exist_ok=True) + self._file = self.path.open("a", encoding="utf8") + self._write({"log": "bandsaunter-aprs", "version": LOG_VERSION, + "started": round(self.started, 3), + "started_local": datetime.fromtimestamp( + self.started).strftime("%Y-%m-%d %H:%M:%S"), + "frequency": frequency, "sample_rate": sample_rate, + "receiver": receiver}) + + def _write(self, body: dict) -> None: + self._file.write(json.dumps(body, separators=(",", ":"), + ensure_ascii=False) + "\n") + self._file.flush() + + def append(self, packet: Packet, frame=None) -> None: + """Record one packet: what arrived, and what was made of it.""" + body: dict = {"t": round(packet.at or time.time(), 3), + "src": packet.source, "dst": packet.destination, + "kind": packet.kind, "info": packet.info} + if packet.path: + body["path"] = list(packet.path) + if frame is not None and frame.raw: + # The frame is the evidence; everything else in the line is an + # opinion about it, and the opinions may improve later. + body["hex"] = frame.raw.hex().upper() + if packet.snr: + body["snr"] = round(packet.snr, 1) + if packet.reported: + body["reported"] = round(packet.reported, 3) + if packet.position is not None: + body["lat"] = round(packet.position.latitude, 6) + body["lon"] = round(packet.position.longitude, 6) + body["sym"] = packet.position.table + packet.position.code + if packet.position.ambiguity: + body["vague"] = packet.position.ambiguity + for name in ("course", "speed", "altitude", "range", "power", + "height", "gain"): + value = getattr(packet, name) + if value is not None: + body[name] = round(float(value), 3) + for name in ("name", "status", "comment", "beam"): + value = getattr(packet, name) + if value: + body[name] = value + if not packet.live: + body["killed"] = True + if packet.weather: + body["wx"] = {n: [m.value, m.unit] + for n, m in packet.weather.items()} + if packet.message is not None: + body["msg"] = {k: v for k, v in vars(packet.message).items() if v} + if packet.telemetry is not None: + body["tlm"] = {"seq": packet.telemetry.sequence, + "a": list(packet.telemetry.analogue), + "d": packet.telemetry.digital} + self.packets += 1 + self._write(body) + + def close(self) -> None: + try: + self._file.close() + except OSError: + pass + + def __enter__(self) -> "AprsLog": + return self + + def __exit__(self, *exc) -> None: + self.close() + + +# --------------------------------------------------------------------------- +# Reading it back +# --------------------------------------------------------------------------- + +def logs_in(directory) -> list[Path]: + try: + found = list(Path(directory).expanduser().glob("aprs_*.jsonl")) + except OSError: + return [] + return sorted(found, key=lambda p: p.stat().st_mtime, reverse=True) + + +def read_logs(paths) -> list[Packet]: + """Every packet in one or more logs, in the order they were heard. + + A line that will not parse is skipped rather than fatal. A log is + appended to while the disk fills and the power goes off, so the last line + of one is quite often half a line. + """ + out: list[Packet] = [] + for path in ([paths] if isinstance(paths, (str, Path)) else paths): + try: + text = Path(path).expanduser().read_text(encoding="utf8") + except OSError: + continue + for line in text.splitlines(): + packet = _packet_from(line) + if packet is not None: + out.append(packet) + out.sort(key=lambda p: p.at) + return out + + +def _packet_from(line: str) -> Packet | None: + line = line.strip() + if not line: + return None + try: + body = json.loads(line) + except ValueError: + return None + if not isinstance(body, dict) or "src" not in body: + return None # the header line, or something else entirely + + packet = Packet(kind=str(body.get("kind", "unparsed")), + source=str(body.get("src", "")), + destination=str(body.get("dst", "")), + path=tuple(body.get("path") or ()), + at=float(body.get("t", 0.0) or 0.0), + reported=float(body.get("reported", 0.0) or 0.0), + snr=float(body.get("snr", 0.0) or 0.0), + info=str(body.get("info", ""))) + if "lat" in body and "lon" in body: + symbol = str(body.get("sym", "/-")) + packet.position = Position(latitude=float(body["lat"]), + longitude=float(body["lon"]), + ambiguity=int(body.get("vague", 0) or 0), + table=symbol[:1] or "/", + code=symbol[1:2] or "-") + for name in ("course", "speed", "altitude", "range", "power", "height", + "gain"): + if name in body: + setattr(packet, name, float(body[name])) + for name in ("name", "status", "comment", "beam"): + if name in body: + setattr(packet, name, str(body[name])) + packet.live = not body.get("killed", False) + for name, value in (body.get("wx") or {}).items(): + if isinstance(value, list) and value: + packet.weather[name] = Measure( + name, float(value[0]), str(value[1]) if len(value) > 1 else "") + if body.get("msg"): + packet.message = Message(**{k: str(v) + for k, v in body["msg"].items() + if k in vars(Message())}) + if body.get("tlm"): + told = body["tlm"] + packet.telemetry = Telemetry(sequence=str(told.get("seq", "")), + analogue=tuple(told.get("a") or ()), + digital=str(told.get("d", ""))) + return packet + + +# --------------------------------------------------------------------------- +# Out to a spreadsheet +# --------------------------------------------------------------------------- + +_IMPERIAL = {"C": "F", "km/h": "mph", "mm": "in", "km": "mi", "m": "ft"} + + +def write_csv(path, heard, imperial: bool = False) -> Path: + """A row per packet, with what it said in columns. + + The union of every weather quantity anything reported, so a channel with + one weather station on it has temperature and pressure columns and every + car leaves them empty. That is the shape a spreadsheet wants. + """ + path = Path(path).expanduser() + path.parent.mkdir(parents=True, exist_ok=True) + quantities: list[str] = [] + for packet in heard: + for name in packet.weather: + if name not in quantities: + quantities.append(name) + + speed_unit = "mph" if imperial else "km/h" + height_unit = "ft" if imperial else "m" + heads = ["time", "unix", "station", "source", "destination", "path", + "kind", "latitude", "longitude", "symbol", "course", + f"speed ({speed_unit})", f"altitude ({height_unit})", + "signal (dB)", "status", "comment"] \ + + [_column(name, heard, imperial) for name in quantities] + with open(path, "w", encoding="utf8", newline="") as fh: + out = csv.writer(fh) + out.writerow(heads) + for packet in heard: + place = packet.position + out.writerow([ + datetime.fromtimestamp(packet.at).isoformat(timespec="seconds") + if packet.at else "", + f"{packet.at:.3f}" if packet.at else "", + packet.station, packet.source, packet.destination, + ",".join(packet.path), packet.kind, + f"{place.latitude:.6f}" if place else "", + f"{place.longitude:.6f}" if place else "", + (place.table + place.code) if place else "", + f"{packet.course:.0f}" if packet.course is not None else "", + _shown(packet.speed, "km/h", imperial), + _shown(packet.altitude, "m", imperial), + f"{packet.snr:.1f}" if packet.snr else "", + packet.status, packet.comment] + + [_shown(packet.weather[name].value, + packet.weather[name].unit, imperial) + if name in packet.weather else "" + for name in quantities]) + return path + + +def _column(name: str, heard, imperial: bool) -> str: + unit = "" + for packet in heard: + if name in packet.weather and packet.weather[name].unit: + unit = packet.weather[name].unit + break + if imperial: + unit = _IMPERIAL.get(unit, unit) + return f"{name} ({unit})" if unit else name + + +def _shown(value, unit: str, imperial: bool) -> str: + """One number, in whichever system was asked for, as a plain figure. + + Plain because a column of "21.5 C" is text and a column of 21.5 is a + temperature, and only one of those can be plotted. + """ + if value is None: + return "" + if imperial: + if unit == "C": + value = value * 9 / 5 + 32 + elif unit in ("km/h", "km"): + value = value / 1.609344 + elif unit == "mm": + value = value / 25.4 + elif unit == "m": + value = value / 0.3048 + return str(round(float(value), 3)) + + +# --------------------------------------------------------------------------- +# Out to a globe +# --------------------------------------------------------------------------- + +def write_kml(path, heard, imperial: bool = False, + title: str = "bandsaunter — APRS stations") -> Path | None: + """A pin where each station was last heard, and a line where it moved. + + Only the stations that said where they were, because a pin at nowhere is + worse than no pin: it puts a station off the west coast of Africa, which + is where nought degrees by nought degrees is and is the reason that bug + has a name. + """ + from xml.sax.saxutils import escape as xml_escape + + tracks: dict[str, list] = {} + latest: dict[str, object] = {} + for packet in heard: + if packet.position is None: + continue + where = (packet.position.longitude, packet.position.latitude, + packet.altitude or 0.0) + line = tracks.setdefault(packet.station, []) + if not line or line[-1][:2] != where[:2]: + line.append(where) + latest[packet.station] = packet + if not latest: + return None + + path = Path(path).expanduser() + path.parent.mkdir(parents=True, exist_ok=True) + out = ['', + '', " ", + f" {xml_escape(title)}", + ' "] + for call, line in sorted(tracks.items()): + packet = latest[call] + told = xml_escape(packet.describe(imperial)) + name = xml_escape(call) + if len(line) > 1: + where = " ".join(f"{lon:.6f},{lat:.6f},{alt:.0f}" + for lon, lat, alt in line) + out += [" ", f" {name}", + f" {told}", + " #track", + " 1", + f" {where}", + " ", " "] + last = line[-1] + out += [" ", f" {name}", + f" {told}", + " " + f"{last[0]:.6f},{last[1]:.6f},{last[2]:.0f}" + "", " "] + out += [" ", "", ""] + path.write_text("\n".join(out), encoding="utf8") + return path diff --git a/bandsaunter/ax25.py b/bandsaunter/ax25.py new file mode 100644 index 0000000..904fb63 --- /dev/null +++ b/bandsaunter/ax25.py @@ -0,0 +1,524 @@ +"""AX.25 over the air: the frames APRS is carried in, and how to recover them. + +Amateur packet radio sends data as HDLC frames over a carrier that is simply +switched between two audio tones inside an ordinary FM transmission -- 1200 Hz +for a mark and 2200 Hz for a space, twelve hundred of them a second, which is +Bell 202 and is what a telephone modem sounded like in 1976. It has stayed +because it works through any FM radio ever made, and because every handheld in +a rucksack is already an AFSK transmitter with a microphone socket. + +Four things happen between the aerial and a frame, and each is a place to get +it wrong. + +**The tones become a soft symbol.** Two correlators, one at each tone, and +the difference between them. A correlator rather than a frequency +discriminator because the tones are less than an octave apart and radio audio +is distorted enough that instantaneous frequency wanders badly; asking which +of the two tones a bit-length window contains more of is a question that +survives a weak signal. + +**The soft symbol becomes bits.** Sampled once a bit, at an instant kept in +the middle of the bit by a phase-locked loop that is nudged at every zero +crossing. The loop is what makes this a receiver rather than a decoder of +recordings: it carries its phase from one block of audio to the next, so a +frame that straddles the boundary is read straight through. + +**The bits become a frame.** NRZI first -- the data is in whether the tone +changed, not which tone it is, which makes the whole thing immune to being +wired up backwards. Then HDLC: frames are delimited by the flag 01111110 and +a zero is stuffed after every five ones so the flag cannot occur inside one. + +**The frame is believed or it is not.** Sixteen bits of CRC, and nothing +without a correct one is reported. That is what makes it safe to run this +over hours of an open squelch: a frame either checks out or it never existed. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass, field + +import numpy as np + +__all__ = ["Address", "Frame", "Receiver", "fcs", "frame_from", "frame_bytes", + "hdlc_frames", "stuff", "unstuff", "modulate", "nrzi", "un_nrzi", + "MARK_HZ", "SPACE_HZ", "BAUD", "FLAG", "APRS_HZ", "APRS_CHANNELS", + "UI_CONTROL", "NO_LAYER_3", "MAX_FRAME"] + + +# Bell 202, as every VHF packet station on earth sends it. +MARK_HZ = 1200.0 +SPACE_HZ = 2200.0 +BAUD = 1200.0 +FLAG = "01111110" + +# Where APRS lives. One channel per region by agreement rather than by +# regulation, which is why there is a list of them rather than a number. +APRS_HZ = 144_390_000.0 +APRS_CHANNELS = ( + ("north-america", 144_390_000.0, "United States, Canada, Mexico"), + ("europe", 144_800_000.0, "IARU Region 1, including the UK"), + ("australia", 145_175_000.0, "Australia and New Zealand"), + ("japan", 144_640_000.0, "Japan"), + ("brazil", 145_570_000.0, "Brazil"), + ("thailand", 145_525_000.0, "Thailand"), +) + +# An unnumbered information frame with no layer-3 protocol, which is what +# every APRS packet is and very nearly all this will ever see. +UI_CONTROL = 0x03 +NO_LAYER_3 = 0xF0 + +# The longest thing worth believing: eight two-byte-addressed hops, a control +# and protocol byte, and 256 bytes of information. +MAX_FRAME = 8 * 7 + 2 + 2 + 256 + 2 + + +# --------------------------------------------------------------------------- +# What a frame is made of +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Address: + """One callsign in a frame's path, with everything packed around it. + + An AX.25 address is six characters and four bits, and the other four bits + of the last byte carry the parts that matter here: which station this is + when a callsign is not unique -- the SSID, the number after the dash -- + and whether a digipeater has already repeated the frame. + """ + + call: str = "" + ssid: int = 0 + repeated: bool = False # the H bit: a digipeater has used this hop + reserved: int = 0b11 # the two bits nobody uses + command: bool = False # the C bit, meaningful only on the first two + + def __str__(self) -> str: + out = f"{self.call}-{self.ssid}" if self.ssid else self.call + return out + "*" if self.repeated else out + + @property + def plain(self) -> str: + """Without the asterisk, for looking a station up.""" + return f"{self.call}-{self.ssid}" if self.ssid else self.call + + +def address_from(raw: bytes) -> Address: + """Seven bytes into a callsign. Everything is shifted up by one bit.""" + call = "".join(chr(byte >> 1) for byte in raw[:6]).rstrip() + last = raw[6] + return Address(call=call, ssid=(last >> 1) & 0x0F, + repeated=bool(last & 0x80), + reserved=(last >> 5) & 0x03, + command=bool(last & 0x80)) + + +def address_bytes(address: Address, last: bool = False) -> bytes: + """One callsign back into seven bytes, ready to send.""" + call = (address.call.upper() + " ")[:6] + flags = ((address.ssid & 0x0F) << 1) | (0x60 if address.reserved else 0) + if address.repeated: + flags |= 0x80 + return bytes((ord(c) << 1) & 0xFE for c in call) + bytes([flags | int(last)]) + + +@dataclass +class Frame: + """One AX.25 frame whose frame-check sequence was correct.""" + + destination: Address = field(default_factory=Address) + source: Address = field(default_factory=Address) + path: tuple = () + control: int = UI_CONTROL + pid: int = NO_LAYER_3 + info: bytes = b"" + raw: bytes = b"" + at: float = 0.0 # when it arrived, as a clock time + snr: float = 0.0 # how far above the noise, in dB + + @property + def unnumbered_information(self) -> bool: + """Whether this is the frame type APRS uses, and nothing else does. + + The control byte has its two low bits set on an unnumbered frame, and + the rest of it says which sort. APRS is always UI with no layer 3; + anything else on this channel is somebody running a real AX.25 + connection, which is a fair thing to see and not an APRS packet. + """ + return (self.control & 0xEF) == UI_CONTROL and self.pid == NO_LAYER_3 + + @property + def kind(self) -> str: + """What sort of AX.25 frame this is, in words.""" + if not self.control & 0x01: + return "information" + if (self.control & 0x03) == 0x01: + return {0x00: "receive ready", 0x04: "receive not ready", + 0x08: "reject", 0x0C: "selective reject"}.get( + self.control & 0x0C, "supervisory") + return {0x03: "unnumbered information", 0x2F: "set async balanced", + 0x43: "disconnect", 0x0F: "disconnect mode", + 0x63: "unnumbered ack", 0x87: "frame reject"}.get( + self.control & 0xEF, "unnumbered") + + @property + def heard_through(self) -> tuple: + """The digipeaters that actually repeated this, in order. + + The ones with the H bit set, which is how a station says "I passed + this on". The rest of the path is where it has yet to go. + """ + return tuple(hop for hop in self.path if hop.repeated) + + def route(self) -> str: + """The frame's path the way every APRS tool in the world writes it.""" + parts = [self.source.plain, self.destination.plain] + parts += [str(hop) for hop in self.path] + return ">".join(parts[:2]) + ("," + ",".join(parts[2:]) + if len(parts) > 2 else "") + + def text(self) -> str: + """The information field as characters, for reading and for parsing.""" + return self.info.decode("latin-1") + + def describe(self) -> str: + body = self.text() + return f"{self.route()}:{body}" if body else self.route() + + +# --------------------------------------------------------------------------- +# The check that makes any of this safe to run +# --------------------------------------------------------------------------- + +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 frame_from(raw: bytes) -> Frame | None: + """One frame out of its bytes, or None if it is not one. + + The check is run first and nothing else is looked at until it passes. + Every field below is read on the strength of sixteen bits of CRC saying + the bytes are what was sent. + """ + if len(raw) < 7 * 2 + 2 + 2: + return None + body, check = raw[:-2], raw[-1] << 8 | raw[-2] + if fcs(body) != check: + return None + + addresses, at = [], 0 + while at + 7 <= len(body) and len(addresses) < 10: + field_ = body[at:at + 7] + addresses.append(address_from(field_)) + at += 7 + if field_[6] & 0x01: # the end-of-address bit + break + else: + return None + if len(addresses) < 2 or at + 2 > len(body): + return None + if not all(_sane_call(a.call) for a in addresses): + return None + return Frame(destination=addresses[0], source=addresses[1], + path=tuple(addresses[2:]), control=body[at], + pid=body[at + 1], info=body[at + 2:], raw=raw) + + +def _sane_call(call: str) -> bool: + """Whether a callsign is one, rather than seven bits of luck. + + Sixteen bits of CRC is a strong check and this is a crowded band; a frame + whose addresses are unprintable is one that passed the check by accident, + and there is no reason to put it on a display. + """ + return bool(call) and all(c.isalnum() or c == "-" for c in call) \ + and call.isascii() and call.upper() == call + + +def frame_bytes(source, destination, info: str | bytes = b"", + path=(), control: int = UI_CONTROL, + pid: int = NO_LAYER_3) -> bytes: + """A complete frame, check included, ready to be keyed out. + + Kept beside the decoder so the two cannot drift apart, and so a test can + put a packet in and take the same one out. Callsigns may be given as + strings -- "W1AW-5", "WIDE2-1*" -- or as addresses. + """ + hops = tuple(_as_address(hop) for hop in path) + out = address_bytes(_as_address(destination)) + out += address_bytes(_as_address(source), last=not hops) + for i, hop in enumerate(hops): + out += address_bytes(hop, last=(i == len(hops) - 1)) + out += bytes([control & 0xFF, pid & 0xFF]) + out += info.encode("latin-1") if isinstance(info, str) else bytes(info) + return out + bytes([fcs(out) & 0xFF, (fcs(out) >> 8) & 0xFF]) + + +def _as_address(value) -> Address: + if isinstance(value, Address): + return value + text = str(value).strip().upper() + repeated = text.endswith("*") + text = text.rstrip("*") + call, _, ssid = text.partition("-") + return Address(call=call, ssid=int(ssid) if ssid.isdigit() else 0, + repeated=repeated) + + +# --------------------------------------------------------------------------- +# HDLC: where one frame ends and the next begins +# --------------------------------------------------------------------------- + +def stuff(bits: str) -> str: + """Insert a zero after every five ones, so no flag can occur inside.""" + out, ones = [], 0 + for bit in bits: + out.append(bit) + if bit == "1": + ones += 1 + if ones == 5: + out.append("0") + ones = 0 + else: + ones = 0 + return "".join(out) + + +def unstuff(bits: str) -> str: + """Take those zeroes back out again.""" + out, ones = [], 0 + for bit in bits: + if ones == 5: + ones = 0 + if bit == "0": + continue # the stuffed bit + out.append(bit) + ones = ones + 1 if bit == "1" else 0 + return "".join(out) + + +def nrzi(bits: str, level: str = "1") -> str: + """Encode: a zero is sent as a change of tone, a one as no change.""" + out = [] + for bit in bits: + if bit == "0": + level = "0" if level == "1" else "1" + out.append(level) + return "".join(out) + + +def un_nrzi(bits: str) -> str: + """Decode the same, which needs no knowledge of which tone is which. + + A one is no change and a zero is a change, so inverting the whole stream + -- swapping mark for space, or wiring a discriminator up backwards -- + decodes to exactly the same data. That is the point of the coding and is + why nothing here ever has to guess at polarity. + """ + return "".join("1" if a == b else "0" for a, b in zip(bits, bits[1:])) + + +def hdlc_frames(bits: str, most: int = 64) -> tuple[list[bytes], int]: + """Split a bit stream at flags and undo the stuffing. + + Returns the frames and how far along the stream was consumed, so a caller + reading a continuous signal knows what it may forget. + """ + frames: list[bytes] = [] + at = bits.find(FLAG) + if at < 0: + return frames, max(0, len(bits) - len(FLAG)) + used = at + while len(frames) < most: + while bits.startswith(FLAG, at): # flags repeat between frames + at += 8 + used = at - 8 # the last flag fully passed + end = bits.find(FLAG, at) + if end < 0: + break # a body still arriving; keep it + body, at = bits[at:end], end + # Everything before this flag has been dealt with. Said here rather + # than at the top of the loop because a caller reading a continuous + # signal trims its buffer by this, and trimming to before a frame + # that has already been reported hands it back again on the next + # block -- every packet counted twice, for ever. + used = end + if len(body) < 8 * 17: # shorter than an empty frame + continue + clean = unstuff(body) + whole = len(clean) - len(clean) % 8 + if not 17 <= whole // 8 <= MAX_FRAME: + continue + # Least significant bit first on the air, which is the one thing + # about AX.25 that catches everybody out. + frames.append(bytes(int(clean[i:i + 8][::-1], 2) + for i in range(0, whole, 8))) + return frames, max(used, 0) + + +# --------------------------------------------------------------------------- +# From audio to frames, without ever stopping +# --------------------------------------------------------------------------- + +# How hard the sampling instant is pulled towards the middle of a bit at each +# zero crossing. Low enough that noise cannot drag it about, high enough to +# pull in within a flag or two of the start of a transmission -- which is +# what the flags at the front of every frame are there to allow. +LOOP_GAIN = 0.15 + +# How much of a bit stream to carry over when no flag has been seen. A frame +# is at most three hundred bytes, so anything older than that has no frame +# in it that has not already been found. +KEEP_BITS = MAX_FRAME * 8 * 2 + + +class Receiver: + """Audio in, frames out, across as many blocks as you care to feed it. + + The state that has to survive a block boundary is the whole point of this + being a class: the tail of the audio, so the correlators see no edge; the + phase of the sampling loop, so a bit is not lost or gained where one + block meets the next; the level the tone was last at, for the NRZI; and + the bits themselves, so a frame that began in one block and ended in + another is read straight through rather than halved. + + A packet takes most of a second at twelve hundred baud and blocks are + about that long, so frames straddling a boundary are not an edge case -- + they are most of them. + """ + + def __init__(self, rate: float, baud: float = BAUD, + mark: float = MARK_HZ, space: float = SPACE_HZ): + self.rate = float(rate) + self.baud = float(baud) + self.window = max(4, int(round(self.rate / self.baud))) + self.frames = 0 + self.bytes_seen = 0 + turn = 2.0 * math.pi * np.arange(self.window) / self.rate + self._mark = (np.cos(mark * turn), np.sin(mark * turn)) + self._space = (np.cos(space * turn), np.sin(space * turn)) + self._tail = np.zeros(self.window - 1, dtype=np.float64) + self._phase = 0.0 + self._was = 0.0 # the last soft sample, for crossings + self._level = "1" # the tone the line was last at + self._bits = "" + + # -- the three stages ------------------------------------------------ + def soft(self, audio: np.ndarray) -> np.ndarray: + """How much more mark than space each moment of audio holds. + + Two correlators a bit long, and the difference of their magnitudes. + Positive is a mark. The tail of the previous block is prepended so + that the first bit of this one is measured against real audio rather + than against the zeroes a convolution would otherwise invent. + """ + x = np.concatenate((self._tail, np.asarray(audio, dtype=np.float64))) + if x.size < self.window: + self._tail = x + return np.zeros(0) + self._tail = x[-(self.window - 1):] if self.window > 1 else x[:0] + x = x - x.mean() + out = [] + for cosine, sine in (self._mark, self._space): + i = np.convolve(x, cosine[::-1], mode="valid") + q = np.convolve(x, sine[::-1], mode="valid") + out.append(np.hypot(i, q)) + return out[0] - out[1] + + def slice(self, soft: np.ndarray) -> str: + """Sample the soft signal once a bit, in the middle of the bit. + + The instant is held there by a loop nudged at every zero crossing: + a crossing is a bit boundary, so it should fall half a bit away from + where the last sample was taken, and any difference is an error to + be taken out gently. + """ + step = self.baud / self.rate + phase, was = self._phase, self._was + out = [] + for value in soft: + phase += step + if (value > 0.0) != (was > 0.0): + error = phase - 0.5 if phase < 1.0 else phase - 1.5 + phase -= error * LOOP_GAIN + if phase >= 1.0: + phase -= 1.0 + out.append("1" if value > 0.0 else "0") + was = value + self._phase, self._was = phase, was + return "".join(out) + + def feed(self, audio, when: float = 0.0, snr: float = 0.0) -> list[Frame]: + """One block of audio. Returns whatever frames finished in it.""" + soft = self.soft(audio) + if soft.size == 0: + return [] + tones = self._level + self.slice(soft) + self._level = tones[-1] + self._bits += un_nrzi(tones) + raw, used = hdlc_frames(self._bits) + self._bits = self._bits[used:][-KEEP_BITS:] + + out = [] + for data in raw: + self.bytes_seen += len(data) + frame = frame_from(data) + if frame is not None: + frame.at, frame.snr = when, snr + self.frames += 1 + out.append(frame) + return out + + def reset(self) -> None: + self._tail = np.zeros(self.window - 1, dtype=np.float64) + self._phase, self._was, self._level, self._bits = 0.0, 0.0, "1", "" + + +# --------------------------------------------------------------------------- +# Keying the same thing out, so the receiver can be held to it +# --------------------------------------------------------------------------- + +def bits_of(frame: bytes, flags: int = 8) -> str: + """One frame as the bits that go on the air: flags, stuffing and all.""" + body = "".join(f"{byte:08b}"[::-1] for byte in frame) + return FLAG * flags + stuff(body) + FLAG * flags + + +def modulate(frames, rate: float = 22_050.0, baud: float = BAUD, + mark: float = MARK_HZ, space: float = SPACE_HZ, + flags: int = 8, amplitude: float = 0.5, noise: float = 0.0, + seed: int = 0, quiet_ms: float = 40.0) -> np.ndarray: + """What a receiver would hear: the audio, not the radio signal. + + The tone is continuous in phase across a bit boundary, because a real + modem's is -- it is one oscillator being retuned, not two being switched + -- and a decoder that only ever saw phase jumps at every bit would be + tested against something nothing transmits. + """ + if isinstance(frames, (bytes, bytearray)): + frames = [frames] + stream = "".join(nrzi(bits_of(bytes(f), flags)) for f in frames) + quiet = int(round(rate * quiet_ms / 1000.0)) + per_bit = rate / baud + total = quiet * 2 + int(round(len(stream) * per_bit)) + out = np.zeros(total, dtype=np.float64) + phase, at = 0.0, float(quiet) + for bit in stream: + tone = mark if bit == "1" else space + n = int(round(at + per_bit)) - int(round(at)) + step = 2.0 * math.pi * tone / rate + out[int(round(at)):int(round(at)) + n] = amplitude * np.sin( + phase + step * np.arange(n)) + phase = (phase + step * n) % (2.0 * math.pi) + at += per_bit + if noise: + out = out + np.random.default_rng(seed).normal(0.0, noise, out.size) + return out.astype(np.float32) diff --git a/bandsaunter/cli.py b/bandsaunter/cli.py index 570bc57..3b6b95a 100755 --- a/bandsaunter/cli.py +++ b/bandsaunter/cli.py @@ -25,6 +25,7 @@ from .config import (DEFAULT_CONFIG_DIR, DEFAULT_CONFIG_PATH, ScanConfig, from .device import RtlSdrError, list_devices, set_driver_messages from .librtlsdr import load_error from . import settings as st +from .ax25 import APRS_CHANNELS as _APRS_CHANNELS from .ranges import (RangeError, ScanRange, build_plan, parse_range_list) from .scanner import Scanner, ScannerCallbacks from .tui import TUIAbort, first_run_setup, run_tui, settings_menu @@ -62,6 +63,9 @@ examples: bandsaunter weather --name 1A2B="back fence" name one while listening bandsaunter readings --csv turn a weather log into a graph bandsaunter sensors what is out there, and what it is called + bandsaunter aprs read the APRS channel on 144.39 MHz + bandsaunter aprs --region europe ...or 144.80 MHz, or wherever you are + bandsaunter packets --kml turn an APRS log into a map bandsaunter scan -b 2m --simulate try it without hardware """) # The GNU form: the version, then who holds the copyright and what the @@ -292,6 +296,90 @@ examples: "phosphor, amber and red") ad.set_defaults(log_frames=True, lookup=True) + # -- aprs ----------------------------------------------------------------- + ap = sub.add_parser("aprs", + help="read the APRS channel: positions, weather, " + "messages, telemetry") + ap.add_argument("--seconds", type=float, default=None, + help="stop after this long (default: until interrupted)") + ap.add_argument("--rate", type=float, default=None, + help="sample rate in Hz; 96 kS/s is the least that holds " + "the channel") + ap.add_argument("--gain", default=None, help="tuner gain in dB, or auto") + ap.add_argument("--device", type=int, default=None, help="which receiver") + ap.add_argument("--region", default=None, + choices=[key for key, _hz, _w in _APRS_CHANNELS], + help="which APRS channel to listen on") + ap.add_argument("--frequency", "--freq", dest="frequency", type=float, + default=None, metavar="HZ", + help="the exact frequency, if the region's channel is " + "not what you want") + ap.add_argument("--simulate", dest="simulate", action="store_true", + default=None, + help="invent a channel full of stations, for a receiver " + "with no aerial") + ap.add_argument("--no-simulate", dest="simulate", action="store_false", + default=None, help="listen to real stations") + ap.add_argument("--log", default=None, metavar="FILE", + help="where to write the packet log " + "(default: aprs_