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_