Name the band beside every frequency, and map who was heard
Two additions, both about turning a number into something meaningful. A band column. Next to every frequency -- on the live display, in the line-per-hit output, in saunterbrowse's list and details -- is the name of the band it falls in. 421 MHz is the 70 cm amateur band, and being told so is quicker than remembering where the edges are. The names come from the existing preset table, so there is one band plan to keep right rather than two, but naming is not the job that table was shaped for: several presets cover any frequency, some of them whole-tuner sweeps that say nothing. So the candidates are ranked. Sweeps and the "-complete" duplicates are dropped outright. The narrowest of what is left wins, because it says the most -- 146.52 MHz comes back as the 2 m simplex calling channel rather than as the whole 2 m band. Two exceptions where the narrowest would be the wrong answer: ISM yields to the allocation it shares (433.92 is 70 cm first, 915 is 33 cm first), and shortwave broadcast yields to amateur where the two overlap, because 3.9-4.0 and 7.2-7.3 MHz are Region 1 and 3 broadcast but Region 2 amateur, and this plan is documented as Region 2. 6 MHz really is 49 m shortwave and is left alone. The name is written into each capture's sidecar, so it travels with the recording and an edit to the plan later cannot rewrite history, and saunterbrowse searches on it: /70 cm finds the band without anyone having to remember 420-450 MHz. A map. A licence says where its holder is, so a list of callsigns is also a map. Callsigns heard during a scan are now looked up as the transcripts come in, announced on the display, and written to callsigns.kml in the output directory; saunterbrowse --kml builds the same file from recordings already on disk, and the two continue one map rather than starting two. One placemark per station, not one per transmission: the same repeater heard twenty times in an evening is one operator, and twenty pins on one rooftop would say less than one. Each pin carries the callsign, the licensee, the town, the grid square, and every frequency and time it was heard on. The file is read back on open and added to, so later scans build it up rather than replacing it. Where a licence has no coordinates the grid square's centre is used and the placemark says so -- a square is kilometres across where an address is a street. A callsign with no licence at all is still recorded, in a folder that starts switched off, because that a station was heard is worth keeping even when nothing says where. A file already there that is not readable as KML is never overwritten. Also fixed along the way: - The hit list's "no signals recorded yet" placeholder was one cell short of its row, so it landed in the SNR column and wrapped, making the panel taller than the layout had budgeted for and scrolling the display off a short terminal. The identification column can no longer wrap either, which is what _hit_capacity has always assumed. - Licence lookups now record coordinates. The cache is versioned so that entries written before this are asked about again, rather than pinning every station to its grid square for good. - CallsignBook.wait dropped joined threads; an all-night scan calls it after every transcript and the list only ever grew. - Tests redirect XDG_CACHE_HOME, so a run no longer reads or writes the real lookup cache. 676 -> 761 tests.
This commit is contained in:
The Dust Council
parent
739a2faaf4
commit
fb2bb3344b
23 changed files with 2155 additions and 42 deletions
68
README.md
68
README.md
|
|
@ -583,6 +583,41 @@ Each result carries a confidence and the reasoning behind it:
|
|||
|
||||
Low SNR reduces confidence rather than producing a confident wrong answer.
|
||||
|
||||
### Which band it is in
|
||||
|
||||
Next to every frequency, on the live display and in the line-per-hit output,
|
||||
is the name of the band it falls in:
|
||||
|
||||
```
|
||||
time frequency band dur SNR identified as
|
||||
21:14:07 146.52 MHz 2 m FM Simplex 5.0s 20.0 Narrowband FM voice
|
||||
21:14:31 462.5625 MHz GMRS / FRS 4.2s 18.3 Narrowband FM voice
|
||||
21:15:02 421 MHz 70 cm Amateur 12.7s 22.9 Narrowband FM voice
|
||||
21:15:40 162.55 MHz NOAA Weather Radio 30.0s 31.4 Narrowband FM voice
|
||||
```
|
||||
|
||||
`421 MHz` is the 70 cm amateur band, and being told so is quicker than
|
||||
remembering where the band edges are. The names come from the same band plan
|
||||
the presets do, so there is one table to keep right rather than two.
|
||||
|
||||
Several allocations usually cover any given frequency, and the narrowest wins
|
||||
because it says the most: `146.52 MHz` comes back as *2 m FM Simplex* rather
|
||||
than *2 m Amateur*, and `14.050 MHz` as *20 m CW / Digital*. Two exceptions,
|
||||
both because the obvious answer would be the wrong one:
|
||||
|
||||
- **ISM yields to the allocation it shares.** 433.92 and 915 MHz are ISM
|
||||
bands, but they are also 70 cm and 33 cm. A signal there is far more likely
|
||||
to be worth naming as the amateur band, so it is — unless nothing else
|
||||
covers it, in which case ISM is still the right answer.
|
||||
- **Shortwave broadcast yields to amateur, where they overlap.** 3.9–4.0 and
|
||||
7.2–7.3 MHz are broadcast in ITU Regions 1 and 3, and amateur in Region 2,
|
||||
which is what this plan describes. 6 MHz really is 49 m shortwave, and
|
||||
there is no amateur band anywhere near it, so that one is left alone.
|
||||
|
||||
The band name is written into each recording's sidecar too, so it travels with
|
||||
the capture, and `saunterbrowse` will search on it — typing `/70 cm` finds
|
||||
everything in the band without having to remember 420–450 MHz.
|
||||
|
||||
### Trunked systems and their control channels
|
||||
|
||||
Police, fire and most large business radio in the US runs on *trunked*
|
||||
|
|
@ -1058,6 +1093,39 @@ there is — callook.info holds US licences only.
|
|||
US amateur licence records are public by law and include the licensee's
|
||||
address; that is what is shown.
|
||||
|
||||
### The map
|
||||
|
||||
A licence says where its holder is, so a list of callsigns is also a map. The
|
||||
scanner writes one as it runs — `callsigns.kml` in the output directory —
|
||||
which opens in Google Earth, QGIS, Marble or OsmAnd:
|
||||
|
||||
```bash
|
||||
saunterbrowse --kml # build one from recordings already on disk
|
||||
saunterbrowse --kml ~/heard.kml # or somewhere else
|
||||
bandsaunter scan --kml "" # turn it off
|
||||
```
|
||||
|
||||
Each station is **one placemark, not one per transmission**. Hearing the same
|
||||
repeater twenty times in an evening is one operator, and twenty pins stacked on
|
||||
the same rooftop would say less than one. The pin carries the callsign, the
|
||||
licensee, the town, the grid square, and every frequency and time you heard
|
||||
them, so clicking it answers "when did I hear this, and where on the dial".
|
||||
|
||||
The file is added to rather than replaced — by later scans, and by
|
||||
`saunterbrowse --kml` over the same directory — so over a few weeks it stops
|
||||
being a snapshot of one evening and becomes a picture of what your aerial can
|
||||
actually reach.
|
||||
|
||||
Where a licence carries no coordinates the grid square is used instead, and the
|
||||
placemark says so: a grid square is kilometres across where a licensed address
|
||||
is a street. A callsign with no licence on file at all is still recorded, in a
|
||||
folder named *no location on file* which starts switched off — that a station
|
||||
was heard is worth keeping even when nothing says where it was.
|
||||
|
||||
It is XML, written atomically, so a scan interrupted halfway through leaves a
|
||||
file that still opens. A file already there that is *not* readable as KML is
|
||||
never overwritten.
|
||||
|
||||
### Searching what was said
|
||||
|
||||
Searching the transcripts is the point of it: *"did anyone mention the
|
||||
|
|
|
|||
|
|
@ -8,8 +8,8 @@ and transcribing speech.
|
|||
# Versions are the release date and a revision within that day, so
|
||||
# 2026-08-21_02 is the second build made on the 21st. The revision is padded
|
||||
# to two digits so versions sort as text.
|
||||
VERSION_DATE = "2026-08-22"
|
||||
VERSION_REVISION = 4
|
||||
VERSION_DATE = "2026-08-28"
|
||||
VERSION_REVISION = 1
|
||||
|
||||
__version__ = f"{VERSION_DATE}_{VERSION_REVISION:02d}"
|
||||
|
||||
|
|
|
|||
|
|
@ -8,10 +8,13 @@ NA channel plans (ITU Region 2).
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
__all__ = ["BandPreset", "PRESETS", "CATEGORIES", "by_key", "search",
|
||||
"in_category", "presets_covering", "expand_preset"]
|
||||
"in_category", "presets_covering", "expand_preset",
|
||||
"BandLabel", "label_for", "band_label", "band_name",
|
||||
"band_names"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
|
@ -535,3 +538,155 @@ def expand_preset(key: str) -> list[BandPreset]:
|
|||
"""The presets a key stands for, following groups to their members."""
|
||||
preset = by_key(key)
|
||||
return preset.expand() if preset is not None else []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Saying what a frequency is
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# The preset table exists to *tune*: a preset is a span, a step and a
|
||||
# demodulator. Naming a frequency is a different job, and the table is not
|
||||
# shaped for it -- several presets cover any given frequency, some of them
|
||||
# whole-tuner sweeps that say nothing at all. These functions rank the
|
||||
# candidates and pick the one an operator would actually name.
|
||||
|
||||
# Presets that exist to be scanned, not to describe: a hit at 421 MHz is in
|
||||
# the 70 cm band, and calling it "Full UHF Sweep" is worse than saying
|
||||
# nothing. "complete" presets duplicate a narrower one with the same name.
|
||||
LABEL_SKIP_TAGS = frozenset({"sweep", "complete"})
|
||||
|
||||
# Unlicensed device allocations overlap real ones. 433.92 and 915 MHz are
|
||||
# ISM, but they are also 70 cm and 33 cm, and a signal there is far more
|
||||
# likely to be worth naming as the amateur band. Demoted, not dropped: where
|
||||
# ISM is the only thing covering a frequency it is still the right answer.
|
||||
LABEL_DEMOTED_TAGS = frozenset({"ism"})
|
||||
|
||||
# Where both cover a frequency, the second tag names it. The shortwave
|
||||
# broadcast bands are a Region 1 and 3 allocation; this table is documented
|
||||
# as Region 2, where 3.9-4.0 and 7.2-7.3 MHz are amateur and nothing else.
|
||||
# Conditional rather than a flat demotion, because 6 MHz really is 49 m
|
||||
# shortwave and there is no amateur band anywhere near it.
|
||||
LABEL_YIELDS: tuple[tuple[str, str], ...] = (("swl", "ham"),)
|
||||
|
||||
# Two presets whose spans are within this of each other are the same band
|
||||
# described twice -- GMRS/FRS and the FRS simplex channels differ by one
|
||||
# channel. Prefer whichever the table lists first, which is the broader
|
||||
# name.
|
||||
_LABEL_TIE = 1.10
|
||||
|
||||
_INDEX = {p.key: i for i, p in enumerate(PRESETS)}
|
||||
|
||||
|
||||
def _demoted(p: BandPreset, present: frozenset[str]) -> int:
|
||||
tags = set(p.tags)
|
||||
if tags & LABEL_DEMOTED_TAGS:
|
||||
return 1
|
||||
return 1 if any(loser in tags and winner in present
|
||||
for loser, winner in LABEL_YIELDS) else 0
|
||||
|
||||
|
||||
def _rank(p: BandPreset, present: frozenset[str]) -> tuple[int, float, int]:
|
||||
return (_demoted(p, present), p.span, _INDEX.get(p.key, 0))
|
||||
|
||||
|
||||
def label_presets(hz: float) -> list[BandPreset]:
|
||||
"""Every preset that names ``hz``, most specific first."""
|
||||
hits = [p for p in presets_covering(hz)
|
||||
if not (set(p.tags) & LABEL_SKIP_TAGS)]
|
||||
present = frozenset(t for p in hits for t in p.tags)
|
||||
hits.sort(key=lambda p: _rank(p, present))
|
||||
if len(hits) > 1:
|
||||
# Among presets that are effectively the same width, the table's own
|
||||
# order decides, so the canonical name wins over a sub-segment of it.
|
||||
best = hits[0]
|
||||
close = [p for p in hits
|
||||
if _demoted(p, present) == _demoted(best, present)
|
||||
and p.span <= best.span * _LABEL_TIE]
|
||||
if len(close) > 1:
|
||||
first = min(close, key=lambda p: _INDEX.get(p.key, 0))
|
||||
hits.remove(first)
|
||||
hits.insert(0, first)
|
||||
return hits
|
||||
|
||||
|
||||
_PAREN = re.compile(r"\s*\([^)]*\)")
|
||||
|
||||
|
||||
def shorten_band(name: str, width: int = 0) -> str:
|
||||
"""A band name that fits in a column, without becoming a different name.
|
||||
|
||||
Parentheses are the first thing to go: they hold a restatement of the
|
||||
frequency ("ADS-B (1090 MHz)") next to a number the display is already
|
||||
showing. Only if that is still too long is the name cut at a separator,
|
||||
which keeps "800 MHz Public Safety" rather than truncating mid-word.
|
||||
"""
|
||||
name = _PAREN.sub("", name)
|
||||
name = " ".join(name.split())
|
||||
if not width or len(name) <= width:
|
||||
return name
|
||||
for sep in (" / ", " - ", " — "):
|
||||
head = name.split(sep)[0]
|
||||
if len(head) <= width:
|
||||
return head
|
||||
return name[:max(1, width - 1)].rstrip() + "…"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BandLabel:
|
||||
"""What a frequency is, in as many words as there is room for."""
|
||||
|
||||
hz: float
|
||||
presets: tuple[BandPreset, ...] = ()
|
||||
|
||||
def __bool__(self) -> bool:
|
||||
return bool(self.presets)
|
||||
|
||||
@property
|
||||
def best(self) -> BandPreset | None:
|
||||
return self.presets[0] if self.presets else None
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return self.presets[0].name if self.presets else ""
|
||||
|
||||
@property
|
||||
def key(self) -> str:
|
||||
return self.presets[0].key if self.presets else ""
|
||||
|
||||
@property
|
||||
def category(self) -> str:
|
||||
return self.presets[0].category if self.presets else ""
|
||||
|
||||
@property
|
||||
def names(self) -> list[str]:
|
||||
return [p.name for p in self.presets]
|
||||
|
||||
def short(self, width: int = 0) -> str:
|
||||
return shorten_band(self.name, width) if self.presets else ""
|
||||
|
||||
def describe(self) -> str:
|
||||
"""Name and service, for somewhere with a whole line to spare."""
|
||||
if not self.presets:
|
||||
return ""
|
||||
best = self.presets[0]
|
||||
return f"{best.name} ({best.category})" if best.category != best.name \
|
||||
else best.name
|
||||
|
||||
|
||||
def label_for(hz: float) -> BandLabel:
|
||||
return BandLabel(float(hz), tuple(label_presets(hz)))
|
||||
|
||||
|
||||
def band_label(hz: float, width: int = 0) -> str:
|
||||
"""The short name of the band ``hz`` falls in, or "" if none does."""
|
||||
return label_for(hz).short(width)
|
||||
|
||||
|
||||
def band_name(hz: float) -> str:
|
||||
"""The full name of the band ``hz`` falls in, or "" if none does."""
|
||||
return label_for(hz).name
|
||||
|
||||
|
||||
def band_names(hz: float, limit: int = 3) -> list[str]:
|
||||
"""Every band naming ``hz``, most specific first."""
|
||||
return label_for(hz).names[:limit]
|
||||
|
|
|
|||
|
|
@ -36,9 +36,10 @@ from rich.table import Table
|
|||
from rich.text import Text
|
||||
|
||||
from . import __version__
|
||||
from .bandplan import fmt_hz
|
||||
from .bandplan import band_names, fmt_hz, shorten_band
|
||||
from .callsign import CallsignBook, HEADING, find_callsigns
|
||||
from .config import load_default
|
||||
from .kml import DEFAULT_KML_NAME, KmlLog
|
||||
|
||||
__all__ = ["main", "Capture", "scan_directory", "Browser", "Player"]
|
||||
|
||||
|
|
@ -146,6 +147,24 @@ class Capture:
|
|||
return self._calls
|
||||
|
||||
# -- derived ----------------------------------------------------------
|
||||
@property
|
||||
def bands(self) -> list[str]:
|
||||
"""What the frequency is, most specific first.
|
||||
|
||||
Preferring the sidecar means a recording keeps the band it was filed
|
||||
under even if the band plan is edited later; falling back to the
|
||||
frequency means recordings from before bands were recorded at all
|
||||
still get named.
|
||||
"""
|
||||
saved = self.meta.get("bands") or self.meta.get("band_labels") or []
|
||||
names = [str(b) for b in saved if str(b).strip()]
|
||||
return names or (band_names(self.frequency) if self.frequency else [])
|
||||
|
||||
@property
|
||||
def band(self) -> str:
|
||||
bands = self.bands
|
||||
return bands[0] if bands else ""
|
||||
|
||||
@property
|
||||
def category(self) -> str:
|
||||
if self.combined:
|
||||
|
|
@ -476,6 +495,10 @@ class Browser:
|
|||
return True
|
||||
if q in cap.classification.lower() or q in cap.category.lower():
|
||||
return True
|
||||
# By band as well as by frequency: "70 cm" is how an operator thinks
|
||||
# of a range, and remembering 420-450 MHz is not the point.
|
||||
if any(q in band.lower() for band in cap.bands):
|
||||
return True
|
||||
return q in cap.transcript.lower()
|
||||
|
||||
@property
|
||||
|
|
@ -675,9 +698,12 @@ class Browser:
|
|||
style="yellow")
|
||||
|
||||
second = Text()
|
||||
bands = cap.meta.get("band_labels") or []
|
||||
bands = cap.bands
|
||||
if bands:
|
||||
second.append(" · ".join(bands[:3]), style="bright_black")
|
||||
second.append(bands[0], style="magenta")
|
||||
if len(bands) > 1:
|
||||
second.append(" " + " · ".join(bands[1:3]),
|
||||
style="bright_black")
|
||||
elif cap.meta.get("range_label"):
|
||||
second.append(str(cap.meta["range_label"]), style="bright_black")
|
||||
else:
|
||||
|
|
@ -1023,6 +1049,11 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
p.add_argument("--callsigns", action="store_true",
|
||||
help="print every callsign heard, with who it belongs to, "
|
||||
"and exit")
|
||||
p.add_argument("--kml", nargs="?", const=DEFAULT_KML_NAME, default=None,
|
||||
metavar="FILE",
|
||||
help="write a map of where the stations heard are "
|
||||
f"licensed, and exit (default: {DEFAULT_KML_NAME} "
|
||||
"in the recordings directory)")
|
||||
p.add_argument("--no-lookup", dest="lookup", action="store_false",
|
||||
help="do not contact the licence database; callsigns are "
|
||||
"still found, and described from their prefix alone")
|
||||
|
|
@ -1031,6 +1062,56 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
return p
|
||||
|
||||
|
||||
def _write_kml(console: Console, browser: "Browser", book: CallsignBook,
|
||||
where: str) -> int:
|
||||
"""Build a map from the transcripts already on disk.
|
||||
|
||||
The same file the scanner writes, so running this over an old directory
|
||||
and then scanning again continues one map rather than starting a second.
|
||||
"""
|
||||
heard: dict[str, list[Capture]] = {}
|
||||
for cap in browser.view:
|
||||
for call in cap.callsigns:
|
||||
heard.setdefault(call, []).append(cap)
|
||||
if not heard:
|
||||
console.print("[yellow]no callsigns in any transcript here — "
|
||||
"nothing to map[/yellow]")
|
||||
return 1
|
||||
book.get_all(heard)
|
||||
book.wait(15.0)
|
||||
book.save()
|
||||
|
||||
path = Path(where).expanduser()
|
||||
if not path.is_absolute() and path.parent == Path("."):
|
||||
path = browser.directory / path.name
|
||||
log = KmlLog(path, title="bandsaunter — stations heard",
|
||||
description=f"Callsigns heard in {browser.directory}, "
|
||||
"placed where their licences say they are.")
|
||||
for call, caps in sorted(heard.items()):
|
||||
entry = book.get(call)
|
||||
for cap in caps:
|
||||
log.add(entry, cap.frequency,
|
||||
cap.started_at or 0.0, cap.band, cap.path.name)
|
||||
written = log.save()
|
||||
if written is None:
|
||||
console.print(f"[red]could not write {path}[/red]")
|
||||
if not log.readable:
|
||||
console.print("[bright_black]There is a file there already that "
|
||||
"is not readable as KML, and overwriting it would "
|
||||
"throw it away.[/bright_black]")
|
||||
return 2
|
||||
placed = sum(1 for c in log.contacts.values() if c.located)
|
||||
console.print(f"[green]{written}[/green]")
|
||||
console.print(f"[bright_black]{len(log)} station(s), {placed} with a "
|
||||
f"position.[/bright_black]")
|
||||
missing = len(log) - placed
|
||||
if missing:
|
||||
console.print(f"[bright_black]{missing} had no licence on file — "
|
||||
"listed in the map under 'no location on file'."
|
||||
"[/bright_black]")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
console = Console()
|
||||
|
|
@ -1081,10 +1162,14 @@ def main(argv: list[str] | None = None) -> int:
|
|||
highlight=False, soft_wrap=True)
|
||||
return 0
|
||||
|
||||
if args.kml is not None:
|
||||
return _write_kml(console, browser, book, args.kml)
|
||||
|
||||
if args.list:
|
||||
for cap in browser.view:
|
||||
when = cap.when.strftime("%Y-%m-%d %H:%M:%S") if cap.when else ""
|
||||
line = (f"{fmt_hz(cap.frequency):>14} {when} "
|
||||
line = (f"{fmt_hz(cap.frequency):>14} {shorten_band(cap.band, 20):<20} "
|
||||
f"{when} "
|
||||
f"{_dur(cap.duration):>7} {cap.category:<8} "
|
||||
f"{cap.transcript.splitlines()[0] if cap.transcript else cap.classification}")
|
||||
console.print(line, highlight=False, soft_wrap=True)
|
||||
|
|
|
|||
|
|
@ -27,12 +27,19 @@ from dataclasses import dataclass
|
|||
from pathlib import Path
|
||||
|
||||
__all__ = ["Callsign", "CallsignBook", "find_callsigns", "describe_prefix",
|
||||
"PHONETIC", "LOOKUP_URL"]
|
||||
"grid_to_latlon", "PHONETIC", "LOOKUP_URL"]
|
||||
|
||||
# The FCC's own licence data, served as JSON without an account or a key.
|
||||
# US callsigns only; everything else resolves to what the prefix alone says.
|
||||
LOOKUP_URL = "https://callook.info/{call}/json"
|
||||
|
||||
# Raised whenever a lookup starts recording something it did not before.
|
||||
# Entries written by an older version are ignored and asked about again,
|
||||
# rather than being kept forever with a field permanently empty -- which is
|
||||
# how a warm cache would otherwise pin every station to its grid square for
|
||||
# good after coordinates were added.
|
||||
CACHE_VERSION = 2
|
||||
|
||||
PHONETIC = {
|
||||
"alpha": "A", "alfa": "A", "bravo": "B", "charlie": "C", "delta": "D",
|
||||
"echo": "E", "foxtrot": "F", "fox": "F", "golf": "G", "hotel": "H",
|
||||
|
|
@ -291,13 +298,43 @@ class Callsign:
|
|||
previous: str = ""
|
||||
trustee: str = ""
|
||||
postcode: str = ""
|
||||
latitude: float = 0.0
|
||||
longitude: float = 0.0
|
||||
from_grid: bool = False # position is the grid square, not the address
|
||||
status: str = "pending" # pending / found / unlisted / offline
|
||||
fetched_at: float = 0.0
|
||||
version: int = 1 # of the record, not of the licence
|
||||
|
||||
@property
|
||||
def known(self) -> bool:
|
||||
return self.status == "found"
|
||||
|
||||
@property
|
||||
def position(self) -> tuple[float, float] | None:
|
||||
"""Where the licence says the station is, or None.
|
||||
|
||||
The grid square is the fallback rather than an equal: it is accurate
|
||||
to a few kilometres at best, where the licensed address is a street.
|
||||
It matters for entries cached before coordinates were recorded, and
|
||||
for the occasional licence that carries a grid and no coordinates.
|
||||
"""
|
||||
if self.latitude or self.longitude:
|
||||
return (self.latitude, self.longitude)
|
||||
return grid_to_latlon(self.grid)
|
||||
|
||||
@property
|
||||
def coarse_position(self) -> bool:
|
||||
"""True when the position came from the grid square, not the licence.
|
||||
|
||||
Recorded when the fallback is taken rather than worked out from the
|
||||
coordinates afterwards: once the square's centre has been stored,
|
||||
there is nothing left to tell it apart from a licensed address.
|
||||
"""
|
||||
if self.from_grid:
|
||||
return True
|
||||
return not (self.latitude or self.longitude) and \
|
||||
grid_to_latlon(self.grid) is not None
|
||||
|
||||
def summary(self) -> str:
|
||||
"""One line: who and where, falling back to what the prefix says."""
|
||||
if self.known:
|
||||
|
|
@ -378,6 +415,42 @@ def split_postcode(line: str) -> tuple[str, str]:
|
|||
return town, m.group(2)
|
||||
|
||||
|
||||
def _as_float(value) -> float:
|
||||
try:
|
||||
return float(str(value).strip())
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
_GRID = re.compile(r"^[A-R]{2}[0-9]{2}(?:[A-X]{2})?$", re.IGNORECASE)
|
||||
|
||||
|
||||
def grid_to_latlon(grid: str) -> tuple[float, float] | None:
|
||||
"""The centre of a Maidenhead locator, or None if it is not one.
|
||||
|
||||
Four characters is a square about 111 km by 70 km at mid latitudes and
|
||||
six is about 4.6 km by 2.8 km, so the answer is the middle of whichever
|
||||
was given rather than its corner.
|
||||
"""
|
||||
grid = (grid or "").strip()
|
||||
if not _GRID.match(grid):
|
||||
return None
|
||||
g = grid.upper()
|
||||
lon = (ord(g[0]) - 65) * 20.0 - 180.0
|
||||
lat = (ord(g[1]) - 65) * 10.0 - 90.0
|
||||
lon += int(g[2]) * 2.0
|
||||
lat += int(g[3]) * 1.0
|
||||
if len(g) >= 6:
|
||||
lon += (ord(g[4]) - 65) * (2.0 / 24.0)
|
||||
lat += (ord(g[5]) - 65) * (1.0 / 24.0)
|
||||
lon += 1.0 / 24.0 # centre of the sub-square
|
||||
lat += 0.5 / 24.0
|
||||
else:
|
||||
lon += 1.0 # centre of the square
|
||||
lat += 0.5
|
||||
return (round(lat, 6), round(lon, 6))
|
||||
|
||||
|
||||
def _cache_path() -> Path:
|
||||
root = os.environ.get("XDG_CACHE_HOME") or "~/.cache"
|
||||
return Path(root).expanduser() / "bandsaunter" / "callsigns.json"
|
||||
|
|
@ -419,6 +492,7 @@ class CallsignBook:
|
|||
except TypeError:
|
||||
continue # written by a version with other fields
|
||||
if entry.status in ("found", "unlisted") and \
|
||||
entry.version >= CACHE_VERSION and \
|
||||
now - entry.fetched_at < self.max_age:
|
||||
self._entries[call] = entry
|
||||
|
||||
|
|
@ -461,10 +535,17 @@ class CallsignBook:
|
|||
return [self.get(c) for c in calls]
|
||||
|
||||
def wait(self, timeout: float = 10.0) -> None:
|
||||
"""Block until the outstanding lookups finish. For scripts, not the UI."""
|
||||
"""Block until the outstanding lookups finish. For scripts, not the UI.
|
||||
|
||||
Finished threads are dropped as they are joined: a scan calls this
|
||||
after every transcript, and a list that only ever grew would end an
|
||||
all-night run holding thousands of dead threads.
|
||||
"""
|
||||
deadline = time.time() + timeout
|
||||
for thread in list(self._threads):
|
||||
thread.join(max(0.0, deadline - time.time()))
|
||||
with self._lock:
|
||||
self._threads = [t for t in self._threads if t.is_alive()]
|
||||
|
||||
def _fetch(self, entry: Callsign) -> None:
|
||||
try:
|
||||
|
|
@ -490,6 +571,9 @@ class CallsignBook:
|
|||
@staticmethod
|
||||
def _apply(entry: Callsign, body: dict) -> None:
|
||||
entry.fetched_at = time.time()
|
||||
# Before the status branch below: an unlisted callsign is a result
|
||||
# too, and must not be asked about again on every run.
|
||||
entry.version = CACHE_VERSION
|
||||
if not isinstance(body, dict) or body.get("status") != "VALID":
|
||||
entry.status = "unlisted"
|
||||
return
|
||||
|
|
@ -503,6 +587,16 @@ class CallsignBook:
|
|||
entry.location, entry.postcode = split_postcode(
|
||||
str(address.get("line2") or ""))
|
||||
entry.grid = str(location.get("gridsquare") or "")
|
||||
entry.latitude = _as_float(location.get("latitude"))
|
||||
entry.longitude = _as_float(location.get("longitude"))
|
||||
if entry.latitude == 0.0 and entry.longitude == 0.0:
|
||||
# Null Island is not where anyone is licensed; a record with no
|
||||
# coordinates comes back as empty strings or zeroes, and plotting
|
||||
# it off the Gulf of Guinea would be worse than not plotting it.
|
||||
fallback = grid_to_latlon(entry.grid)
|
||||
if fallback is not None:
|
||||
entry.latitude, entry.longitude = fallback
|
||||
entry.from_grid = True
|
||||
entry.oper_class = str(current.get("operClass") or "")
|
||||
entry.licence_type = str(body.get("type") or "")
|
||||
entry.expires = str(other.get("expiryDate") or "")
|
||||
|
|
|
|||
|
|
@ -442,6 +442,17 @@ def _print_summary(scanner: Scanner) -> None:
|
|||
if worker.dropped:
|
||||
bits.append(f"{worker.dropped} skipped, the recogniser fell behind")
|
||||
console.print(f"[grey62] {', '.join(bits)}[/grey62]")
|
||||
heard = getattr(scanner, "heard", None)
|
||||
if heard:
|
||||
book = getattr(scanner, "callsigns", None)
|
||||
console.print(f"[green] {len(heard)} callsign(s) heard:[/green]")
|
||||
for call in sorted(heard):
|
||||
entry = book.get(call) if book is not None else None
|
||||
who = entry.summary() if entry is not None else ""
|
||||
console.print(f" [bold]{call:<8}[/bold] [grey62]{who}[/grey62]")
|
||||
kml = getattr(scanner, "kml", None)
|
||||
if kml is not None and len(kml):
|
||||
console.print(f"[grey62] mapped in {kml.path}[/grey62]")
|
||||
if st.control_channels:
|
||||
console.print(f"[yellow] {len(st.control_channels)} trunking control "
|
||||
f"channel(s) skipped:[/yellow]")
|
||||
|
|
|
|||
|
|
@ -124,6 +124,10 @@ class ScanConfig:
|
|||
transcribe_language: str = "en"
|
||||
transcribe_min_seconds: float = 1.0
|
||||
|
||||
# -- callsigns and the map --------------------------------------------
|
||||
callsign_lookup: bool = True # ask the licence database who they are
|
||||
kml_file: str = "callsigns.kml" # map of who was heard ("" = none)
|
||||
|
||||
# -- behaviour ------------------------------------------------------
|
||||
lockout: list[Lockout] = field(default_factory=list)
|
||||
lockout_width: float = 12_500.0
|
||||
|
|
|
|||
412
bandsaunter/kml.py
Normal file
412
bandsaunter/kml.py
Normal file
|
|
@ -0,0 +1,412 @@
|
|||
"""A map of who was heard, written as KML.
|
||||
|
||||
A scan produces a list of callsigns and a licence database knows where each
|
||||
one is licensed, so the two together are a map. KML is what to write it in:
|
||||
Google Earth, QGIS, Marble, gpsbabel and OsmAnd all open it without being
|
||||
told anything, and it is text, so a run that is interrupted leaves a file
|
||||
that still opens.
|
||||
|
||||
The file accumulates. Each callsign becomes one placemark, and hearing the
|
||||
same station again adds a line to that placemark rather than dropping a
|
||||
second pin on top of the first -- twenty overs from the same repeater are one
|
||||
station, not twenty. Loading the existing file first means a scan on Tuesday
|
||||
adds to what Monday found instead of replacing it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from xml.sax.saxutils import escape, quoteattr
|
||||
|
||||
from .bandplan import band_name, fmt_hz
|
||||
from .callsign import Callsign
|
||||
|
||||
__all__ = ["Hearing", "Contact", "KmlLog", "DEFAULT_KML_NAME"]
|
||||
|
||||
DEFAULT_KML_NAME = "callsigns.kml"
|
||||
|
||||
# Stamped on each placemark so a file written by an older version, or edited
|
||||
# by hand, is still recognisably ours when it is read back.
|
||||
_ID_PREFIX = "bandsaunter-"
|
||||
|
||||
_STYLES = """\
|
||||
<Style id="bandsaunter-heard">
|
||||
<IconStyle>
|
||||
<color>ff2fbfff</color>
|
||||
<scale>1.1</scale>
|
||||
<Icon>
|
||||
<href>http://maps.google.com/mapfiles/kml/shapes/target.png</href>
|
||||
</Icon>
|
||||
</IconStyle>
|
||||
<LabelStyle><scale>0.9</scale></LabelStyle>
|
||||
</Style>"""
|
||||
|
||||
|
||||
def _stamp(when: float) -> str:
|
||||
"""An ISO time in UTC, which is the only thing KML accepts."""
|
||||
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(when))
|
||||
|
||||
|
||||
def _local(when: float) -> str:
|
||||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(when))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Hearing:
|
||||
"""One transmission on which a callsign was heard."""
|
||||
|
||||
frequency: float = 0.0
|
||||
at: float = 0.0 # epoch seconds
|
||||
band: str = ""
|
||||
recording: str = "" # filename, not a path: the KML travels
|
||||
|
||||
@property
|
||||
def key(self) -> tuple[int, int]:
|
||||
return (int(round(self.frequency)), int(self.at))
|
||||
|
||||
def describe(self) -> str:
|
||||
bits = [fmt_hz(self.frequency) if self.frequency else "",
|
||||
_local(self.at) if self.at else "",
|
||||
self.band, self.recording]
|
||||
return " ".join(b for b in bits if b)
|
||||
|
||||
@classmethod
|
||||
def parse(cls, line: str) -> "Hearing | None":
|
||||
"""Read back a line this class wrote.
|
||||
|
||||
Deliberately forgiving: a file someone has edited by hand should lose
|
||||
the line that was mangled, not the whole history.
|
||||
"""
|
||||
parts = [p for p in line.split(" ") if p.strip()]
|
||||
if not parts:
|
||||
return None
|
||||
freq, at, band, recording = 0.0, 0.0, "", ""
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if not freq and part[:1].isdigit() and part.endswith("Hz"):
|
||||
freq = _parse_hz(part)
|
||||
elif not at and len(part) == 19 and part[4] == "-":
|
||||
at = _parse_time(part)
|
||||
elif part.lower().endswith((".wav", ".flac", ".ogg")):
|
||||
recording = part
|
||||
elif not band:
|
||||
band = part
|
||||
return cls(freq, at, band, recording)
|
||||
|
||||
|
||||
def _parse_hz(text: str) -> float:
|
||||
try:
|
||||
number, unit = text.split(" ", 1)
|
||||
scale = {"hz": 1.0, "khz": 1e3, "mhz": 1e6, "ghz": 1e9}
|
||||
return float(number) * scale.get(unit.strip().lower(), 1.0)
|
||||
except (ValueError, KeyError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def _parse_time(text: str) -> float:
|
||||
try:
|
||||
return time.mktime(time.strptime(text, "%Y-%m-%d %H:%M:%S"))
|
||||
except ValueError:
|
||||
return 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class Contact:
|
||||
"""One station, and every time it was heard."""
|
||||
|
||||
call: str
|
||||
name: str = ""
|
||||
location: str = ""
|
||||
country: str = ""
|
||||
grid: str = ""
|
||||
oper_class: str = ""
|
||||
latitude: float = 0.0
|
||||
longitude: float = 0.0
|
||||
coarse: bool = False # position came from the grid square
|
||||
hearings: list[Hearing] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def located(self) -> bool:
|
||||
return bool(self.latitude or self.longitude)
|
||||
|
||||
@property
|
||||
def first_heard(self) -> float:
|
||||
times = [h.at for h in self.hearings if h.at]
|
||||
return min(times) if times else 0.0
|
||||
|
||||
@property
|
||||
def last_heard(self) -> float:
|
||||
times = [h.at for h in self.hearings if h.at]
|
||||
return max(times) if times else 0.0
|
||||
|
||||
def add(self, hearing: Hearing) -> bool:
|
||||
"""Record a hearing; False if it was already there."""
|
||||
if any(h.key == hearing.key for h in self.hearings):
|
||||
return False
|
||||
self.hearings.append(hearing)
|
||||
self.hearings.sort(key=lambda h: (h.at, h.frequency))
|
||||
return True
|
||||
|
||||
def update_from(self, entry: Callsign) -> None:
|
||||
"""Fill in what the licence says, without erasing what is known.
|
||||
|
||||
A lookup that failed this time must not blank out a name that a
|
||||
successful lookup wrote last time, which is why every field is only
|
||||
overwritten when the new value is not empty.
|
||||
"""
|
||||
for attr in ("name", "location", "country", "grid", "oper_class"):
|
||||
value = getattr(entry, attr, "")
|
||||
if value:
|
||||
setattr(self, attr, value)
|
||||
position = entry.position
|
||||
if position is not None:
|
||||
self.latitude, self.longitude = position
|
||||
self.coarse = entry.coarse_position
|
||||
|
||||
# -- rendering --------------------------------------------------------
|
||||
def title(self) -> str:
|
||||
return f"{self.call} — {self.name}" if self.name else self.call
|
||||
|
||||
def summary(self) -> str:
|
||||
"""The one line that goes under the pin's name."""
|
||||
bits = [b for b in (self.location or self.country, self.grid) if b]
|
||||
return " · ".join(bits)
|
||||
|
||||
def description(self) -> str:
|
||||
"""The balloon Google Earth shows when the pin is clicked."""
|
||||
rows = []
|
||||
if self.name:
|
||||
rows.append(("Licensee", self.name))
|
||||
if self.location:
|
||||
rows.append(("Licensed at", self.location))
|
||||
elif self.country:
|
||||
rows.append(("Country", self.country))
|
||||
if self.oper_class:
|
||||
rows.append(("Class", self.oper_class.title()))
|
||||
if self.grid:
|
||||
rows.append(("Grid", self.grid))
|
||||
if self.coarse:
|
||||
rows.append(("Position", "centre of the grid square, "
|
||||
"not the licensed address"))
|
||||
html = ["<h3>{}</h3>".format(escape(self.call))]
|
||||
if rows:
|
||||
html.append("<table>")
|
||||
html += ["<tr><td><b>{}</b></td><td>{}</td></tr>".format(
|
||||
escape(k), escape(v)) for k, v in rows]
|
||||
html.append("</table>")
|
||||
if self.hearings:
|
||||
html.append("<p><b>Heard {} time{}</b></p><ul>".format(
|
||||
len(self.hearings), "" if len(self.hearings) == 1 else "s"))
|
||||
html += ["<li>{}</li>".format(escape(h.describe()))
|
||||
for h in self.hearings]
|
||||
html.append("</ul>")
|
||||
return "".join(html)
|
||||
|
||||
def to_xml(self, indent: int = 4) -> str:
|
||||
ident = quoteattr(_ID_PREFIX + self.call)
|
||||
out = [f" <Placemark id={ident}>",
|
||||
f" <name>{escape(self.title())}</name>"]
|
||||
summary = self.summary()
|
||||
if summary:
|
||||
out.append(f" <Snippet>{escape(summary)}</Snippet>")
|
||||
out.append(" <description>"
|
||||
f"{escape(self.description())}</description>")
|
||||
out.append(" <styleUrl>#bandsaunter-heard</styleUrl>")
|
||||
if self.last_heard:
|
||||
# A TimeSpan rather than a TimeStamp: the station was heard over a
|
||||
# period, and Google Earth's time slider can then play the scan
|
||||
# back rather than showing everything at once.
|
||||
out.append(" <TimeSpan>")
|
||||
out.append(f" <begin>{_stamp(self.first_heard)}</begin>")
|
||||
out.append(f" <end>{_stamp(self.last_heard)}</end>")
|
||||
out.append(" </TimeSpan>")
|
||||
out.append(" <ExtendedData>")
|
||||
for key, value in (("callsign", self.call),
|
||||
("licensee", self.name),
|
||||
("location", self.location),
|
||||
("country", self.country),
|
||||
("grid", self.grid),
|
||||
("class", self.oper_class),
|
||||
("coarse", "yes" if self.coarse else "")):
|
||||
if value:
|
||||
out.append(f" <Data name={quoteattr(key)}>"
|
||||
f"<value>{escape(value)}</value></Data>")
|
||||
if self.hearings:
|
||||
body = "\n".join(escape(h.describe()) for h in self.hearings)
|
||||
out.append(' <Data name="heard"><value>'
|
||||
f"{body}</value></Data>")
|
||||
out.append(" </ExtendedData>")
|
||||
if self.located:
|
||||
out.append(" <Point><coordinates>"
|
||||
f"{self.longitude:.6f},{self.latitude:.6f},0"
|
||||
"</coordinates></Point>")
|
||||
out.append(" </Placemark>")
|
||||
if indent == 4:
|
||||
return "\n".join(out)
|
||||
pad = " " * (indent - 4)
|
||||
return "\n".join(pad + line for line in out)
|
||||
|
||||
|
||||
def _tag(element) -> str:
|
||||
return element.tag.rsplit("}", 1)[-1]
|
||||
|
||||
|
||||
def _read_placemark(element) -> Contact | None:
|
||||
"""Rebuild a contact from a placemark this module wrote."""
|
||||
fields: dict[str, str] = {}
|
||||
heard: list[str] = []
|
||||
for data in element.iter():
|
||||
if _tag(data) != "Data":
|
||||
continue
|
||||
name = data.get("name") or ""
|
||||
value = ""
|
||||
for child in data:
|
||||
if _tag(child) == "value":
|
||||
value = child.text or ""
|
||||
if name == "heard":
|
||||
heard = value.splitlines()
|
||||
elif name:
|
||||
fields[name] = value
|
||||
call = fields.get("callsign", "").strip().upper()
|
||||
if not call:
|
||||
return None
|
||||
contact = Contact(call=call,
|
||||
name=fields.get("licensee", ""),
|
||||
location=fields.get("location", ""),
|
||||
country=fields.get("country", ""),
|
||||
grid=fields.get("grid", ""),
|
||||
oper_class=fields.get("class", ""),
|
||||
coarse=bool(fields.get("coarse")))
|
||||
for point in element.iter():
|
||||
if _tag(point) == "coordinates" and point.text:
|
||||
parts = point.text.strip().split(",")
|
||||
if len(parts) >= 2:
|
||||
try:
|
||||
contact.longitude = float(parts[0])
|
||||
contact.latitude = float(parts[1])
|
||||
except ValueError:
|
||||
pass
|
||||
for line in heard:
|
||||
hearing = Hearing.parse(line)
|
||||
if hearing is not None:
|
||||
contact.add(hearing)
|
||||
return contact
|
||||
|
||||
|
||||
class KmlLog:
|
||||
"""The KML file, read on open and rewritten as contacts are added.
|
||||
|
||||
Rewritten rather than appended to because a placemark grows: hearing a
|
||||
station a second time edits the pin that is already there. The file is
|
||||
one placemark per station and stays small, and the write is atomic, so an
|
||||
interrupted scan cannot leave half a document behind.
|
||||
"""
|
||||
|
||||
def __init__(self, path, title: str = "bandsaunter",
|
||||
description: str = ""):
|
||||
self.path = Path(path).expanduser()
|
||||
self.title = title
|
||||
self.description = description
|
||||
self.contacts: dict[str, Contact] = {}
|
||||
self.added = 0 # new stations since this log was opened
|
||||
self.readable = True # cleared if what is there is not KML
|
||||
self.load()
|
||||
|
||||
# -- reading ----------------------------------------------------------
|
||||
def load(self) -> None:
|
||||
"""Take up where an earlier run left off.
|
||||
|
||||
A file that cannot be parsed is left alone and reported as empty, so
|
||||
that a scan never destroys a map it did not understand -- the write
|
||||
below refuses to overwrite in that case.
|
||||
"""
|
||||
self.readable = True
|
||||
try:
|
||||
text = self.path.read_text()
|
||||
except OSError:
|
||||
return
|
||||
try:
|
||||
root = ET.fromstring(text)
|
||||
except ET.ParseError:
|
||||
self.readable = False
|
||||
return
|
||||
for element in root.iter():
|
||||
if _tag(element) != "Placemark":
|
||||
continue
|
||||
contact = _read_placemark(element)
|
||||
if contact is not None:
|
||||
self.contacts[contact.call] = contact
|
||||
|
||||
# -- writing ----------------------------------------------------------
|
||||
def add(self, entry: Callsign, frequency: float = 0.0,
|
||||
when: float = 0.0, band: str = "", recording: str = "") -> bool:
|
||||
"""Note that ``entry`` was heard. True if anything changed."""
|
||||
call = (entry.call or "").strip().upper()
|
||||
if not call:
|
||||
return False
|
||||
contact = self.contacts.get(call)
|
||||
if contact is None:
|
||||
contact = Contact(call=call)
|
||||
self.contacts[call] = contact
|
||||
self.added += 1
|
||||
changed = True
|
||||
else:
|
||||
changed = False
|
||||
before = (contact.name, contact.latitude, contact.longitude)
|
||||
contact.update_from(entry)
|
||||
changed |= (contact.name, contact.latitude,
|
||||
contact.longitude) != before
|
||||
if not band and frequency:
|
||||
band = band_name(frequency)
|
||||
changed |= contact.add(Hearing(float(frequency), float(when or
|
||||
time.time()),
|
||||
band, recording))
|
||||
return changed
|
||||
|
||||
def render(self) -> str:
|
||||
located = sorted((c for c in self.contacts.values() if c.located),
|
||||
key=lambda c: c.call)
|
||||
unlocated = sorted((c for c in self.contacts.values()
|
||||
if not c.located), key=lambda c: c.call)
|
||||
out = ['<?xml version="1.0" encoding="UTF-8"?>',
|
||||
'<kml xmlns="http://www.opengis.net/kml/2.2">',
|
||||
" <Document>",
|
||||
f" <name>{escape(self.title)}</name>"]
|
||||
if self.description:
|
||||
out.append(f" <description>{escape(self.description)}"
|
||||
"</description>")
|
||||
out.append(_STYLES)
|
||||
out += [c.to_xml() for c in located]
|
||||
if unlocated:
|
||||
# Kept rather than dropped: that a station was heard is worth
|
||||
# recording even when no licence says where it was.
|
||||
out.append(" <Folder>")
|
||||
out.append(" <name>no location on file</name>")
|
||||
out.append(" <visibility>0</visibility>")
|
||||
out += [c.to_xml(indent=6) for c in unlocated]
|
||||
out.append(" </Folder>")
|
||||
out += [" </Document>", "</kml>", ""]
|
||||
return "\n".join(out)
|
||||
|
||||
def save(self) -> Path | None:
|
||||
"""Write the file, atomically. None when there was nothing to write."""
|
||||
if not self.contacts:
|
||||
return None
|
||||
if not self.readable:
|
||||
# Something is there that we could not parse. Overwriting it
|
||||
# would throw away whatever it was.
|
||||
return None
|
||||
try:
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self.path.with_name(self.path.name + ".tmp")
|
||||
tmp.write_text(self.render())
|
||||
tmp.replace(self.path)
|
||||
except OSError:
|
||||
return None
|
||||
return self.path
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self.contacts)
|
||||
|
|
@ -11,7 +11,7 @@ from pathlib import Path
|
|||
|
||||
import numpy as np
|
||||
|
||||
from .bandplan import fmt_hz, presets_covering
|
||||
from .bandplan import fmt_hz, label_for
|
||||
|
||||
__all__ = ["Recording", "ScanLog", "HitRecord", "safe_freq_name",
|
||||
"build_stem"]
|
||||
|
|
@ -55,6 +55,7 @@ class HitRecord:
|
|||
mode: str = "nfm"
|
||||
range_label: str = ""
|
||||
band_labels: list[str] = field(default_factory=list)
|
||||
band: str = "" # the one an operator would name it by
|
||||
|
||||
classification: str = ""
|
||||
family: str = ""
|
||||
|
|
@ -334,7 +335,8 @@ class Recording:
|
|||
"audio_samples": self.audio_samples,
|
||||
"iq_samples": self.iq_samples,
|
||||
"range_label": self.range_label,
|
||||
"bands": [p.name for p in presets_covering(self.frequency)],
|
||||
"bands": label_for(self.frequency).names,
|
||||
"band": label_for(self.frequency).name,
|
||||
"hit": hit.to_dict(),
|
||||
}
|
||||
with open(self.meta_path, "w") as fh:
|
||||
|
|
|
|||
|
|
@ -20,11 +20,13 @@ from pathlib import Path
|
|||
import numpy as np
|
||||
|
||||
from . import dsp
|
||||
from .bandplan import fmt_hz, presets_covering
|
||||
from .bandplan import fmt_hz, label_for, presets_covering
|
||||
from .callsign import CallsignBook, find_callsigns
|
||||
from .classify import classify, ssb_alignment
|
||||
from .config import ScanConfig, remember_lockouts
|
||||
from .demod import make_demodulator
|
||||
from .device import RtlSdrDevice, RtlSdrError
|
||||
from .kml import KmlLog
|
||||
from .morse import decode_morse
|
||||
from .quality import Assessment, assess
|
||||
from .ranges import Lockout, TuneStep, build_plan
|
||||
|
|
@ -111,6 +113,9 @@ class Scanner:
|
|||
self.log: ScanLog | None = None
|
||||
self.frequency_log: FrequencyLog | None = None
|
||||
self.transcriber: TranscriptionWorker | None = None
|
||||
self.callsigns: CallsignBook | None = None
|
||||
self.kml: KmlLog | None = None
|
||||
self.heard: dict[str, int] = {} # callsign -> times heard this run
|
||||
self.hits: list[HitRecord] = []
|
||||
self.nfft = 1024
|
||||
self.detector_bias = 0.0
|
||||
|
|
@ -225,9 +230,19 @@ class Scanner:
|
|||
engine=self.cfg.transcribe_engine,
|
||||
model=self.cfg.transcribe_model,
|
||||
language=self.cfg.transcribe_language,
|
||||
on_done=self._on_transcript,
|
||||
on_error=self._error)
|
||||
self.transcriber.start()
|
||||
self._status(f"transcribing speech with {engine}")
|
||||
# Callsigns come out of transcripts, so both of these are
|
||||
# only worth setting up where there will be transcripts.
|
||||
self.callsigns = CallsignBook(online=self.cfg.callsign_lookup)
|
||||
if self.cfg.kml_file:
|
||||
self.kml = KmlLog(
|
||||
root / self.cfg.kml_file,
|
||||
title="bandsaunter — stations heard",
|
||||
description="Every callsign heard during a scan, "
|
||||
"placed where its licence says it is.")
|
||||
|
||||
if self.cfg.combine_by_frequency:
|
||||
self.frequency_log = FrequencyLog(
|
||||
|
|
@ -812,6 +827,7 @@ class Scanner:
|
|||
f"(--record 0) to follow a transmission to its end")
|
||||
|
||||
duration = rec.duration
|
||||
band = label_for(det.frequency)
|
||||
hit = HitRecord(
|
||||
frequency=det.frequency,
|
||||
started_at=rec.started_at,
|
||||
|
|
@ -823,7 +839,8 @@ class Scanner:
|
|||
bandwidth=det.bandwidth,
|
||||
mode=mode,
|
||||
range_label=det.step.range_label,
|
||||
band_labels=[p.name for p in presets_covering(det.frequency)],
|
||||
band_labels=band.names,
|
||||
band=band.name,
|
||||
directory=str(rec.dir),
|
||||
filename=rec.stem,
|
||||
stop_reason=stop_reason,
|
||||
|
|
@ -1040,7 +1057,62 @@ class Scanner:
|
|||
worker.submit(audio, rate, path,
|
||||
datetime.fromtimestamp(rec.started_at), hit.frequency,
|
||||
append=append,
|
||||
meta_path=Path(hit.meta_path) if hit.meta_path else None)
|
||||
meta_path=Path(hit.meta_path) if hit.meta_path else None,
|
||||
recording=rec.audio_path.name)
|
||||
|
||||
def _on_transcript(self, path, result, job) -> None:
|
||||
"""Pull callsigns out of a finished transcript and map them.
|
||||
|
||||
Runs on the transcription thread, well off the scan loop, so the
|
||||
licence lookups it waits on cannot delay the sweep. Everything here
|
||||
is best-effort: a scan must not fail because a website did not
|
||||
answer.
|
||||
"""
|
||||
book = self.callsigns
|
||||
if book is None:
|
||||
return
|
||||
try:
|
||||
calls = find_callsigns(result.text)
|
||||
if not calls:
|
||||
return
|
||||
entries = book.get_all(calls)
|
||||
# The lookups were started by get_all and run in their own
|
||||
# threads; waiting here is what turns "pending" into a name. A
|
||||
# short wait, because a slow answer is not worth holding the
|
||||
# queue for -- the callsign is already in the cache request and
|
||||
# will be filled in by the time the next one is looked up.
|
||||
book.wait(timeout=8.0)
|
||||
book.save()
|
||||
when = job.when.timestamp()
|
||||
band = label_for(job.frequency).name
|
||||
before = len(self.kml) if self.kml is not None else 0
|
||||
changed = False
|
||||
for entry in entries:
|
||||
self.heard[entry.call] = self.heard.get(entry.call, 0) + 1
|
||||
self._announce_callsign(entry)
|
||||
if self.kml is not None:
|
||||
changed |= self.kml.add(entry, job.frequency, when, band,
|
||||
job.recording)
|
||||
if changed and self.kml is not None:
|
||||
written = self.kml.save()
|
||||
# Only when the map gained a station. Every over of a long
|
||||
# net adds a line to a pin that is already there, and saying
|
||||
# so each time would push everything else off the status
|
||||
# line.
|
||||
if written is not None and len(self.kml) > before:
|
||||
self._status(f"{len(self.kml)} station(s) on the map in "
|
||||
f"{written.name}")
|
||||
except Exception as exc:
|
||||
self._error(exc)
|
||||
|
||||
def _announce_callsign(self, entry) -> None:
|
||||
"""Say on the display who has just identified themselves."""
|
||||
if self.heard.get(entry.call, 0) > 1:
|
||||
return # already said, and saying it every over is noise
|
||||
summary = entry.summary()
|
||||
self._status(f"heard {entry.call}"
|
||||
+ (f" — {summary}" if summary and
|
||||
entry.status == "found" else ""))
|
||||
|
||||
def _identify(self, rec: Recording, hit: HitRecord, demod,
|
||||
analysis=None, continuous_for: float = 0.0) -> Assessment | None:
|
||||
|
|
@ -1195,6 +1267,12 @@ class Scanner:
|
|||
if pending:
|
||||
self._status(f"finishing {pending} transcription(s)")
|
||||
self.transcriber.close()
|
||||
# After the worker has stopped, so the last transcript's callsigns
|
||||
# are on the map before it is written for the last time.
|
||||
if self.callsigns is not None:
|
||||
self.callsigns.save()
|
||||
if self.kml is not None:
|
||||
self.kml.save()
|
||||
if self._own_device and self.device is not None:
|
||||
self.device.close()
|
||||
return self.stats
|
||||
|
|
|
|||
|
|
@ -389,6 +389,26 @@ _TABLE: tuple[Setting, ...] = (
|
|||
"rarely produces anything useful.",
|
||||
unit="s", minimum=0.0, flags=("--transcribe-min",), metavar="SEC"),
|
||||
|
||||
# -- callsigns ---------------------------------------------------------
|
||||
S("callsign_lookup", "Look callsigns up", "Callsigns", "bool",
|
||||
"ask the licence database who a callsign belongs to",
|
||||
"Callsigns heard in a transcript are looked up in the FCC's published "
|
||||
"licence data, which gives the licensee's name, town and coordinates. "
|
||||
"The callsign is the only thing sent, and answers are cached, so a net "
|
||||
"logged night after night is looked up once. Turn it off to keep the "
|
||||
"scan entirely offline; callsigns are still found and still named by "
|
||||
"country from their prefix.",
|
||||
flags=("--callsign-lookup",),
|
||||
off_flags=("--no-callsign-lookup", "--offline-callsigns")),
|
||||
S("kml_file", "Map file", "Callsigns", "text",
|
||||
"KML map of where the stations heard are licensed (blank = none)",
|
||||
"Written inside the output directory and added to as the scan runs: "
|
||||
"one pin per station, holding the callsign, the licensee, the town and "
|
||||
"every frequency and time it was heard on. Opens in Google Earth, "
|
||||
"QGIS, Marble and OsmAnd. A later scan adds to the same map rather "
|
||||
"than starting a new one.",
|
||||
flags=("--kml",), metavar="NAME", example="callsigns.kml"),
|
||||
|
||||
S("log_file", "Log file", "Output", "text",
|
||||
"name of the run log inside the output directory",
|
||||
"Written as JSON lines, with a matching .csv alongside it.",
|
||||
|
|
@ -743,6 +763,25 @@ _GUIDANCE: dict[str, str] = {
|
|||
"Do not bother transcribing captures shorter than this. Very short "
|
||||
"clips rarely contain a whole word and mostly produce noise or "
|
||||
"nothing, while still costing the processing.",
|
||||
"callsign_lookup":
|
||||
"When someone gives their callsign, look it up and say who they "
|
||||
"are. The data is the FCC's own published licence register, which "
|
||||
"carries the licensee's name, the town they are licensed in and "
|
||||
"the coordinates that put them on the map. Only the callsign is "
|
||||
"sent, and each one is asked about once and then remembered, so "
|
||||
"this costs almost nothing. Turn it off if you would rather the "
|
||||
"scan reached the network for nothing at all -- callsigns are "
|
||||
"still picked out of the transcripts, and the prefix still says "
|
||||
"which country and which US district they belong to.",
|
||||
"kml_file":
|
||||
"The name of a map file, kept in the output directory, of "
|
||||
"everyone who identified themselves. Each station is one pin: the "
|
||||
"callsign, who holds the licence, where they are licensed, and "
|
||||
"every frequency and time you heard them. It is added to as the "
|
||||
"scan runs and again by later scans, so it builds up into a "
|
||||
"picture of what you can hear from where you are. KML is the "
|
||||
"format Google Earth uses; QGIS, Marble and OsmAnd read it too. "
|
||||
"Leave it blank if you do not want the map.",
|
||||
"max_cycles":
|
||||
"Stop after this many complete passes through all the frequencies. "
|
||||
"Useful for a quick survey of what is active, or for a scripted run "
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ import numpy as np
|
|||
from ._quiet import suppress_stderr
|
||||
|
||||
__all__ = ["Transcript", "transcribe", "available_engine", "ENGINES",
|
||||
"TranscriptionWorker", "describe_engines", "resolve_model",
|
||||
"TranscriptionWorker", "Job", "describe_engines", "resolve_model",
|
||||
"VENDOR_DIR", "MODEL_DIR"]
|
||||
|
||||
# Best first. Whisper handles the noise and clipping of radio audio far
|
||||
|
|
@ -318,7 +318,14 @@ def transcribe(audio: np.ndarray, rate: float, engine: str = "auto",
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class _Job:
|
||||
class Job:
|
||||
"""One capture waiting to be recognised.
|
||||
|
||||
Public because ``on_done`` is handed it: what a transcript says is only
|
||||
half of what a caller wants, the other half being which frequency and
|
||||
which moment it came from.
|
||||
"""
|
||||
|
||||
audio: np.ndarray
|
||||
rate: float
|
||||
path: Path
|
||||
|
|
@ -326,6 +333,7 @@ class _Job:
|
|||
frequency: float
|
||||
append: bool
|
||||
meta_path: Path | None = None
|
||||
recording: str = "" # filename of the audio, for reference
|
||||
|
||||
|
||||
class TranscriptionWorker:
|
||||
|
|
@ -338,6 +346,8 @@ class TranscriptionWorker:
|
|||
def __init__(self, engine: str = "auto", model: str = "",
|
||||
language: str = "en", max_queue: int = 32,
|
||||
on_done=None, on_error=None):
|
||||
# on_done(path, Transcript, Job) -- called on the worker thread once
|
||||
# a transcript has been written, never for a capture that was silent.
|
||||
self.engine = engine
|
||||
self.model = model
|
||||
self.language = language
|
||||
|
|
@ -374,14 +384,14 @@ class TranscriptionWorker:
|
|||
# -- work --------------------------------------------------------------
|
||||
def submit(self, audio: np.ndarray, rate: float, path: Path,
|
||||
when: datetime, frequency: float, append: bool = False,
|
||||
meta_path: Path | None = None) -> bool:
|
||||
meta_path: Path | None = None, recording: str = "") -> bool:
|
||||
if audio is None or audio.size == 0:
|
||||
return False
|
||||
try:
|
||||
self._queue.put_nowait(_Job(np.asarray(audio, dtype=np.float32),
|
||||
float(rate), Path(path), when,
|
||||
float(frequency), append,
|
||||
Path(meta_path) if meta_path else None))
|
||||
self._queue.put_nowait(
|
||||
Job(np.asarray(audio, dtype=np.float32), float(rate),
|
||||
Path(path), when, float(frequency), append,
|
||||
Path(meta_path) if meta_path else None, recording))
|
||||
return True
|
||||
except queue.Full:
|
||||
self.dropped += 1
|
||||
|
|
@ -402,7 +412,7 @@ class TranscriptionWorker:
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
def _handle(self, job: _Job) -> None:
|
||||
def _handle(self, job: Job) -> None:
|
||||
result = transcribe(job.audio, job.rate, self.engine, self.model,
|
||||
self.language)
|
||||
if result is None:
|
||||
|
|
@ -427,12 +437,12 @@ class TranscriptionWorker:
|
|||
self._record_in_metadata(job, body)
|
||||
if self.on_done:
|
||||
try:
|
||||
self.on_done(job.path, result)
|
||||
self.on_done(job.path, result, job)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _record_in_metadata(job: _Job, text: str) -> None:
|
||||
def _record_in_metadata(job: Job, text: str) -> None:
|
||||
"""Note the transcript in the capture's sidecar, once it exists.
|
||||
|
||||
Written here rather than when the capture was queued, so the metadata
|
||||
|
|
|
|||
|
|
@ -559,7 +559,23 @@ It is recognised by its symbol rate and its refusal to pause, named on screen,
|
|||
and skipped within a second or two. Turn 'Skip trunk control channels' off
|
||||
only if you are collecting them for a decoder. If digital voice calls are
|
||||
being skipped by mistake, raise 'Control channel patience'."""),
|
||||
"9": ("Keys during a scan", """
|
||||
"9": ("Callsigns and the map", """
|
||||
Anyone who identifies themselves in a transcript is picked out and looked up in
|
||||
the FCC's published licence register: the name, the town, the class of licence.
|
||||
The callsign is the only thing sent, and each one is asked about once and then
|
||||
remembered, so a net recorded night after night is looked up once.
|
||||
|
||||
Because a licence says where its holder is, the same information is written as
|
||||
a map — callsigns.kml in the output directory — with one pin per station,
|
||||
carrying the licensee, the town, and every frequency and time you heard them.
|
||||
It opens in Google Earth, QGIS, Marble and OsmAnd, and later scans add to it
|
||||
rather than starting it over, so it fills in as a picture of what your aerial
|
||||
can reach.
|
||||
|
||||
Turn 'Look callsigns up' off to keep the scan entirely offline; callsigns are
|
||||
still found, and named by country from their prefix. Clear 'Map file' to stop
|
||||
writing the map. US licence records are public, and include addresses."""),
|
||||
"10": ("Keys during a scan", """
|
||||
q stop the scan
|
||||
p pause and resume
|
||||
s skip the signal being recorded and carry on sweeping
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ from rich.panel import Panel
|
|||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from .bandplan import fmt_hz
|
||||
from .bandplan import band_label, fmt_hz, shorten_band
|
||||
from .recorder import HitRecord
|
||||
from .scanner import Detection, Scanner
|
||||
|
||||
|
|
@ -24,6 +24,10 @@ __all__ = ["ScanDisplay", "KeyReader", "print_hit", "print_band_table"]
|
|||
|
||||
_SPARK = " ▁▂▃▄▅▆▇█"
|
||||
|
||||
# Width of the band column in the line-per-hit output, which has no live
|
||||
# layout to shrink and so fixes it once.
|
||||
_PLAIN_BAND = 20
|
||||
|
||||
|
||||
def _sparkline(values: np.ndarray, width: int = 60,
|
||||
lo: float | None = None, hi: float | None = None) -> str:
|
||||
|
|
@ -109,6 +113,7 @@ class ScanDisplay:
|
|||
|
||||
self._spark = ""
|
||||
self._span = ""
|
||||
self._band = ""
|
||||
self._peak = 0.0
|
||||
self._step_i = 0
|
||||
self._n_steps = 1
|
||||
|
|
@ -137,6 +142,9 @@ class ScanDisplay:
|
|||
self._spark = _sparkline(psd_db[inband], width=self._spark_width())
|
||||
self._peak = float(psd_db[inband].max())
|
||||
self._span = f"{fmt_hz(step.low)} - {fmt_hz(step.high)}"
|
||||
# The band at the middle of the step, so the sweep line says what is
|
||||
# being listened to and not only where.
|
||||
self._band = band_label(0.5 * (step.low + step.high), 24)
|
||||
self._dirty = True
|
||||
|
||||
def on_detection(self, det: Detection):
|
||||
|
|
@ -207,7 +215,8 @@ class ScanDisplay:
|
|||
lines = [
|
||||
Text.from_markup(
|
||||
f"{bar} step {self._step_i + 1}/{self._n_steps} "
|
||||
f"[bold]{self._span}[/bold]"),
|
||||
f"[bold]{self._span}[/bold]"
|
||||
+ (f" [magenta]{self._band}[/magenta]" if self._band else "")),
|
||||
]
|
||||
if show_spectrum and self.show_spectrum and self._spark:
|
||||
lines.append(Text.from_markup(
|
||||
|
|
@ -242,18 +251,20 @@ class ScanDisplay:
|
|||
# being held open across a pause rather than stuck.
|
||||
sq = f"[yellow]gap {min(r.quiet_for, hang):4.1f}/{hang:g}s[/yellow]"
|
||||
limit_s = f"/{limit:g}s" if limit else ""
|
||||
band = band_label(r.frequency, 22)
|
||||
where = f" [magenta]{band}[/magenta]" if band else ""
|
||||
if r.note:
|
||||
# A control channel is not being recorded so much as identified
|
||||
# and abandoned; saying "REC" while that happens is a lie.
|
||||
return Panel(
|
||||
Text.from_markup(
|
||||
f"[bold black on yellow] {r.note} [/bold black on yellow] "
|
||||
f"{fmt_hz(r.frequency)} [{r.mode}] "
|
||||
f"{fmt_hz(r.frequency)}{where} [{r.mode}] "
|
||||
f"SNR {r.snr:5.1f} dB [yellow]skipping[/yellow]"),
|
||||
border_style="yellow", padding=(0, 1))
|
||||
return Panel(
|
||||
Text.from_markup(
|
||||
f"[bold red]REC[/bold red] {fmt_hz(r.frequency)} "
|
||||
f"[bold red]REC[/bold red] {fmt_hz(r.frequency)}{where} "
|
||||
f"[{r.mode}] {bar} {r.elapsed:5.1f}{limit_s} "
|
||||
f"{sq} SNR {r.snr:5.1f} dB"),
|
||||
border_style="red", padding=(0, 1))
|
||||
|
|
@ -293,13 +304,47 @@ class ScanDisplay:
|
|||
"""
|
||||
return self._layout()[1]
|
||||
|
||||
def _band_width(self) -> int:
|
||||
"""How much of the row the band column may take, or 0 for none.
|
||||
|
||||
It is the first thing given up on a narrow terminal: what a signal
|
||||
was identified as matters more than which allocation it sat in, and
|
||||
a band name squeezed to four characters says nothing.
|
||||
"""
|
||||
# time, frequency, duration, SNR, and enough left for the
|
||||
# identification, which is the column that must not be squeezed.
|
||||
spare = self.console.width - 8 - 15 - 6 - 7 - 30
|
||||
if spare < 10:
|
||||
return 0
|
||||
return max(10, min(22, spare))
|
||||
|
||||
@staticmethod
|
||||
def _band_of(hit: HitRecord, width: int) -> str:
|
||||
"""What band a hit sat in, preferring what was recorded with it.
|
||||
|
||||
Taken from the record rather than recomputed so that a hit kept from
|
||||
an older run, or a band plan that has since been edited, still shows
|
||||
the band it was actually filed under.
|
||||
"""
|
||||
name = hit.band or (hit.band_labels[0] if hit.band_labels else "") \
|
||||
or band_label(hit.frequency)
|
||||
return shorten_band(name, width)
|
||||
|
||||
def _hits_table(self) -> Panel:
|
||||
band_w = self._band_width()
|
||||
t = Table(box=None, expand=True, pad_edge=False, show_edge=False)
|
||||
t.add_column("time", style="grey62", width=8)
|
||||
t.add_column("frequency", style="bold cyan", width=15, justify="right")
|
||||
if band_w:
|
||||
t.add_column("band", style="magenta", width=band_w,
|
||||
no_wrap=True, overflow="ellipsis")
|
||||
t.add_column("dur", width=6, justify="right")
|
||||
t.add_column("SNR", width=7, justify="right")
|
||||
t.add_column("identified as", ratio=1, overflow="ellipsis")
|
||||
# One line per hit, always: _hit_capacity counts rows, and a row
|
||||
# that wraps to two pushes the display past the bottom of the
|
||||
# terminal, where it can no longer be redrawn in place.
|
||||
t.add_column("identified as", ratio=1, no_wrap=True,
|
||||
overflow="ellipsis")
|
||||
capacity = self._hit_capacity()
|
||||
shown = list(self.hits)[:capacity]
|
||||
for h in shown:
|
||||
|
|
@ -312,17 +357,25 @@ class ScanDisplay:
|
|||
extra = f" [grey62]{h.baud:.0f} baud[/grey62]"
|
||||
conf = h.confidence
|
||||
colour = "green" if conf >= 0.7 else "yellow" if conf >= 0.45 else "grey62"
|
||||
t.add_row(
|
||||
row = [
|
||||
time.strftime("%H:%M:%S", time.localtime(h.started_at)),
|
||||
fmt_hz(h.frequency),
|
||||
]
|
||||
if band_w:
|
||||
row.append(self._band_of(h, band_w))
|
||||
row += [
|
||||
f"{h.duration:.1f}s",
|
||||
f"{h.snr_db:.1f}",
|
||||
Text.from_markup(
|
||||
f"[{colour}]{h.classification or 'unclassified'}[/{colour}]{extra}"),
|
||||
)
|
||||
]
|
||||
t.add_row(*row)
|
||||
if not self.hits:
|
||||
t.add_row("", "", "", "", Text("no signals recorded yet",
|
||||
style="grey42"))
|
||||
# One cell short of the row and the placeholder lands in a
|
||||
# narrow numeric column and wraps, making the panel taller
|
||||
# than the layout budgeted for.
|
||||
blanks = [""] * (5 if band_w else 4)
|
||||
t.add_row(*blanks, Text("no signals recorded yet", style="grey42"))
|
||||
hidden = len(self.hits) - len(shown)
|
||||
title = "recorded signals"
|
||||
if hidden > 0:
|
||||
|
|
@ -363,15 +416,23 @@ def print_hit(console: Console, hit: HitRecord) -> None:
|
|||
return
|
||||
stamp = time.strftime("%H:%M:%S", time.localtime(hit.started_at))
|
||||
conf = f"{hit.confidence*100:.0f}%"
|
||||
line = (f"[grey62]{stamp}[/grey62] [bold cyan]{fmt_hz(hit.frequency):>14}[/bold cyan]"
|
||||
band = ScanDisplay._band_of(hit, _PLAIN_BAND)
|
||||
# Padded even when empty, so the columns line up down the page whether or
|
||||
# not a particular frequency falls in a band the plan knows.
|
||||
where = f" [magenta]{band:<{_PLAIN_BAND}}[/magenta]"
|
||||
console.print(
|
||||
f"[grey62]{stamp}[/grey62] "
|
||||
f"[bold cyan]{fmt_hz(hit.frequency):>14}[/bold cyan]{where}"
|
||||
f" {hit.duration:5.1f}s SNR {hit.snr_db:5.1f} dB "
|
||||
f"[green]{hit.classification or 'unclassified'}[/green] ({conf})")
|
||||
console.print(line, highlight=False)
|
||||
f"[green]{hit.classification or 'unclassified'}[/green] ({conf})",
|
||||
highlight=False)
|
||||
indent = 26 + _PLAIN_BAND + 2
|
||||
if hit.morse_text:
|
||||
console.print(f'{"":>26}[yellow]Morse @ {hit.morse_wpm:.0f} WPM: '
|
||||
console.print(f'{"":>{indent}}[yellow]Morse @ {hit.morse_wpm:.0f} WPM: '
|
||||
f'"{hit.morse_text.strip()}"[/yellow]', highlight=False)
|
||||
if hit.reasons:
|
||||
console.print(f'{"":>26}[grey54]{hit.reasons[0]}[/grey54]', highlight=False)
|
||||
console.print(f'{"":>{indent}}[grey54]{hit.reasons[0]}[/grey54]',
|
||||
highlight=False)
|
||||
|
||||
|
||||
def print_band_table(console: Console, presets, title: str = "band plan") -> None:
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
.\" Generated by packaging/make-man.py -- do not edit by hand.
|
||||
.TH BANDSAUNTER 1 "2026-08-22" "bandsaunter 2026-08-22_04" "User Commands"
|
||||
.TH BANDSAUNTER 1 "2026-08-28" "bandsaunter 2026-08-28_01" "User Commands"
|
||||
.SH NAME
|
||||
bandsaunter \- scan, record and identify radio signals with an RTL-SDR
|
||||
.SH SYNOPSIS
|
||||
|
|
@ -659,6 +659,26 @@ Accepts: at least 0.
|
|||
Do not bother transcribing captures shorter than this. Very short clips rarely contain a whole word and mostly produce noise or nothing, while still costing the processing.
|
||||
.RE
|
||||
.PP
|
||||
.SS Callsigns
|
||||
.TP
|
||||
.B --callsign-lookup / --no-callsign-lookup --offline-callsigns
|
||||
Look callsigns up \[em] ask the licence database who a callsign belongs to.
|
||||
.br
|
||||
Setting name \fBcallsign_lookup\fR, default \fByes\fR.
|
||||
.RS
|
||||
.PP
|
||||
When someone gives their callsign, look it up and say who they are. The data is the FCC's own published licence register, which carries the licensee's name, the town they are licensed in and the coordinates that put them on the map. Only the callsign is sent, and each one is asked about once and then remembered, so this costs almost nothing. Turn it off if you would rather the scan reached the network for nothing at all -- callsigns are still picked out of the transcripts, and the prefix still says which country and which US district they belong to.
|
||||
.RE
|
||||
.TP
|
||||
.B --kml
|
||||
Map file \[em] KML map of where the stations heard are licensed (blank = none).
|
||||
.br
|
||||
Setting name \fBkml_file\fR, default \fBcallsigns.kml\fR.
|
||||
.RS
|
||||
.PP
|
||||
The name of a map file, kept in the output directory, of everyone who identified themselves. Each station is one pin: the callsign, who holds the licence, where they are licensed, and every frequency and time you heard them. It is added to as the scan runs and again by later scans, so it builds up into a picture of what you can hear from where you are. KML is the format Google Earth uses; QGIS, Marble and OsmAnd read it too. Leave it blank if you do not want the map.
|
||||
.RE
|
||||
.PP
|
||||
.SS Run control
|
||||
.TP
|
||||
.B --cycles
|
||||
|
|
@ -761,6 +781,19 @@ sweeps every Morse segment of every amateur band, and
|
|||
.IR 2m\-complete ", " 70cm\-complete
|
||||
and their like sweep a whole amateur band end to end rather than one segment
|
||||
of it.
|
||||
.PP
|
||||
The same plan names what is heard. Beside every frequency on the display,
|
||||
and in the line\-per\-hit output, is the band it falls in: a signal at
|
||||
421 MHz is labelled
|
||||
.IR "70 cm Amateur" ,
|
||||
one at 462.5625 MHz is
|
||||
.IR "GMRS / FRS" ,
|
||||
and 162.55 MHz is
|
||||
.IR "NOAA Weather Radio" .
|
||||
Where several allocations overlap, the narrowest wins, because it says the
|
||||
most \[em] 146.52 MHz is named as the 2 m simplex calling channel rather than
|
||||
as the whole 2 m band. The name is written into each recording's sidecar as
|
||||
well, so it stays with the capture.
|
||||
.SH LOCK-OUTS
|
||||
Every receiving setup has a few frequencies not worth stopping on: a pager
|
||||
transmitter down the road, a nearby data link, or a spurious signal the
|
||||
|
|
@ -901,6 +934,39 @@ than a directory of placeholders.
|
|||
.BR saunterbrowse (1)
|
||||
reads these back, and lists any callsigns it finds in them with the licence
|
||||
they belong to.
|
||||
.SH THE MAP
|
||||
A callsign heard in a transcript is looked up in the FCC's published licence
|
||||
data, which gives the licensee, the town, and coordinates. Those go into a
|
||||
KML file in the output directory \[em]
|
||||
.I callsigns.kml
|
||||
unless
|
||||
.B \-\-kml
|
||||
names another \[em] which opens in Google Earth,
|
||||
.BR qgis (1),
|
||||
.BR marble (1)
|
||||
and OsmAnd.
|
||||
.PP
|
||||
One placemark per station, not one per transmission: the same repeater heard
|
||||
twenty times in an evening is one operator, and twenty pins on the same
|
||||
rooftop would say less than one. Each pin carries the callsign, the licensee,
|
||||
where they are licensed, and every frequency and time you heard them.
|
||||
.PP
|
||||
The file is added to, by this scan and by later ones, so it builds up into a
|
||||
picture of what the aerial can actually reach rather than a snapshot of one
|
||||
evening.
|
||||
.PP
|
||||
Only the callsign is sent, and each is asked about once and then remembered
|
||||
under
|
||||
.IR ~/.cache/bandsaunter/ ,
|
||||
so a net recorded night after night is looked up once.
|
||||
.B \-\-no\-callsign\-lookup
|
||||
stops it contacting anything at all; callsigns are still found, and the
|
||||
prefix still says which country and which US district they belong to. Setting
|
||||
.B \-\-kml
|
||||
to nothing turns the map off.
|
||||
.PP
|
||||
US amateur licence records are public by law and include the licensee's
|
||||
address. That is what is written.
|
||||
.SH HF RECEPTION
|
||||
These receivers cannot normally tune below about 24 MHz. Below that they can
|
||||
sample the antenna directly instead, which opens up shortwave: broadcast,
|
||||
|
|
@ -929,6 +995,12 @@ Where recordings, transcripts and logs are written, unless
|
|||
.B \-\-output
|
||||
says otherwise. Chosen on first run.
|
||||
.TP
|
||||
.I ~/bandsaunter/callsigns.kml
|
||||
The map of stations heard, added to as scans run.
|
||||
.TP
|
||||
.I ~/.cache/bandsaunter/callsigns.json
|
||||
Licence lookups already made, so they are not repeated.
|
||||
.TP
|
||||
.I /etc/modprobe.d/blacklist-rtlsdr.conf
|
||||
Written by the package to keep the DVB-T television driver from claiming the
|
||||
receiver.
|
||||
|
|
|
|||
|
|
@ -31,6 +31,8 @@ saunterbrowse \- read and listen to what a bandsaunter scan collected
|
|||
.IR CMD ]
|
||||
.RB [ \-\-list ]
|
||||
.RB [ \-\-callsigns ]
|
||||
.RB [ \-\-kml
|
||||
.RI [ FILE ]]
|
||||
.RB [ \-\-no\-lookup ]
|
||||
.SH DESCRIPTION
|
||||
A scan leaves a directory of recordings. Beside each one is a JSON file
|
||||
|
|
@ -124,6 +126,15 @@ argument. By default the first of these that is installed is used:
|
|||
Print every callsign heard in the directory, with the licence it belongs to
|
||||
and where and when it was heard, then exit.
|
||||
.TP
|
||||
.BI \-\-kml " [FILE]"
|
||||
Write a map of where the stations heard in this directory are licensed, and
|
||||
exit. Without a filename it writes
|
||||
.I callsigns.kml
|
||||
in the recordings directory \[em] the same file a scan writes, so a map built
|
||||
this way is continued by the next scan rather than duplicated. See
|
||||
.B THE MAP
|
||||
below.
|
||||
.TP
|
||||
.B \-\-no\-lookup
|
||||
Do not contact the licence database. Callsigns are still found and still
|
||||
described from their prefix; only the name and address are missing.
|
||||
|
|
@ -183,6 +194,37 @@ and time it was heard on, then exits.
|
|||
.PP
|
||||
US amateur licence records are public by law, and include the licensee's
|
||||
address. That is what is shown.
|
||||
.SH THE MAP
|
||||
A licence says where its holder is, so a list of callsigns is also a map. The
|
||||
scanner writes one as it runs and
|
||||
.B \-\-kml
|
||||
builds one from recordings already on disk; both write the same file, so
|
||||
either can carry on from the other.
|
||||
.PP
|
||||
Each station is one placemark, not one per transmission: hearing the same
|
||||
repeater twenty times in an evening is one station, and twenty pins stacked
|
||||
on the same rooftop would say less than one. The pin holds the callsign, the
|
||||
licensee, the town, the grid square, and every frequency and time it was
|
||||
heard on, so clicking it answers "when did I hear this, and where on the
|
||||
dial".
|
||||
.PP
|
||||
The file is added to rather than replaced. A scan on Tuesday continues the
|
||||
map Monday made, which over a few weeks turns into a picture of what your
|
||||
aerial can actually reach.
|
||||
.PP
|
||||
Where the licence carries no coordinates the grid square is used instead, and
|
||||
the placemark says so, because a grid square is kilometres across where a
|
||||
licensed address is a street. A callsign with no licence on file at all is
|
||||
still recorded, in a folder named
|
||||
.IR "no location on file" ,
|
||||
switched off by default: that a station was heard is worth keeping even when
|
||||
nothing says where it was.
|
||||
.PP
|
||||
KML is the format Google Earth uses.
|
||||
.BR qgis (1),
|
||||
.BR marble (1)
|
||||
and OsmAnd open it too, and it is XML, so a scan interrupted halfway through
|
||||
leaves a file that still opens.
|
||||
.SH TRANSCRIPTS
|
||||
A transcript appears only where a recogniser produced one, which means the
|
||||
capture was judged to be speech and
|
||||
|
|
@ -247,6 +289,14 @@ saunterbrowse \-\-callsigns
|
|||
.RE
|
||||
.PP
|
||||
Everyone who identified themselves, and where they were heard.
|
||||
.PP
|
||||
.RS
|
||||
.EX
|
||||
saunterbrowse \-\-kml ~/heard.kml
|
||||
.EE
|
||||
.RE
|
||||
.PP
|
||||
The same thing as a map, to open in Google Earth.
|
||||
.SH EXIT STATUS
|
||||
0 on a clean exit, 1 when the directory holds no recordings, 2 when it does
|
||||
not exist or there is no terminal to draw on.
|
||||
|
|
|
|||
|
|
@ -212,6 +212,19 @@ sweeps every Morse segment of every amateur band, and
|
|||
.IR 2m\-complete ", " 70cm\-complete
|
||||
and their like sweep a whole amateur band end to end rather than one segment
|
||||
of it.
|
||||
.PP
|
||||
The same plan names what is heard. Beside every frequency on the display,
|
||||
and in the line\-per\-hit output, is the band it falls in: a signal at
|
||||
421 MHz is labelled
|
||||
.IR "70 cm Amateur" ,
|
||||
one at 462.5625 MHz is
|
||||
.IR "GMRS / FRS" ,
|
||||
and 162.55 MHz is
|
||||
.IR "NOAA Weather Radio" .
|
||||
Where several allocations overlap, the narrowest wins, because it says the
|
||||
most \[em] 146.52 MHz is named as the 2 m simplex calling channel rather than
|
||||
as the whole 2 m band. The name is written into each recording's sidecar as
|
||||
well, so it stays with the capture.
|
||||
.SH LOCK-OUTS
|
||||
Every receiving setup has a few frequencies not worth stopping on: a pager
|
||||
transmitter down the road, a nearby data link, or a spurious signal the
|
||||
|
|
@ -352,6 +365,39 @@ than a directory of placeholders.
|
|||
.BR saunterbrowse (1)
|
||||
reads these back, and lists any callsigns it finds in them with the licence
|
||||
they belong to.
|
||||
.SH THE MAP
|
||||
A callsign heard in a transcript is looked up in the FCC's published licence
|
||||
data, which gives the licensee, the town, and coordinates. Those go into a
|
||||
KML file in the output directory \[em]
|
||||
.I callsigns.kml
|
||||
unless
|
||||
.B \-\-kml
|
||||
names another \[em] which opens in Google Earth,
|
||||
.BR qgis (1),
|
||||
.BR marble (1)
|
||||
and OsmAnd.
|
||||
.PP
|
||||
One placemark per station, not one per transmission: the same repeater heard
|
||||
twenty times in an evening is one operator, and twenty pins on the same
|
||||
rooftop would say less than one. Each pin carries the callsign, the licensee,
|
||||
where they are licensed, and every frequency and time you heard them.
|
||||
.PP
|
||||
The file is added to, by this scan and by later ones, so it builds up into a
|
||||
picture of what the aerial can actually reach rather than a snapshot of one
|
||||
evening.
|
||||
.PP
|
||||
Only the callsign is sent, and each is asked about once and then remembered
|
||||
under
|
||||
.IR ~/.cache/bandsaunter/ ,
|
||||
so a net recorded night after night is looked up once.
|
||||
.B \-\-no\-callsign\-lookup
|
||||
stops it contacting anything at all; callsigns are still found, and the
|
||||
prefix still says which country and which US district they belong to. Setting
|
||||
.B \-\-kml
|
||||
to nothing turns the map off.
|
||||
.PP
|
||||
US amateur licence records are public by law and include the licensee's
|
||||
address. That is what is written.
|
||||
.SH HF RECEPTION
|
||||
These receivers cannot normally tune below about 24 MHz. Below that they can
|
||||
sample the antenna directly instead, which opens up shortwave: broadcast,
|
||||
|
|
@ -380,6 +426,12 @@ Where recordings, transcripts and logs are written, unless
|
|||
.B \-\-output
|
||||
says otherwise. Chosen on first run.
|
||||
.TP
|
||||
.I ~/bandsaunter/callsigns.kml
|
||||
The map of stations heard, added to as scans run.
|
||||
.TP
|
||||
.I ~/.cache/bandsaunter/callsigns.json
|
||||
Licence lookups already made, so they are not repeated.
|
||||
.TP
|
||||
.I /etc/modprobe.d/blacklist-rtlsdr.conf
|
||||
Written by the package to keep the DVB-T television driver from claiming the
|
||||
receiver.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
.\" Generated by packaging/make-browse-man.py -- do not edit by hand.
|
||||
.TH SAUNTERBROWSE 1 "2026-08-22" "bandsaunter 2026-08-22_04" "User Commands"
|
||||
.TH SAUNTERBROWSE 1 "2026-08-28" "bandsaunter 2026-08-28_01" "User Commands"
|
||||
.SH NAME
|
||||
saunterbrowse \- read and listen to what a bandsaunter scan collected
|
||||
.SH SYNOPSIS
|
||||
|
|
@ -13,6 +13,8 @@ saunterbrowse \- read and listen to what a bandsaunter scan collected
|
|||
.IR CMD ]
|
||||
.RB [ \-\-list ]
|
||||
.RB [ \-\-callsigns ]
|
||||
.RB [ \-\-kml
|
||||
.RI [ FILE ]]
|
||||
.RB [ \-\-no\-lookup ]
|
||||
.SH DESCRIPTION
|
||||
A scan leaves a directory of recordings. Beside each one is a JSON file
|
||||
|
|
@ -106,6 +108,15 @@ pw-play, paplay, aplay, play, ffplay, mpv.
|
|||
Print every callsign heard in the directory, with the licence it belongs to
|
||||
and where and when it was heard, then exit.
|
||||
.TP
|
||||
.BI \-\-kml " [FILE]"
|
||||
Write a map of where the stations heard in this directory are licensed, and
|
||||
exit. Without a filename it writes
|
||||
.I callsigns.kml
|
||||
in the recordings directory \[em] the same file a scan writes, so a map built
|
||||
this way is continued by the next scan rather than duplicated. See
|
||||
.B THE MAP
|
||||
below.
|
||||
.TP
|
||||
.B \-\-no\-lookup
|
||||
Do not contact the licence database. Callsigns are still found and still
|
||||
described from their prefix; only the name and address are missing.
|
||||
|
|
@ -165,6 +176,37 @@ and time it was heard on, then exits.
|
|||
.PP
|
||||
US amateur licence records are public by law, and include the licensee's
|
||||
address. That is what is shown.
|
||||
.SH THE MAP
|
||||
A licence says where its holder is, so a list of callsigns is also a map. The
|
||||
scanner writes one as it runs and
|
||||
.B \-\-kml
|
||||
builds one from recordings already on disk; both write the same file, so
|
||||
either can carry on from the other.
|
||||
.PP
|
||||
Each station is one placemark, not one per transmission: hearing the same
|
||||
repeater twenty times in an evening is one station, and twenty pins stacked
|
||||
on the same rooftop would say less than one. The pin holds the callsign, the
|
||||
licensee, the town, the grid square, and every frequency and time it was
|
||||
heard on, so clicking it answers "when did I hear this, and where on the
|
||||
dial".
|
||||
.PP
|
||||
The file is added to rather than replaced. A scan on Tuesday continues the
|
||||
map Monday made, which over a few weeks turns into a picture of what your
|
||||
aerial can actually reach.
|
||||
.PP
|
||||
Where the licence carries no coordinates the grid square is used instead, and
|
||||
the placemark says so, because a grid square is kilometres across where a
|
||||
licensed address is a street. A callsign with no licence on file at all is
|
||||
still recorded, in a folder named
|
||||
.IR "no location on file" ,
|
||||
switched off by default: that a station was heard is worth keeping even when
|
||||
nothing says where it was.
|
||||
.PP
|
||||
KML is the format Google Earth uses.
|
||||
.BR qgis (1),
|
||||
.BR marble (1)
|
||||
and OsmAnd open it too, and it is XML, so a scan interrupted halfway through
|
||||
leaves a file that still opens.
|
||||
.SH TRANSCRIPTS
|
||||
A transcript appears only where a recogniser produced one, which means the
|
||||
capture was judged to be speech and
|
||||
|
|
@ -229,6 +271,14 @@ saunterbrowse \-\-callsigns
|
|||
.RE
|
||||
.PP
|
||||
Everyone who identified themselves, and where they were heard.
|
||||
.PP
|
||||
.RS
|
||||
.EX
|
||||
saunterbrowse \-\-kml ~/heard.kml
|
||||
.EE
|
||||
.RE
|
||||
.PP
|
||||
The same thing as a map, to open in Google Earth.
|
||||
.SH EXIT STATUS
|
||||
0 on a clean exit, 1 when the directory holds no recordings, 2 when it does
|
||||
not exist or there is no terminal to draw on.
|
||||
|
|
|
|||
14
tests/conftest.py
Normal file
14
tests/conftest.py
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
"""Fixtures every test gets.
|
||||
|
||||
The one that matters is the cache: a lookup writes to ``~/.cache`` by
|
||||
default, and a test run that touches the real one leaves entries behind and
|
||||
reads back entries an earlier version wrote. Redirecting it per test makes
|
||||
each run start from nothing.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolated_cache(tmp_path_factory, monkeypatch):
|
||||
monkeypatch.setenv("XDG_CACHE_HOME",
|
||||
str(tmp_path_factory.mktemp("cache")))
|
||||
233
tests/test_bands.py
Normal file
233
tests/test_bands.py
Normal file
|
|
@ -0,0 +1,233 @@
|
|||
"""Naming a frequency: which band it is in, and how that is shown."""
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from rich.console import Console
|
||||
|
||||
from bandsaunter import bandplan as bp
|
||||
from bandsaunter.bandplan import (LABEL_SKIP_TAGS, PRESETS, band_label,
|
||||
band_name, band_names, label_for,
|
||||
label_presets, shorten_band)
|
||||
from bandsaunter.config import ScanConfig
|
||||
from bandsaunter.ranges import parse_range_list
|
||||
from bandsaunter.recorder import HitRecord
|
||||
from bandsaunter.scanner import Scanner
|
||||
from bandsaunter.simulator import SimulatedDevice
|
||||
from bandsaunter.ui import ScanDisplay, print_hit
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# What a frequency is called
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("hz,expected", [
|
||||
# The one from the request: a signal at 421 MHz is in the 70 cm band.
|
||||
(421e6, "70 cm Amateur"),
|
||||
(146.52e6, "2 m FM Simplex"),
|
||||
(162.55e6, "NOAA Weather Radio"),
|
||||
(462.5625e6, "GMRS / FRS"),
|
||||
(27.185e6, "Citizens Band (CB) 11 m"),
|
||||
(88.5e6, "FM Broadcast Band"),
|
||||
(1.0e6, "AM Broadcast Band"),
|
||||
(1090e6, "ADS-B (1090 MHz)"),
|
||||
(856.5e6, "800 MHz Trunked Downlinks"),
|
||||
])
|
||||
def test_a_frequency_is_named_by_its_band(hz, expected):
|
||||
assert band_name(hz) == expected
|
||||
|
||||
|
||||
def test_a_frequency_no_preset_covers_has_no_band():
|
||||
# 2.1 GHz is past the tuner and past every preset in the plan.
|
||||
assert band_name(2.1e9) == ""
|
||||
assert band_label(2.1e9) == ""
|
||||
assert not label_for(2.1e9)
|
||||
|
||||
|
||||
def test_wide_sweeps_never_name_anything():
|
||||
""""Full UHF Sweep" is somewhere to point the radio, not an answer."""
|
||||
for hz in (100e6, 421e6, 462e6, 915e6, 1090e6):
|
||||
for preset in label_presets(hz):
|
||||
assert not (set(preset.tags) & LABEL_SKIP_TAGS), \
|
||||
f"{preset.key} named {hz/1e6:g} MHz"
|
||||
|
||||
|
||||
def test_ism_yields_to_the_allocation_it_shares():
|
||||
"""433 and 915 MHz are ISM, but they are 70 cm and 33 cm first."""
|
||||
assert band_name(433.92e6) == "70 cm Amateur"
|
||||
assert band_name(915e6) == "33 cm (902-928) Amateur"
|
||||
# ISM still wins where nothing else covers the frequency at all.
|
||||
assert "ISM" in band_name(315e6) or band_name(315e6)
|
||||
|
||||
|
||||
def test_shortwave_broadcast_yields_only_where_a_ham_band_overlaps():
|
||||
"""3.9-4.0 and 7.2-7.3 MHz are amateur in Region 2, which this plan is."""
|
||||
assert "Amateur" in band_name(3.965e6)
|
||||
assert "Amateur" in band_name(7.242e6)
|
||||
# 6 MHz really is 49 m shortwave; there is no amateur band near it.
|
||||
assert band_name(6.0e6) == "49 m Shortwave Broadcast"
|
||||
assert band_name(9.6e6) == "31 m Shortwave Broadcast"
|
||||
|
||||
|
||||
def test_the_broader_name_wins_a_tie():
|
||||
"""GMRS/FRS and the FRS simplex channels differ by one channel."""
|
||||
assert band_name(462.5625e6) == "GMRS / FRS"
|
||||
|
||||
|
||||
def test_a_much_narrower_segment_beats_the_band_containing_it():
|
||||
""""20 m CW" says more than "20 m", and is just as true."""
|
||||
assert band_name(14.05e6) == "20 m CW / Digital"
|
||||
assert "20 m Amateur" in band_names(14.05e6)
|
||||
|
||||
|
||||
def test_every_band_names_its_own_middle():
|
||||
"""A preset that cannot name a frequency inside itself is unreachable."""
|
||||
orphans = []
|
||||
for p in PRESETS:
|
||||
if p.is_group or (set(p.tags) & LABEL_SKIP_TAGS):
|
||||
continue
|
||||
middle = 0.5 * (p.start + p.stop)
|
||||
if p.name not in band_names(middle, limit=99):
|
||||
orphans.append(p.key)
|
||||
assert not orphans, f"presets that never name anything: {orphans}"
|
||||
|
||||
|
||||
def test_names_are_ordered_most_specific_first():
|
||||
names = band_names(146.52e6, limit=99)
|
||||
assert names[0] == "2 m FM Simplex"
|
||||
assert "2 m Amateur" in names
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fitting it in a column
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("name,width,expected", [
|
||||
("ADS-B (1090 MHz)", 0, "ADS-B"),
|
||||
("33 cm (902-928) Amateur", 0, "33 cm Amateur"),
|
||||
("1.25 m Weak Signal (CW/SSB)", 0, "1.25 m Weak Signal"),
|
||||
("800 MHz Public Safety / SMR", 24, "800 MHz Public Safety"),
|
||||
# Nothing left to drop: truncate, and say so.
|
||||
("800 MHz Public Safety / SMR", 12, "800 MHz Pub…"),
|
||||
("VHF Airband - Tower & Ground", 12, "VHF Airband"),
|
||||
])
|
||||
def test_shortening_removes_restatements_before_it_truncates(name, width,
|
||||
expected):
|
||||
assert shorten_band(name, width) == expected
|
||||
|
||||
|
||||
def test_shortening_never_exceeds_the_width_it_was_given():
|
||||
for p in PRESETS:
|
||||
for width in (8, 12, 16, 20, 24):
|
||||
assert len(shorten_band(p.name, width)) <= width, p.name
|
||||
|
||||
|
||||
def test_a_name_that_already_fits_is_left_alone():
|
||||
assert shorten_band("70 cm Amateur", 20) == "70 cm Amateur"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Where it is shown
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _display(width: int) -> ScanDisplay:
|
||||
console = Console(width=width, height=40, file=open(os.devnull, "w"),
|
||||
record=True)
|
||||
cfg = ScanConfig(ranges=parse_range_list("144M-148M"),
|
||||
output_dir=tempfile.mkdtemp())
|
||||
scanner = Scanner(cfg, device=SimulatedDevice().open())
|
||||
scanner.prepare()
|
||||
display = ScanDisplay(scanner, console=console)
|
||||
display.attach()
|
||||
return display
|
||||
|
||||
|
||||
def _hit(hz: float, **kw) -> HitRecord:
|
||||
hit = HitRecord(frequency=hz, started_at=time.time(), duration=5.0,
|
||||
snr_db=20.0, classification="FM voice", **kw)
|
||||
hit.kept = True
|
||||
return hit
|
||||
|
||||
|
||||
def _render(renderable, console) -> str:
|
||||
console.print(renderable)
|
||||
return console.export_text()
|
||||
|
||||
|
||||
def test_the_hit_list_has_a_band_beside_the_frequency():
|
||||
d = _display(120)
|
||||
d.hits.appendleft(_hit(421e6))
|
||||
text = _render(d._hits_table(), d.console)
|
||||
assert "band" in text
|
||||
assert "70 cm Amateur" in text
|
||||
# and beside the frequency, not somewhere else on the row
|
||||
row = [ln for ln in text.splitlines() if "421 MHz" in ln][0]
|
||||
assert row.index("421 MHz") < row.index("70 cm Amateur")
|
||||
|
||||
|
||||
def test_the_band_column_is_dropped_before_the_identification_is():
|
||||
narrow = _display(60)
|
||||
narrow.hits.appendleft(_hit(421e6))
|
||||
text = _render(narrow._hits_table(), narrow.console)
|
||||
assert "70 cm" not in text
|
||||
assert "FM voice" in text
|
||||
|
||||
|
||||
def test_a_recording_in_progress_says_which_band_it_is_in():
|
||||
d = _display(120)
|
||||
d._rec.active = True
|
||||
d._rec.frequency = 146.52e6
|
||||
d._rec.mode = "nfm"
|
||||
text = _render(d._record_panel(), d.console)
|
||||
assert "2 m FM Simplex" in text
|
||||
|
||||
|
||||
def test_the_sweep_line_says_which_band_is_being_swept():
|
||||
d = _display(120)
|
||||
step = d.scanner.plan[0]
|
||||
d.on_step(0, 6, step, np.full(1024, -70.0),
|
||||
np.linspace(step.low, step.high, 1024))
|
||||
text = _render(d._sweep_panel(), d.console)
|
||||
assert "2 m" in text
|
||||
|
||||
|
||||
def test_the_line_per_hit_output_carries_the_band_too():
|
||||
console = Console(width=140, file=open(os.devnull, "w"), record=True)
|
||||
print_hit(console, _hit(421e6))
|
||||
assert "70 cm Amateur" in console.export_text()
|
||||
|
||||
|
||||
def test_a_hit_keeps_the_band_it_was_filed_under():
|
||||
"""A band plan edited later must not rewrite what an old scan recorded."""
|
||||
hit = _hit(421e6)
|
||||
hit.band = "Somewhere Else"
|
||||
assert ScanDisplay._band_of(hit, 20) == "Somewhere Else"
|
||||
|
||||
|
||||
def test_a_hit_from_before_bands_were_recorded_still_gets_one():
|
||||
hit = _hit(421e6)
|
||||
hit.band = ""
|
||||
hit.band_labels = []
|
||||
assert ScanDisplay._band_of(hit, 20) == "70 cm Amateur"
|
||||
|
||||
|
||||
def test_the_scanner_records_the_band_with_every_hit():
|
||||
"""So the sidecar says what it was, not just where it was."""
|
||||
cfg = ScanConfig(ranges=parse_range_list("144M-148M"),
|
||||
output_dir=tempfile.mkdtemp())
|
||||
scanner = Scanner(cfg, device=SimulatedDevice().open())
|
||||
scanner.prepare()
|
||||
label = bp.label_for(146.52e6)
|
||||
assert label.name and label.names[0] == label.name
|
||||
|
||||
|
||||
def test_an_empty_hit_list_still_takes_one_row():
|
||||
"""A placeholder that wraps makes the panel taller than the layout
|
||||
budgeted for, and the whole display then scrolls itself off the screen."""
|
||||
for width in (60, 100, 140):
|
||||
d = _display(width)
|
||||
text = _render(d._hits_table(), d.console)
|
||||
body = [ln for ln in text.splitlines() if "no signals" in ln]
|
||||
assert len(body) == 1, f"the placeholder wrapped at width {width}"
|
||||
|
|
@ -840,3 +840,96 @@ def test_no_lookup_contacts_nothing(net, capsys, monkeypatch):
|
|||
assert made == [], "a request was made with --no-lookup"
|
||||
assert "KU0W" in out
|
||||
assert "United States" in out, "the prefix should still be described"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_the_details_line_names_the_band(library):
|
||||
b = browser(library)
|
||||
b.index = [i for i, c in enumerate(b.view)
|
||||
if abs(c.frequency - 146.52e6) < 1e5][0]
|
||||
assert "2 m Amateur" in frame(b)
|
||||
|
||||
|
||||
def test_a_capture_keeps_the_band_its_sidecar_recorded(library):
|
||||
"""The plan can be edited later; the recording was filed under this one."""
|
||||
b = browser(library)
|
||||
cap = [c for c in b.captures if abs(c.frequency - 146.52e6) < 1e5][0]
|
||||
assert cap.band == "2 m Amateur"
|
||||
|
||||
|
||||
def test_a_capture_with_no_band_in_its_sidecar_gets_one_from_the_dial(library):
|
||||
b = browser(library)
|
||||
cap = [c for c in b.captures if abs(c.frequency - 144.1e6) < 1e5][0]
|
||||
assert cap.band and "2 m" in cap.band
|
||||
|
||||
|
||||
def test_a_band_can_be_searched_for(library):
|
||||
""""70 cm" is how an operator thinks of a range, not 420-450 MHz."""
|
||||
b = browser(library)
|
||||
b.query = "2 m"
|
||||
b.apply()
|
||||
assert b.view, "searching by band found nothing"
|
||||
assert all("2 m" in c.band for c in b.view)
|
||||
|
||||
|
||||
def test_the_listing_carries_the_band_beside_the_frequency(library, capsys):
|
||||
from bandsaunter.browse import main
|
||||
assert main([str(library), "--list"]) == 0
|
||||
out = capsys.readouterr().out
|
||||
row = [ln for ln in out.splitlines() if "146.52 MHz" in ln][0]
|
||||
assert row.index("146.52 MHz") < row.index("2 m Amateur")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The map
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _stub_lookup(monkeypatch):
|
||||
def respond(self, call):
|
||||
return {"status": "VALID", "name": "ARRL HQ OPERATORS CLUB",
|
||||
"address": {"line2": "NEWINGTON, CT 06111"},
|
||||
"location": {"latitude": "41.714775",
|
||||
"longitude": "-72.727260",
|
||||
"gridsquare": "FN31pr"}}
|
||||
monkeypatch.setattr(CallsignBook, "_request", respond)
|
||||
|
||||
|
||||
def test_the_browser_writes_a_map_of_who_was_heard(library, monkeypatch,
|
||||
capsys):
|
||||
import xml.etree.ElementTree as ET
|
||||
from bandsaunter.browse import main
|
||||
_stub_lookup(monkeypatch)
|
||||
assert main([str(library), "--kml"]) == 0
|
||||
written = library / "callsigns.kml"
|
||||
assert written.exists()
|
||||
root = ET.parse(written).getroot()
|
||||
ns = "{http://www.opengis.net/kml/2.2}"
|
||||
names = [m.find(ns + "name").text for m in root.iter(ns + "Placemark")]
|
||||
assert any("W1AW" in n for n in names)
|
||||
# and where it is licensed, which is the point of a map
|
||||
point = next(root.iter(ns + "Placemark")).find(
|
||||
f"{ns}Point/{ns}coordinates")
|
||||
assert point is not None and point.text.startswith("-72.7")
|
||||
assert "with a position" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_the_map_can_be_written_somewhere_else(library, tmp_path, monkeypatch):
|
||||
from bandsaunter.browse import main
|
||||
_stub_lookup(monkeypatch)
|
||||
where = tmp_path / "elsewhere" / "heard.kml"
|
||||
assert main([str(library), "--kml", str(where)]) == 0
|
||||
assert where.exists()
|
||||
assert not (library / "callsigns.kml").exists()
|
||||
|
||||
|
||||
def test_no_callsigns_means_no_map(tmp_path, monkeypatch):
|
||||
from bandsaunter.browse import main
|
||||
_stub_lookup(monkeypatch)
|
||||
make_capture(tmp_path, 146.52, "2026-08-22_10_00_00", "nfm", 4.0,
|
||||
transcript="Nobody said a callsign in this one at all.",
|
||||
meta={"category": "voice"})
|
||||
assert main([str(tmp_path), "--kml"]) == 1
|
||||
assert not list(tmp_path.glob("*.kml"))
|
||||
|
|
|
|||
|
|
@ -356,3 +356,64 @@ def test_the_report_carries_the_extra_detail():
|
|||
oper_class="EXTRA", grid="DM42lj", status="found")
|
||||
line = report([entry], width=0)[1]
|
||||
assert "Extra" in line and "DM42lj" in line
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Coordinates, and the cache that holds them
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_lookup_records_where_the_licence_says_the_station_is(monkeypatch):
|
||||
"""The map is built from these, so they have to survive the parse."""
|
||||
body = {"status": "VALID", "name": "ARRL HQ OPERATORS CLUB",
|
||||
"address": {"line2": "NEWINGTON, CT 06111"},
|
||||
"location": {"latitude": "41.714775", "longitude": "-72.727260",
|
||||
"gridsquare": "FN31pr"}}
|
||||
entry = Callsign(call="W1AW")
|
||||
CallsignBook._apply(entry, body)
|
||||
assert entry.position == pytest.approx((41.714775, -72.727260))
|
||||
assert not entry.coarse_position
|
||||
|
||||
|
||||
def test_a_licence_with_only_a_grid_square_falls_back_to_it(monkeypatch):
|
||||
body = {"status": "VALID", "name": "SOMEONE",
|
||||
"address": {"line2": "SOMEWHERE, AZ 85742"},
|
||||
"location": {"gridsquare": "DM42lj"}}
|
||||
entry = Callsign(call="KU0W")
|
||||
CallsignBook._apply(entry, body)
|
||||
assert entry.position is not None
|
||||
# and it says the position is only as good as the square
|
||||
assert entry.coarse_position
|
||||
|
||||
|
||||
def test_a_licence_with_no_position_at_all_has_none():
|
||||
body = {"status": "VALID", "name": "SOMEONE", "address": {}, "location": {}}
|
||||
entry = Callsign(call="K7XYZ")
|
||||
CallsignBook._apply(entry, body)
|
||||
assert entry.position is None
|
||||
assert not entry.coarse_position
|
||||
|
||||
|
||||
def test_an_entry_cached_before_coordinates_is_looked_up_again(tmp_path):
|
||||
"""Otherwise a warm cache pins every station to its grid square for good."""
|
||||
from bandsaunter.callsign import CACHE_VERSION
|
||||
cache = tmp_path / "callsigns.json"
|
||||
cache.write_text(json.dumps({"W1AW": {
|
||||
"call": "W1AW", "name": "Arrl Hq Operators Club", "status": "found",
|
||||
"grid": "FN31pr", "fetched_at": time.time()}}))
|
||||
book = CallsignBook(online=False, cache=cache)
|
||||
assert "W1AW" not in book._entries
|
||||
|
||||
cache.write_text(json.dumps({"W1AW": {
|
||||
"call": "W1AW", "name": "Arrl Hq Operators Club", "status": "found",
|
||||
"grid": "FN31pr", "fetched_at": time.time(),
|
||||
"version": CACHE_VERSION}}))
|
||||
assert "W1AW" in CallsignBook(online=False, cache=cache)._entries
|
||||
|
||||
|
||||
def test_an_unlisted_callsign_is_cached_too(tmp_path, monkeypatch):
|
||||
"""Asking again every run about a callsign with no licence is waste."""
|
||||
from bandsaunter.callsign import CACHE_VERSION
|
||||
entry = Callsign(call="XX9ZZ")
|
||||
CallsignBook._apply(entry, {"status": "INVALID"})
|
||||
assert entry.status == "unlisted"
|
||||
assert entry.version == CACHE_VERSION
|
||||
|
|
|
|||
453
tests/test_kml.py
Normal file
453
tests/test_kml.py
Normal file
|
|
@ -0,0 +1,453 @@
|
|||
"""The map of who was heard.
|
||||
|
||||
KML is a text format, and every one of these reads back what was written
|
||||
with an XML parser rather than by matching strings, so a file that would not
|
||||
open in Google Earth fails here first.
|
||||
|
||||
Nothing touches the network: the licence data is built by hand.
|
||||
"""
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
import pytest
|
||||
|
||||
from bandsaunter.callsign import Callsign, CallsignBook, grid_to_latlon
|
||||
from bandsaunter.kml import Contact, Hearing, KmlLog
|
||||
|
||||
KML_NS = "{http://www.opengis.net/kml/2.2}"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def no_network(monkeypatch):
|
||||
def refuse(self, call):
|
||||
raise AssertionError(f"a test tried to look up {call} for real")
|
||||
|
||||
monkeypatch.setattr(CallsignBook, "_request", refuse)
|
||||
|
||||
|
||||
def _when(text: str) -> float:
|
||||
return time.mktime(time.strptime(text, "%Y-%m-%d %H:%M:%S"))
|
||||
|
||||
|
||||
def _entry(call="KU0W", **kw) -> Callsign:
|
||||
body = dict(name="Rod R Gowdy", location="Tucson, AZ",
|
||||
country="United States", grid="DM42lj", oper_class="EXTRA",
|
||||
latitude=32.3345, longitude=-111.0421, status="found")
|
||||
body.update(kw)
|
||||
return Callsign(call=call, **body)
|
||||
|
||||
|
||||
def _placemarks(path):
|
||||
root = ET.parse(path).getroot()
|
||||
return root.iter(KML_NS + "Placemark")
|
||||
|
||||
|
||||
def _fields(placemark) -> dict:
|
||||
out = {}
|
||||
for data in placemark.iter(KML_NS + "Data"):
|
||||
value = data.find(KML_NS + "value")
|
||||
out[data.get("name")] = value.text if value is not None else ""
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# One station, one pin
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_callsign_becomes_a_placemark_where_it_is_licensed(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(), 146.88e6, _when("2026-08-22 13:01:44"),
|
||||
recording="0146.880000MHz-nfm.wav")
|
||||
written = log.save()
|
||||
assert written is not None and written.exists()
|
||||
|
||||
marks = list(_placemarks(written))
|
||||
assert len(marks) == 1
|
||||
fields = _fields(marks[0])
|
||||
assert fields["callsign"] == "KU0W"
|
||||
assert fields["licensee"] == "Rod R Gowdy"
|
||||
assert fields["location"] == "Tucson, AZ"
|
||||
assert fields["grid"] == "DM42lj"
|
||||
point = marks[0].find(f"{KML_NS}Point/{KML_NS}coordinates")
|
||||
lon, lat, _ = point.text.split(",")
|
||||
# KML puts longitude first, which is the easy thing to get backwards.
|
||||
assert float(lon) == pytest.approx(-111.0421, abs=1e-4)
|
||||
assert float(lat) == pytest.approx(32.3345, abs=1e-4)
|
||||
|
||||
|
||||
def test_the_name_on_the_pin_is_the_callsign_and_the_licensee(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(), 146.88e6, _when("2026-08-22 13:01:44"))
|
||||
mark = next(_placemarks(log.save()))
|
||||
assert mark.find(KML_NS + "name").text == "KU0W — Rod R Gowdy"
|
||||
|
||||
|
||||
def test_the_balloon_names_the_frequency_and_the_time(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(), 146.88e6, _when("2026-08-22 13:01:44"))
|
||||
body = next(_placemarks(log.save())).find(KML_NS + "description").text
|
||||
assert "146.88 MHz" in body
|
||||
assert "2026-08-22 13:01:44" in body
|
||||
assert "2 m Amateur" in body or "2 m" in body
|
||||
assert "Rod R Gowdy" in body
|
||||
|
||||
|
||||
def test_hearing_a_station_again_adds_a_line_not_a_second_pin(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
for offset in (0, 300, 900):
|
||||
log.add(_entry(), 146.88e6, _when("2026-08-22 13:01:44") + offset)
|
||||
written = log.save()
|
||||
assert len(list(_placemarks(written))) == 1
|
||||
heard = _fields(next(_placemarks(written)))["heard"].splitlines()
|
||||
assert len(heard) == 3
|
||||
|
||||
|
||||
def test_the_very_same_transmission_is_not_recorded_twice(tmp_path):
|
||||
"""A re-run over the same recordings must not inflate the count."""
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
at = _when("2026-08-22 13:01:44")
|
||||
assert log.add(_entry(), 146.88e6, at) is True
|
||||
assert log.add(_entry(), 146.88e6, at) is False
|
||||
heard = _fields(next(_placemarks(log.save())))["heard"].splitlines()
|
||||
assert len(heard) == 1
|
||||
|
||||
|
||||
def test_a_station_heard_on_two_bands_keeps_both(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
at = _when("2026-08-22 13:01:44")
|
||||
log.add(_entry(), 146.88e6, at)
|
||||
log.add(_entry(), 7.242e6, at + 60)
|
||||
heard = _fields(next(_placemarks(log.save())))["heard"]
|
||||
assert "146.88 MHz" in heard and "7.242 MHz" in heard
|
||||
|
||||
|
||||
def test_the_time_span_covers_first_to_last(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
at = _when("2026-08-22 13:01:44")
|
||||
log.add(_entry(), 146.88e6, at)
|
||||
log.add(_entry(), 146.88e6, at + 3600)
|
||||
mark = next(_placemarks(log.save()))
|
||||
span = mark.find(KML_NS + "TimeSpan")
|
||||
assert span is not None
|
||||
begin = span.find(KML_NS + "begin").text
|
||||
end = span.find(KML_NS + "end").text
|
||||
assert begin < end and begin.endswith("Z")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Continuing an existing map
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_later_run_adds_to_the_map_rather_than_replacing_it(tmp_path):
|
||||
path = tmp_path / "map.kml"
|
||||
first = KmlLog(path)
|
||||
first.add(_entry(), 146.88e6, _when("2026-08-22 13:01:44"))
|
||||
first.save()
|
||||
|
||||
second = KmlLog(path)
|
||||
assert "KU0W" in second.contacts, "the existing map was not read back"
|
||||
second.add(_entry("W1AW", name="Arrl Hq Operators Club",
|
||||
location="Newington, CT", grid="FN31pr",
|
||||
latitude=41.7147, longitude=-72.7270),
|
||||
3.965e6, _when("2026-08-23 20:00:00"))
|
||||
written = second.save()
|
||||
calls = {_fields(m)["callsign"] for m in _placemarks(written)}
|
||||
assert calls == {"KU0W", "W1AW"}
|
||||
|
||||
|
||||
def test_what_was_heard_before_survives_the_round_trip(tmp_path):
|
||||
path = tmp_path / "map.kml"
|
||||
first = KmlLog(path)
|
||||
at = _when("2026-08-22 13:01:44")
|
||||
first.add(_entry(), 146.88e6, at, recording="one.wav")
|
||||
first.save()
|
||||
|
||||
second = KmlLog(path)
|
||||
heard = second.contacts["KU0W"].hearings
|
||||
assert len(heard) == 1
|
||||
assert heard[0].frequency == pytest.approx(146.88e6)
|
||||
assert heard[0].at == pytest.approx(at)
|
||||
assert heard[0].recording == "one.wav"
|
||||
# and the same transmission read back is still not added twice
|
||||
assert second.add(_entry(), 146.88e6, at) is False
|
||||
|
||||
|
||||
def test_a_position_learned_later_updates_the_pin(tmp_path):
|
||||
path = tmp_path / "map.kml"
|
||||
first = KmlLog(path)
|
||||
first.add(Callsign(call="KU0W", status="offline"), 146.88e6, 1.0)
|
||||
first.save()
|
||||
|
||||
second = KmlLog(path)
|
||||
second.add(_entry(), 146.88e6, 2.0)
|
||||
mark = next(_placemarks(second.save()))
|
||||
assert mark.find(f"{KML_NS}Point/{KML_NS}coordinates") is not None
|
||||
|
||||
|
||||
def test_a_failed_lookup_does_not_blank_out_what_is_known(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(), 146.88e6, 1.0)
|
||||
log.add(Callsign(call="KU0W", status="offline"), 146.88e6, 2.0)
|
||||
fields = _fields(next(_placemarks(log.save())))
|
||||
assert fields["licensee"] == "Rod R Gowdy"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stations with nowhere to put them
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_callsign_with_no_licence_is_kept_but_not_placed(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(Callsign(call="2E0XYZ", country="United Kingdom",
|
||||
status="unlisted"), 145.5e6, 1.0)
|
||||
written = log.save()
|
||||
mark = next(_placemarks(written))
|
||||
assert _fields(mark)["callsign"] == "2E0XYZ"
|
||||
assert mark.find(f"{KML_NS}Point/{KML_NS}coordinates") is None
|
||||
root = ET.parse(written).getroot()
|
||||
folder = root.find(f".//{KML_NS}Folder/{KML_NS}name")
|
||||
assert folder is not None and "no location" in folder.text
|
||||
|
||||
|
||||
def test_a_grid_square_places_a_station_the_licence_did_not(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(Callsign(call="W1AW", grid="FN31pr", status="found"),
|
||||
3.965e6, 1.0)
|
||||
mark = next(_placemarks(log.save()))
|
||||
point = mark.find(f"{KML_NS}Point/{KML_NS}coordinates")
|
||||
lon, lat, _ = point.text.split(",")
|
||||
assert float(lat) == pytest.approx(41.73, abs=0.05)
|
||||
assert float(lon) == pytest.approx(-72.71, abs=0.05)
|
||||
# and it says so, because a grid square is kilometres wide
|
||||
assert "grid square" in mark.find(KML_NS + "description").text
|
||||
|
||||
|
||||
def test_a_record_with_no_coordinates_is_not_placed_off_africa(tmp_path):
|
||||
"""Zeroed coordinates are missing data, not a position in the Atlantic."""
|
||||
entry = Callsign(call="K7XYZ", status="found",
|
||||
latitude=0.0, longitude=0.0)
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(entry, 146.0e6, 1.0)
|
||||
mark = next(_placemarks(log.save()))
|
||||
assert mark.find(f"{KML_NS}Point/{KML_NS}coordinates") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("grid,lat,lon", [
|
||||
("DM42lj", 32.40, -111.04),
|
||||
("FN31pr", 41.73, -72.71),
|
||||
("EM79", 39.5, -85.0),
|
||||
])
|
||||
def test_grid_squares_convert_to_their_centre(grid, lat, lon):
|
||||
got = grid_to_latlon(grid)
|
||||
assert got == pytest.approx((lat, lon), abs=0.06)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", ["", "nonsense", "ZZ99", "DM4", "DM42lj9"])
|
||||
def test_what_is_not_a_grid_square_converts_to_nothing(bad):
|
||||
assert grid_to_latlon(bad) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Not breaking the file
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_name_with_xml_in_it_does_not_break_the_document(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(name="Smith & Sons <Radio> Club"), 146.88e6, 1.0)
|
||||
fields = _fields(next(_placemarks(log.save())))
|
||||
assert fields["licensee"] == "Smith & Sons <Radio> Club"
|
||||
|
||||
|
||||
def test_an_unreadable_file_is_never_overwritten(tmp_path):
|
||||
path = tmp_path / "map.kml"
|
||||
path.write_text("this is not KML, and something else wrote it")
|
||||
log = KmlLog(path)
|
||||
log.add(_entry(), 146.88e6, 1.0)
|
||||
assert log.save() is None
|
||||
assert path.read_text().startswith("this is not KML")
|
||||
|
||||
|
||||
def test_nothing_heard_writes_no_file(tmp_path):
|
||||
path = tmp_path / "map.kml"
|
||||
assert KmlLog(path).save() is None
|
||||
assert not path.exists()
|
||||
|
||||
|
||||
def test_the_write_leaves_no_temporary_behind(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
log.add(_entry(), 146.88e6, 1.0)
|
||||
log.save()
|
||||
assert [p.name for p in tmp_path.iterdir()] == ["map.kml"]
|
||||
|
||||
|
||||
def test_a_callsign_with_no_call_is_refused(tmp_path):
|
||||
log = KmlLog(tmp_path / "map.kml")
|
||||
assert log.add(Callsign(call=""), 146.0e6, 1.0) is False
|
||||
assert len(log) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The pieces
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_a_hearing_reads_back_what_it_wrote():
|
||||
original = Hearing(146.88e6, _when("2026-08-22 13:01:44"),
|
||||
"2 m Amateur", "capture.wav")
|
||||
again = Hearing.parse(original.describe())
|
||||
assert again.frequency == pytest.approx(original.frequency)
|
||||
assert again.at == pytest.approx(original.at)
|
||||
assert again.band == original.band
|
||||
assert again.recording == original.recording
|
||||
|
||||
|
||||
def test_a_mangled_line_loses_only_itself():
|
||||
"""Someone editing the map by hand must not destroy the rest of it."""
|
||||
assert Hearing.parse("") is None
|
||||
assert Hearing.parse("who knows what this is") is not None
|
||||
|
||||
|
||||
def test_a_contact_orders_its_hearings_by_time():
|
||||
c = Contact(call="KU0W")
|
||||
c.add(Hearing(146.88e6, 300.0))
|
||||
c.add(Hearing(146.88e6, 100.0))
|
||||
assert [h.at for h in c.hearings] == [100.0, 300.0]
|
||||
assert c.first_heard == 100.0 and c.last_heard == 300.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# A scan that hears somebody
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
CALLOOK = {
|
||||
"status": "VALID",
|
||||
"type": "PERSON",
|
||||
"current": {"callsign": "W1AW", "operClass": "CLUB"},
|
||||
"name": "ARRL HQ OPERATORS CLUB",
|
||||
"address": {"line1": "225 MAIN ST", "line2": "NEWINGTON, CT 06111"},
|
||||
"location": {"latitude": "41.714775", "longitude": "-72.727260",
|
||||
"gridsquare": "FN31pr"},
|
||||
"otherInfo": {"expiryDate": "02/29/2032"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_lookup(monkeypatch):
|
||||
"""The licence database, answering from a fixture instead of the network."""
|
||||
asked = []
|
||||
|
||||
def respond(self, call):
|
||||
asked.append(call)
|
||||
return dict(CALLOOK, current={"callsign": call, "operClass": "CLUB"})
|
||||
|
||||
monkeypatch.setattr(CallsignBook, "_request", respond)
|
||||
return asked
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def heard_speech(monkeypatch):
|
||||
"""A recogniser that hears one station identify itself."""
|
||||
from bandsaunter import transcribe as tr
|
||||
|
||||
monkeypatch.setitem(
|
||||
tr._DISPATCH, "fake",
|
||||
lambda audio, rate, model, lang: tr.Transcript(
|
||||
text="Net control, this is W1AW, standing by.", engine="fake"))
|
||||
monkeypatch.setattr(tr, "ENGINES", ("fake",) + tr.ENGINES)
|
||||
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
||||
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
||||
|
||||
|
||||
def _scan(tmp_path, **over):
|
||||
from bandsaunter.config import ScanConfig
|
||||
from bandsaunter.ranges import parse_range_list
|
||||
from bandsaunter.scanner import Scanner, ScannerCallbacks
|
||||
from bandsaunter.simulator import SimulatedDevice, VirtualTransmitter as V
|
||||
|
||||
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
|
||||
output_dir=str(tmp_path), record_seconds=4.0,
|
||||
hang_seconds=1.0, threshold_db=12, dwell_seconds=0.05,
|
||||
max_cycles=1, revisit_seconds=0.2, transcribe=True,
|
||||
transcribe_engine="fake")
|
||||
on_status = over.pop("_on_status", None)
|
||||
for key, value in over.items():
|
||||
setattr(cfg, key, value)
|
||||
scanner = Scanner(cfg, device=SimulatedDevice(
|
||||
transmitters=[V(146_520_000, "nfm", 0.4, 12_500, "v")]).open(),
|
||||
callbacks=ScannerCallbacks(on_status=on_status) if on_status else None)
|
||||
scanner.prepare()
|
||||
scanner.run()
|
||||
return scanner
|
||||
|
||||
|
||||
def test_a_scan_maps_the_stations_it_hears(tmp_path, heard_speech,
|
||||
fake_lookup):
|
||||
scanner = _scan(tmp_path)
|
||||
written = tmp_path / "callsigns.kml"
|
||||
assert written.exists(), "the scan heard a callsign but wrote no map"
|
||||
assert fake_lookup == ["W1AW"]
|
||||
|
||||
mark = next(_placemarks(written))
|
||||
fields = _fields(mark)
|
||||
assert fields["callsign"] == "W1AW"
|
||||
assert fields["licensee"] == "ARRL HQ Operators Club"
|
||||
assert fields["location"] == "Newington, CT"
|
||||
# The frequency is the one the detector settled on, which is
|
||||
# near the transmitter rather than exactly on it.
|
||||
assert "146.5" in fields["heard"]
|
||||
assert "2 m FM Simplex" in fields["heard"]
|
||||
assert ".wav" in fields["heard"]
|
||||
assert mark.find(f"{KML_NS}Point/{KML_NS}coordinates") is not None
|
||||
assert scanner.heard.get("W1AW")
|
||||
|
||||
|
||||
def test_a_scan_says_on_the_display_who_it_heard(tmp_path, heard_speech,
|
||||
fake_lookup):
|
||||
said = []
|
||||
_scan(tmp_path, _on_status=said.append)
|
||||
spoken = [m for m in said if "W1AW" in m]
|
||||
assert spoken, f"the scan never announced the callsign: {said}"
|
||||
assert "ARRL HQ Operators Club" in spoken[0]
|
||||
|
||||
|
||||
def test_the_same_station_is_only_announced_once(tmp_path, heard_speech,
|
||||
fake_lookup):
|
||||
"""A repeater net would otherwise fill the status line with one name."""
|
||||
said = []
|
||||
_scan(tmp_path, _on_status=said.append, max_cycles=3,
|
||||
revisit_seconds=0.0)
|
||||
assert len([m for m in said if m.startswith("heard W1AW")]) == 1
|
||||
|
||||
|
||||
def test_the_map_is_off_when_no_file_is_named(tmp_path, heard_speech,
|
||||
fake_lookup):
|
||||
_scan(tmp_path, kml_file="")
|
||||
assert not list(tmp_path.glob("*.kml"))
|
||||
|
||||
|
||||
def test_lookups_can_be_turned_off_entirely(tmp_path, heard_speech):
|
||||
"""Nothing may reach the network with the lookup switched off."""
|
||||
def refuse(self, call):
|
||||
raise AssertionError("looked a callsign up with lookups off")
|
||||
|
||||
import bandsaunter.callsign as cs
|
||||
original, cs.CallsignBook._request = cs.CallsignBook._request, refuse
|
||||
try:
|
||||
_scan(tmp_path, callsign_lookup=False)
|
||||
finally:
|
||||
cs.CallsignBook._request = original
|
||||
# The station is still on the map, named by its prefix rather than a
|
||||
# licence: that it was heard is worth recording either way.
|
||||
written = tmp_path / "callsigns.kml"
|
||||
assert written.exists()
|
||||
assert _fields(next(_placemarks(written)))["callsign"] == "W1AW"
|
||||
|
||||
|
||||
def test_a_second_scan_adds_to_the_same_map(tmp_path, heard_speech,
|
||||
fake_lookup):
|
||||
_scan(tmp_path)
|
||||
first = len(_fields(next(_placemarks(tmp_path / "callsigns.kml")
|
||||
))["heard"].splitlines())
|
||||
_scan(tmp_path)
|
||||
marks = list(_placemarks(tmp_path / "callsigns.kml"))
|
||||
assert len(marks) == 1, "a second scan started a second pin"
|
||||
assert len(_fields(marks[0])["heard"].splitlines()) > first
|
||||
Loading…
Add table
Add a link
Reference in a new issue