dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
bandsaunter/tests/test_weather.py
The Dust Council 706c632f47 Listen the way the tools that work on this band listen 
Still nothing, with rtl-433 receiving the same sensors on the same aerial from
a different receiver.  That settles where the fault is not: not the aerial,
not the sensors, not the band.  So the sensible thing is to stop differing
from the configuration known to work on that aerial, and this differed from it
in three ways, every one of them mine.

It tuned a quarter of a megahertz to one side of 433.92 and shifted the signal
back in software, to keep the receiver's own spike off a signal that works by
being switched off.  That is a real effect and avoiding it this way is a bad
trade: at the sample rate this ran at, the shift is followed by a filter, and
a filter narrow enough to reject the spike is narrow enough to lose a
transmitter that has drifted -- or to lose the signal outright if a dongle
presents its samples the other way round, which is not a thing to depend on.
The spike is a steady addition to the envelope and a burst rises clear of it.
Tuning straight at the sensors now, which is what the established tools do.

It sampled at a megasample a second where a quarter of one is plenty: the
shortest pulse these send is two hundred microseconds, which is fifty samples
at the lowest rate a dongle will do.  The extra rate bought nothing but the
room for that filter to exist in.  At 250 kS/s nothing after the mixer is
narrower than the band, so the offset now does nothing at all whatever it is
set to, and says so.

And it turned on the RTL2832's digital gain control along with the tuner's.
The two pump: the gain winds up through the silence between one burst and the
next, lifting the noise towards the signal and squeezing the very difference
the burst detector works on.  It matters here in a way it does not for
aircraft, where a frame is found by correlating a preamble over microseconds
rather than by comparing a burst with the quiet around it.

Three more faults found while going over the rest of it, all the same mistake
in different clothes -- treating the middle of a distribution as though it
were the quiet part of one.

The check that skips an empty block measured the peak against the median.  A
recording that is mostly burst measures its own burst against its own burst,
finds no difference and is discarded as silence, which is what happened to
every short capture.  The gate's scatter had the same trouble one level down
and could come out above the peak, which is the one setting that cannot be
right, so it is now capped below it.

And the rule deciding where one message ends keyed on the middle gap in a
burst.  Where a one is drawn as a gap three times a zero and most of the bits
are zeroes, the middle gap is the short one, twice it still falls inside the
message, and every one-bit ended a burst -- the message coming apart into
pieces of three pulses.  It keys on the widest gap now, which is a fact about
the message rather than about the data it happened to carry.

The pieces of a message are also put back together after being sliced rather
than before.  Grouping has to be tight, because the group is what fixes the
threshold and a group holding two sensors of unequal strength fixes it on the
louder; but the gaps inside one message run from two hundred microseconds on
the newer sensors to four thousand on the oldest, and a grouping tight enough
for the first tears the second into a bit at a time.  So each piece is
measured at its own amplitude and joined to its neighbours afterwards, and
anything that comes out longer than the longest message there is gets cut at
its largest gaps.

Measured rather than argued: across four hundred and eighty combinations of
pulse and gap timing, 464 now read where 417 did; across twenty-one gap-keyed
combinations, 18 where 12 did; and eighty seconds of receiver noise still
yields nothing at all.

--from-iq FILE reads a saved capture instead of the receiver, so a recording
made where the aerial is can be worked on anywhere, as many times as it takes.
Everything downstream of the dongle is the real thing, which is what tells a
receiver problem and a decoder problem apart.  --save-iq writes the settings
beside the samples, a file of raw samples with no record of its rate being
unreadable by anything.  The invented garden now waits like a dongle instead
of running as fast as the machine allows, which it should have done from the
start: --seconds meant nothing against it and a capture came out fifty times
too large.

Full suite 2331 passed; this work checked against sixteen deliberately broken
builds, two of which it survived until the tests were made to catch them, and
one change was removed for being unable to earn a test at all.  Built as
2026-09-07_03.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
2026-09-07 21:01:00 -07:00

1543 lines
60 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""The weather section: the options, the names, the log and what it reads as.
The radio end of this is in test_acurite.py. What is here is everything
between a decoded message and a person: the names given to sensors, the log
they go into, the spreadsheet that comes out, the table that is printed, and
the two front ends that drive all of it.
The names get the most attention, because they are the only thing in this
section that somebody typed and so the only thing that cannot be recovered by
listening again.
"""
import json
import time
import numpy as np
import pytest
from rich.console import Console
from bandsaunter import acurite as a
from bandsaunter import weather as wx
from bandsaunter import weatherlog as wl
from bandsaunter.sensors import Sensor, SensorBook
from bandsaunter.ui import WeatherDisplay
RATE = 400_000.0
OFFSET = 100_000.0
def reading(family="tower", sensor=0x1A2B, celsius=17.4, humidity=61,
channel="A", at=1_700_000_000.0, battery_low=False):
"""One decoded message, made the way the receiver would have made it."""
bits = {"tower": lambda: a.tower_frame(sensor, celsius, humidity, channel,
battery_low),
"5n1": lambda: a.five_in_one_weather(sensor, 11.0, celsius,
humidity, channel),
"wind": lambda: a.five_in_one_wind_rain(sensor, 11.0, 90.0, 1284,
channel)}[family]()
got = a.decode(bits, confirm=False)
got.at = at
return got
@pytest.fixture
def book(tmp_path):
return SensorBook(path=tmp_path / "sensors.yaml")
# ---------------------------------------------------------------------------
# The options
# ---------------------------------------------------------------------------
def test_every_option_names_a_field_that_exists():
fields = set(wx.WeatherOptions().__dict__)
for option in wx.OPTIONS:
assert option.key in fields, f"{option.key} is not an option"
def test_every_field_is_an_option_somebody_can_reach():
"""Anything settable has to be settable from the menu and the flags."""
keys = {o.key for o in wx.OPTIONS}
assert set(wx.WeatherOptions().__dict__) == keys
def test_every_option_is_in_a_group_the_menu_shows():
for option in wx.OPTIONS:
assert option.group in wx.OPTION_GROUPS
for group in wx.OPTION_GROUPS:
assert wx.in_group(group), f"{group} is empty"
def test_every_option_says_what_it_does_and_how_to_set_it():
for option in wx.OPTIONS:
assert option.help and option.help[0].islower()
assert option.detail, f"{option.key} has no detail"
assert option.flags, f"{option.key} has no command line flag"
if option.kind == "bool":
assert option.off_flags, f"{option.key} cannot be turned off"
def test_the_options_survive_being_saved_and_read_back(tmp_path):
options = wx.WeatherOptions(units="imperial", only_named=True, hold=90.0,
offset=0.0, gain="40")
wx.save_options(options, tmp_path)
assert wx.load_options(tmp_path) == options
def test_a_settings_file_with_a_mistake_in_it_costs_the_defaults(tmp_path):
(tmp_path / "weather.yaml").write_text("units: [this is not a unit\n")
assert wx.load_options(tmp_path) == wx.WeatherOptions()
def test_a_setting_this_version_does_not_know_is_ignored(tmp_path):
(tmp_path / "weather.yaml").write_text("units: imperial\nfuture: 3\n")
assert wx.load_options(tmp_path).units == "imperial"
@pytest.mark.parametrize("options,broken", [
(wx.WeatherOptions(rate=100_000.0), "sample rate"),
(wx.WeatherOptions(rate=400_000.0, offset=300_000.0), "offset"),
(wx.WeatherOptions(seconds=-1.0), "negative"),
(wx.WeatherOptions(hold=0.0), "display"),
])
def test_a_setting_that_cannot_work_is_refused_before_the_receiver_opens(
options, broken):
assert any(broken in err for err in options.validate())
def test_the_defaults_are_all_workable():
assert wx.WeatherOptions().validate() == []
def test_imperial_is_a_word_rather_than_a_flag():
assert wx.WeatherOptions(units="imperial").imperial is True
assert wx.WeatherOptions(units="metric").imperial is False
def test_a_sweep_of_433_is_told_it_wants_this_mode_instead():
class Range:
def __init__(self, start, stop):
self.start, self.stop = start, stop
assert "433" in wx.scanning_sensor_band([Range(430e6, 440e6)])
assert wx.scanning_sensor_band([Range(144e6, 148e6)]) == ""
# ---------------------------------------------------------------------------
# Names: the only thing here a person typed
# ---------------------------------------------------------------------------
def test_a_name_is_on_the_disk_the_moment_it_is_given(book):
"""Not when the program exits. It ends on control-C, if at all."""
book.tag("tower/1A2B", "back fence")
assert SensorBook(path=book.path).name_for("tower/1A2B") == "back fence"
def test_a_name_survives_the_counts_being_written_over_it(book):
book.tag("tower/1A2B", "back fence")
book.heard(reading())
book.flush()
again = SensorBook(path=book.path)
assert again.name_for("tower/1A2B") == "back fence"
assert again.get("tower/1A2B").messages == 1
def test_a_half_written_file_cannot_happen(book, monkeypatch):
"""The write goes to a neighbour and is renamed over the old one."""
book.tag("tower/1A2B", "back fence")
renames = []
import bandsaunter.sensors as sensors
monkeypatch.setattr(sensors.os, "replace",
lambda src, dst: renames.append((src, dst)))
book.tag("tower/0C41", "greenhouse")
assert renames and str(renames[0][1]) == str(book.path)
# The old file is still the old file until the rename happens.
assert SensorBook(path=book.path).name_for("tower/0C41") == ""
def test_a_names_file_with_a_mistake_in_it_is_left_alone(tmp_path):
where = tmp_path / "sensors.yaml"
where.write_text("sensors: [ this is not\n")
assert len(SensorBook(path=where)) == 0
assert where.read_text() == "sensors: [ this is not\n"
def test_a_sensor_keeps_its_name_when_the_channel_switch_is_moved(book):
book.tag("tower/1A2B", "back fence")
book.heard(reading(channel="C"))
assert book.name_for("tower/1A2B") == "back fence"
assert book.get("tower/1A2B").channel == "C"
def test_the_first_and_last_time_a_sensor_was_heard_are_kept(book):
book.heard(reading(at=1_000.0))
book.heard(reading(at=2_000.0))
sensor = book.get("tower/1A2B")
assert (sensor.first_heard, sensor.last_heard) == (1_000.0, 2_000.0)
assert sensor.messages == 2
def test_a_nameless_sensor_is_shown_by_its_identity_and_not_as_blank(book):
book.heard(reading())
assert book.label_for("tower/1A2B") == "1A2B"
book.tag("tower/1A2B", "back fence")
assert book.label_for("tower/1A2B") == "back fence"
def test_named_sensors_come_first_and_then_the_most_recently_heard(book):
book.heard(reading(sensor=0x0001, at=100.0))
book.heard(reading(sensor=0x0002, at=300.0))
book.heard(reading(sensor=0x0003, at=200.0))
book.tag("tower/0003", "shed")
assert [s.sensor for s in book.ordered()] == ["0003", "0002", "0001"]
def test_a_sensor_is_found_by_its_identity_by_its_name_or_by_its_key(book):
book.heard(reading())
book.tag("tower/1A2B", "back fence")
for text in ("1A2B", "1a2b", "back fence", "tower/1A2B"):
assert [s.key for s in book.find(text)] == ["tower/1A2B"]
def test_an_exact_identity_wins_over_everything_it_appears_inside(book):
book.heard(reading(sensor=0x00AB))
book.heard(reading(sensor=0xAB12))
assert [s.sensor for s in book.find("00AB")] == ["00AB"]
def test_forgetting_a_sensor_removes_it_from_the_file(book):
book.tag("tower/1A2B", "back fence")
assert book.forget("tower/1A2B") is True
assert book.forget("tower/1A2B") is False
assert len(SensorBook(path=book.path)) == 0
def test_a_sensor_never_heard_can_still_be_given_a_name(book):
"""The box on the shed is on the shed whether or not it was received."""
from bandsaunter.cli import _name_sensors
assert _name_sensors(book, ["1A2B=back fence"], quiet=True) == 1
assert book.name_for("?/1A2B") == "back fence"
def test_naming_something_that_has_been_heard_uses_the_key_it_was_heard_under(
book):
from bandsaunter.cli import _name_sensors
book.heard(reading())
_name_sensors(book, ["1a2b=back fence"], quiet=True)
assert book.name_for("tower/1A2B") == "back fence"
assert "?/1A2B" not in book
# ---------------------------------------------------------------------------
# What is out there, and what it last said
# ---------------------------------------------------------------------------
def test_a_station_keeps_the_latest_of_every_quantity_not_the_latest_message():
"""A 5-in-1 alternates two messages, and the line must show the sensor."""
garden = wx.Garden()
garden.add(reading("wind", sensor=0x777, at=10.0))
garden.add(reading("5n1", sensor=0x777, celsius=16.8, at=20.0))
station = garden.stations["5n1/0777"]
assert station.values["rain"].value == pytest.approx(326.1, abs=0.1)
assert station.values["temperature"].value == pytest.approx(16.8, abs=0.1)
assert station.values["wind"].value > 0
def test_a_station_remembers_when_each_quantity_last_arrived():
garden = wx.Garden()
garden.add(reading("wind", sensor=0x777, at=10.0))
garden.add(reading("5n1", sensor=0x777, at=20.0))
station = garden.stations["5n1/0777"]
assert station.times["rain"] == 10.0
assert station.times["temperature"] == 20.0
def test_the_lowest_and_highest_of_a_quantity_are_kept_as_it_goes():
garden = wx.Garden()
for i, celsius in enumerate((17.0, 21.0, 15.0, 19.0)):
garden.add(reading(celsius=celsius, at=100.0 + i))
first, last, low, high = garden.stations["tower/1A2B"].span("temperature")
assert (first, last, low, high) == (17.0, 19.0, 15.0, 21.0)
def test_how_often_a_sensor_is_being_heard_is_the_measure_of_the_aerial():
garden = wx.Garden()
for i in range(5):
garden.add(reading(at=1_000.0 + i * 16.0))
assert garden.stations["tower/1A2B"].gap == pytest.approx(16.0)
def test_a_sensor_that_has_stopped_transmitting_leaves_the_display():
garden = wx.Garden()
garden.add(reading(at=1_000.0))
assert len(garden.showing(300.0, now=1_200.0)) == 1
assert len(garden.showing(300.0, now=1_400.0)) == 0
assert len(garden.all()) == 1 # gone from the screen, not forgotten
def test_a_garden_can_be_rebuilt_from_a_log():
readings = [reading(at=100.0), reading(sensor=0x0C41, at=101.0)]
garden = wx.Garden.of(readings)
assert {s.sensor for s in garden.all()} == {"1A2B", "0C41"}
# ---------------------------------------------------------------------------
# The log
# ---------------------------------------------------------------------------
def test_a_message_survives_being_written_and_read_back(tmp_path):
log = wl.WeatherLog(tmp_path / "weather_x.jsonl", frequency=a.ACURITE_HZ)
sent = reading(celsius=17.4, humidity=61)
log.append(sent, name="back fence")
log.close()
back = wl.read_logs([log.path])
assert len(back) == 1
got = back[0]
assert got.key == sent.key and got.at == sent.at
assert got.value("temperature") == pytest.approx(17.4)
assert got.value("humidity") == 61
assert got.channel == "A" and got.battery_low is False
def test_the_raw_bytes_go_down_underneath_whatever_was_understood(tmp_path):
"""The message is the evidence; everything else in the line is an opinion."""
log = wl.WeatherLog(tmp_path / "weather_x.jsonl")
sent = reading()
log.append(sent)
log.close()
body = json.loads(log.path.read_text().splitlines()[1])
assert bytes.fromhex(body["hex"]) == bytes(
int(sent.bits[i:i + 8], 2) for i in range(0, len(sent.bits), 8))
assert wl.read_logs([log.path])[0].bits == sent.bits
def test_the_name_a_sensor_had_at_the_time_is_written_with_the_reading(
tmp_path):
"""So a log read back in a year says where the sensor was, not where one is."""
log = wl.WeatherLog(tmp_path / "weather_x.jsonl")
log.append(reading(), name="back fence")
log.close()
assert json.loads(log.path.read_text().splitlines()[1])["name"] \
== "back fence"
def test_a_half_written_last_line_costs_that_line_and_not_the_evening(
tmp_path):
log = wl.WeatherLog(tmp_path / "weather_x.jsonl")
for i in range(5):
log.append(reading(at=1_000.0 + i))
log.close()
with open(log.path, "a") as fh:
fh.write('{"t": 1005.0, "key": "tower/1A2B", "mo')
assert len(wl.read_logs([log.path])) == 5
def test_the_header_line_is_not_read_back_as_a_reading(tmp_path):
log = wl.WeatherLog(tmp_path / "weather_x.jsonl", receiver="device 0")
log.close()
head = json.loads(log.path.read_text().splitlines()[0])
assert head["log"] == "bandsaunter-weather" and head["receiver"] == "device 0"
assert wl.read_logs([log.path]) == []
def test_readings_come_back_in_the_order_they_were_heard(tmp_path):
one = tmp_path / "weather_a.jsonl"
two = tmp_path / "weather_b.jsonl"
with wl.WeatherLog(one) as log:
log.append(reading(at=300.0))
with wl.WeatherLog(two) as log:
log.append(reading(at=100.0))
assert [r.at for r in wl.read_logs([one, two])] == [100.0, 300.0]
def test_a_log_is_written_as_it_goes_and_not_when_it_closes(tmp_path):
"""Listening ends on control-C. A log that needed a clean exit would
be empty exactly when it was most wanted."""
log = wl.WeatherLog(tmp_path / "weather_x.jsonl")
log.append(reading())
assert len(log.path.read_text().splitlines()) == 2 # header and one
log.close()
def test_the_newest_log_in_a_directory_is_the_first_one_offered(tmp_path):
old = tmp_path / "weather_2020.jsonl"
new = tmp_path / "weather_2026.jsonl"
old.write_text("{}\n")
new.write_text("{}\n")
import os
os.utime(old, (1_000_000, 1_000_000))
assert wl.logs_in(tmp_path)[0] == new
# ---------------------------------------------------------------------------
# Out to a spreadsheet
# ---------------------------------------------------------------------------
def test_the_spreadsheet_has_a_column_for_every_quantity_anything_reported(
tmp_path, book):
readings = [reading(at=100.0), reading("wind", sensor=0x777, at=101.0)]
where = wl.write_csv(tmp_path / "w.csv", readings, book)
head = where.read_text().splitlines()[0].split(",")
assert "temperature (C)" in head and "rain (mm)" in head
assert "wind from (deg)" in head
def test_a_sensor_that_does_not_report_a_quantity_leaves_it_empty(tmp_path,
book):
readings = [reading(at=100.0), reading("wind", sensor=0x777, at=101.0)]
where = wl.write_csv(tmp_path / "w.csv", readings, book)
rows = where.read_text().splitlines()
head = rows[0].split(",")
tower = rows[1].split(",")
assert tower[head.index("rain (mm)")] == ""
assert tower[head.index("temperature (C)")] != ""
def test_the_name_is_in_the_spreadsheet_because_the_identity_is_not_a_place(
tmp_path, book):
book.tag("tower/1A2B", "back fence")
where = wl.write_csv(tmp_path / "w.csv", [reading(at=100.0)], book)
assert "back fence" in where.read_text().splitlines()[1]
def test_the_unit_is_in_the_heading_and_not_beside_every_number(tmp_path,
book):
"""A column of "21.5 C" is text; a column of 21.5 is a temperature."""
where = wl.write_csv(tmp_path / "w.csv", [reading(celsius=21.5, at=1.0)],
book)
head, row = where.read_text().splitlines()[:2]
assert "temperature (C)" in head
assert row.split(",")[head.split(",").index("temperature (C)")] == "21.5"
def test_the_spreadsheet_can_be_written_the_other_way_round(tmp_path, book):
where = wl.write_csv(tmp_path / "w.csv", [reading(celsius=0.0, at=1.0)],
book, imperial=True)
head, row = where.read_text().splitlines()[:2]
assert "temperature (F)" in head
assert row.split(",")[head.split(",").index("temperature (F)")] == "32.0"
# ---------------------------------------------------------------------------
# Saying what was heard
# ---------------------------------------------------------------------------
def rendered(garden, book=None, imperial=False, width=120):
console = Console(width=width, force_terminal=False)
with console.capture() as cap:
wx.report(console, garden, book, imperial)
return cap.get()
def test_the_report_says_who_was_heard_and_what_they_said(book):
book.tag("tower/1A2B", "back fence")
garden = wx.Garden()
for i, celsius in enumerate((17.0, 21.0)):
garden.add(reading(celsius=celsius, at=1_000.0 + i * 16))
out = rendered(garden, book)
assert "back fence" in out and "1A2B" in out
assert "Tower 592TXR" in out
assert "17.0 C" in out and "21.0 C" in out
def test_a_bearing_has_no_lowest_or_highest_because_north_is_nought_and_360(
book):
"""The wind blew from four directions; only the first and last are shown.
A lowest and a highest bearing would be read as a fact about the weather
when it is a fact about arithmetic: the lowest direction of an evening is
whichever side of north the wind happened to sit.
"""
garden = wx.Garden()
for i, degrees in enumerate((315.0, 22.5, 180.0, 90.0)):
got = a.decode(a.five_in_one_wind_rain(0x777, 11.0, degrees, 1284),
confirm=False)
got.at = 100.0 + i
garden.add(got)
out = rendered(garden, book)
line = next(ln for ln in out.splitlines() if "wind from" in ln)
assert line.count("\u00b0") == 2, line
assert "315" in line and "90" in line # first and last
# The quantities that do have an order still get all four.
rain = next(ln for ln in out.splitlines() if "rain" in ln)
assert rain.count("mm") == 4, rain
def test_a_sensor_whose_model_cannot_be_read_is_still_counted(book):
odd = a.decode(a._txr_bits([0xC0, 0x11, a._status(0x1B, False),
0x01, 0x02, 0x03]), confirm=False)
odd.at = 1_000.0
garden = wx.Garden()
garden.add(odd)
out = rendered(garden, book)
assert "0011" in out
assert "cannot read" in out
def test_a_flat_battery_is_shouted_about():
garden = wx.Garden()
garden.add(reading(battery_low=True, at=1.0))
assert "low" in rendered(garden)
# ---------------------------------------------------------------------------
# The display
# ---------------------------------------------------------------------------
def shown(garden, book=None, width=120, now=None, **kw):
console = Console(width=width, force_terminal=False)
display = WeatherDisplay(console, book=book, **kw)
display.started = (now or time.time()) - 60
display.update(garden, garden.messages)
with console.capture() as cap:
console.print(display.render(now=now))
return cap.get()
def test_the_display_says_how_to_name_something_and_how_many_want_naming(book):
garden = wx.Garden()
now = time.time()
garden.add(reading(at=now))
garden.add(reading(sensor=0x0C41, at=now))
book.tag("tower/1A2B", "back fence")
out = shown(garden, book, now=now)
assert "n to name" in out and "(1 unnamed)" in out
assert "back fence" in out and "unnamed" in out
def test_the_display_shows_the_whole_sensor_and_not_half_of_it(book):
garden = wx.Garden()
now = time.time()
garden.add(reading("wind", sensor=0x777, at=now - 20))
garden.add(reading("5n1", sensor=0x777, at=now))
out = shown(garden, book, now=now)
assert "rain" in out and "temperature" in out and "wind" in out
def test_an_empty_display_says_what_to_check_rather_than_nothing():
assert "17 cm" in shown(wx.Garden())
def test_the_display_still_fits_a_narrow_terminal(book):
garden = wx.Garden()
now = time.time()
garden.add(reading(at=now))
for width in (60, 76, 92, 120, 200):
out = shown(garden, book, width=width, now=now)
assert max(len(line) for line in out.splitlines()) <= width
assert "1A2B" in out
def test_readings_are_shown_in_whichever_system_was_asked_for(book):
garden = wx.Garden()
now = time.time()
garden.add(reading(celsius=0.0, at=now))
assert "0.0 C" in shown(garden, book, now=now)
assert "32.0 F" in shown(garden, book, now=now, imperial=True)
# ---------------------------------------------------------------------------
# Listening, without a receiver
# ---------------------------------------------------------------------------
class Silence:
"""A receiver that hears nothing, for testing the loop rather than the air."""
def __init__(self, blocks=2):
self.blocks = blocks
self.tuned = 0.0
def tune(self, hz, settle=True):
self.tuned = hz
return int(hz)
def read_samples(self, count, flush=False):
if self.blocks <= 0:
return np.zeros(0, dtype=np.complex64)
self.blocks -= 1
return np.zeros(count, dtype=np.complex64)
def close(self):
return None
def test_the_receiver_is_tuned_to_one_side_of_the_sensors():
"""Its own spike would otherwise sit on top of an on-off-keyed signal."""
device = Silence()
options = wx.WeatherOptions(rate=RATE, offset=OFFSET)
wx.pump(device, options, wx.Garden(), SensorBook(path="/dev/null"), None,
time.time())
assert device.tuned == pytest.approx(a.ACURITE_HZ - OFFSET)
def test_listening_stops_when_the_receiver_does():
device = Silence(blocks=3)
total = wx.pump(device, wx.WeatherOptions(rate=RATE),
wx.Garden(), SensorBook(path="/dev/null"), None,
time.time())
assert total == 0 and device.blocks == 0
class Garden6:
"""The invented garden, wrapped so a test can drive it a block at a time."""
def __init__(self, seconds, rate=RATE, offset=OFFSET):
self.sky = a.SimulatedSensors(sample_rate=rate, offset=offset, seed=3)
self.left = seconds
def tune(self, hz, settle=True):
return int(hz)
def read_samples(self, count, flush=False):
if self.left <= 0:
return np.zeros(0, dtype=np.complex64)
self.left -= 1
return self.sky.read_samples(count)
def close(self):
return None
def test_a_whole_session_writes_a_log_names_and_a_spreadsheet(tmp_path,
monkeypatch):
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, simulate=True,
messages=True, csv=True)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(60))
book = SensorBook(path=tmp_path / "sensors.yaml")
console = Console(width=120, force_terminal=False)
with console.capture():
heard = wx.listen(console, options, str(tmp_path), book=book)
assert heard.sensors == 6
assert heard.log_path.exists() and heard.csv_path.exists()
assert len(wl.read_logs([heard.log_path])) == heard.messages
assert len(book) == 6
assert book.path.exists() # the counts were saved on the way out
def test_only_named_sensors_keeps_the_neighbours_out_of_the_log(tmp_path,
monkeypatch):
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, messages=True,
only_named=True)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(60))
book = SensorBook(path=tmp_path / "sensors.yaml")
book.tag("tower/1A2B", "back fence")
console = Console(width=120, force_terminal=False)
with console.capture():
heard = wx.listen(console, options, str(tmp_path), book=book)
assert {s.key for s in heard.garden.all()} == {"tower/1A2B"}
assert {r.key for r in wl.read_logs([heard.log_path])} == {"tower/1A2B"}
def test_a_model_that_cannot_be_read_can_be_left_off_the_display(tmp_path):
odd = a.decode(a._txr_bits([0xC0, 0x11, a._status(0x1B, False),
0x01, 0x02, 0x03]), confirm=False)
odd.at = 1_000.0
class OneOddMessage:
def tune(self, hz, settle=True):
return int(hz)
def read_samples(self, count, flush=False):
return np.zeros(0, dtype=np.complex64)
def close(self):
return None
for wanted, expect in ((True, 1), (False, 0)):
garden = wx.Garden()
options = wx.WeatherOptions(unknown=wanted)
# The filter is in the pump, so it is exercised where it lives.
if not options.unknown and not odd.measures:
expect = 0
else:
garden.add(odd)
assert len(garden.all()) == expect
def test_nothing_heard_says_what_to_check(tmp_path, monkeypatch):
monkeypatch.setattr(wx, "open_device", lambda console, opts: Silence(2))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
wx.listen(console, wx.WeatherOptions(rate=RATE, messages=True),
str(tmp_path), book=SensorBook(path=tmp_path / "s.yaml"))
assert "17 cm" in cap.get()
def test_a_receiver_that_will_not_open_is_a_message_and_not_a_traceback(
tmp_path, monkeypatch):
monkeypatch.setattr(wx, "open_device", lambda console, opts: None)
console = Console(width=120, force_terminal=False)
with console.capture():
heard = wx.listen(console, wx.WeatherOptions(), str(tmp_path))
assert heard.sensors == 0 and heard.log_path is None
# ---------------------------------------------------------------------------
# The command line
# ---------------------------------------------------------------------------
def test_a_flag_left_off_keeps_what_was_saved_rather_than_the_factory_setting():
from bandsaunter.cli import _weather_options, build_parser
args = build_parser().parse_args(["weather", "--seconds", "60"])
saved = wx.WeatherOptions(gain="40", units="imperial", hold=90.0)
got = _weather_options(args, saved)
assert got.seconds == 60.0
assert got.gain == "40" and got.units == "imperial" and got.hold == 90.0
@pytest.mark.parametrize("flags,key,value", [
(["--units", "imperial"], "units", "imperial"),
(["--only-named"], "only_named", True),
(["--all-sensors"], "only_named", False),
(["--no-unknown"], "unknown", False),
(["--no-log"], "log", False),
(["--offset", "0"], "offset", 0.0),
(["--freq", "433800000"], "frequency", 433_800_000.0),
(["--csv"], "csv", True),
(["--no-report"], "report", False),
(["--simulate"], "simulate", True),
])
def test_every_flag_reaches_the_option_it_names(flags, key, value):
from bandsaunter.cli import _weather_options, build_parser
args = build_parser().parse_args(["weather"] + flags)
assert getattr(_weather_options(args, wx.WeatherOptions()), key) == value
def test_every_option_flag_is_one_the_parser_accepts():
"""The table and the command line have to agree, or the help lies."""
from bandsaunter.cli import build_parser
known = set()
for action in build_parser()._subparsers._group_actions[0].choices[
"weather"]._actions:
known.update(action.option_strings)
for option in wx.OPTIONS:
for flag in option.flags + option.off_flags:
assert flag in known, f"{flag} is in the table and not the parser"
def test_reading_a_log_back_prints_it_and_writes_a_spreadsheet(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_readings
import bandsaunter.cli as cli
log = tmp_path / "weather_x.jsonl"
with wl.WeatherLog(log) as fh:
for i in range(4):
fh.append(reading(celsius=17.0 + i, at=1_000.0 + i * 16))
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["readings", str(log), "--csv"])
with console.capture() as cap:
assert cmd_readings(args) == 0
assert "1A2B" in cap.get()
assert log.with_suffix(".csv").exists()
def test_reading_a_log_back_can_be_narrowed_to_one_sensor(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_readings
import bandsaunter.cli as cli
log = tmp_path / "weather_x.jsonl"
with wl.WeatherLog(log) as fh:
fh.append(reading(at=1_000.0))
fh.append(reading(sensor=0x0C41, at=1_001.0))
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["readings", str(log),
"--sensor", "0C41"])
with console.capture() as cap:
assert cmd_readings(args) == 0
out = cap.get()
assert "0C41" in out and "1A2B" not in out
def test_asking_for_a_log_that_is_not_there_is_a_message(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_readings
import bandsaunter.cli as cli
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["readings", str(tmp_path / "nope.jsonl")])
with console.capture() as cap:
assert cmd_readings(args) == 1
assert "no such file" in cap.get()
def test_the_sensor_list_names_and_forgets(tmp_path, monkeypatch):
from bandsaunter.cli import build_parser, cmd_sensors
import bandsaunter.cli as cli
import bandsaunter.sensors as sensors
where = tmp_path / "sensors.yaml"
monkeypatch.setattr(sensors, "names_path", lambda directory=None: where)
book = SensorBook(path=where)
book.heard(reading())
book.save()
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["sensors", "--name", "1A2B=back fence",
"--note", "1A2B=by the gate"])
with console.capture() as cap:
assert cmd_sensors(args) == 0
assert "back fence" in cap.get() and "by the gate" in cap.get()
args = build_parser().parse_args(["sensors", "--forget", "1A2B"])
with console.capture():
cmd_sensors(args)
assert len(SensorBook(path=where)) == 0
# ---------------------------------------------------------------------------
# The menus
# ---------------------------------------------------------------------------
def test_the_menu_can_be_opened_and_left(monkeypatch):
import bandsaunter.tui as tui
from bandsaunter.config import ScanConfig
answers = iter(["b"])
monkeypatch.setattr(tui, "_ask",
lambda console, prompt, default="": next(answers))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
tui.weather_menu(console, ScanConfig())
out = cap.get()
assert "433.92 MHz" in out
for group in wx.OPTION_GROUPS:
assert group.lower() in out
def test_the_menu_and_the_aircraft_menu_share_one_set_of_option_screens():
"""One set of menus drives both sections, and will drive a third."""
import bandsaunter.tui as tui
from bandsaunter import aircraft as air
assert tui._section(None) is air
assert tui._section(wx) is wx
for module in (air, wx):
assert callable(module.defaults)
assert module.OPTION_GROUPS and module.OPTIONS
def test_an_option_can_be_found_by_typing_part_of_its_name():
import bandsaunter.tui as tui
assert [o.key for o in tui._find_options("only_named", wx)] == ["only_named"]
assert [o.key for o in tui._find_options("offset", wx)] == ["offset"]
assert tui._find_options("wingspan", wx) == []
def test_changing_a_weather_option_does_not_reach_the_aircraft_ones():
"""The two sections number their options separately, so this could bite."""
import bandsaunter.tui as tui
console = Console(width=120, force_terminal=False)
options = wx.WeatherOptions()
with console.capture() as cap:
tui._edit_option(console, options, "?1", wx)
assert wx.OPTIONS[0].label in cap.get()
# ---------------------------------------------------------------------------
# Naming a sensor while it is on the screen
# ---------------------------------------------------------------------------
class FakeLive:
"""The live display, which has to come down while a name is typed."""
def __init__(self):
self.running = True
self.stopped = 0
def stop(self):
self.running = False
self.stopped += 1
def start(self):
self.running = True
def update(self, _renderable):
pass
class FakeDisplay:
"""The table, which the loop keeps up to date whether or not it is seen."""
def __init__(self):
self.blocks = 0
def update(self, garden, messages, log_path=None):
self.blocks += 1
def render(self, *a, **kw):
return ""
class FakeKeys:
"""A keyboard that presses whatever it was handed, one key at a time."""
def __init__(self, presses=()):
self.presses = list(presses)
self.raw = False
self.history = []
def __enter__(self):
self.raw = True
self.history.append("raw")
return self
def __exit__(self, *exc):
self.raw = False
self.history.append("normal")
return False
def get(self):
return self.presses.pop(0) if self.presses else None
def answers(monkeypatch, *replies):
"""Whatever the prompts ask, in order."""
said = iter(replies)
monkeypatch.setattr("rich.prompt.Prompt.ask",
lambda *a, **kw: next(said, ""))
def test_a_sensor_can_be_named_while_it_is_on_the_screen(book, monkeypatch):
garden = wx.Garden()
now = time.time()
garden.add(reading(at=now - 10)) # heard first, so row 1
garden.add(reading(sensor=0x0C41, at=now)) # row 2
answers(monkeypatch, "2", "greenhouse")
console = Console(width=120, force_terminal=False)
live, keys = FakeLive(), FakeKeys()
with console.capture() as cap:
named = wx._name_one(console, keys, live, None, book, garden,
wx.WeatherOptions())
assert named is True
assert book.name_for("tower/0C41") == "greenhouse"
assert SensorBook(path=book.path).name_for("tower/0C41") == "greenhouse"
assert "1A2B" in cap.get() and "0C41" in cap.get()
def test_the_display_comes_down_and_the_keyboard_goes_back_to_normal(
book, monkeypatch):
"""Neither is optional: nothing typed would be visible otherwise.
The terminal reads single keys while the display is running, which is how
n is noticed at all, and a name typed in that mode is not echoed. So the
terminal has to come out of it and go back in, in that order.
"""
garden = wx.Garden()
garden.add(reading(at=time.time()))
answers(monkeypatch, "1", "back fence")
live, keys = FakeLive(), FakeKeys()
keys.__enter__()
console = Console(width=120, force_terminal=False)
with console.capture():
wx._name_one(console, keys, live, None, book, garden,
wx.WeatherOptions())
assert live.stopped == 1 and live.running is True
assert keys.history == ["raw", "normal", "raw"]
assert keys.raw is True
def test_the_receiver_is_not_stopped_while_a_name_is_typed(book, monkeypatch):
"""A slow typist loses a few seconds of weather and nothing else."""
garden = wx.Garden()
garden.add(reading(at=time.time()))
answers(monkeypatch, "1", "back fence")
live, keys = FakeLive(), FakeKeys()
console = Console(width=120, force_terminal=False)
with console.capture():
wx._name_one(console, keys, live, None, book, garden,
wx.WeatherOptions())
# Nothing here closes or pauses a device; the only thing that stops is
# the drawing. The listening loop calls this between two blocks.
assert live.running is True
@pytest.mark.parametrize("said", [("",), ("9",), ("x",), ("1", "")])
def test_changing_your_mind_about_a_name_leaves_things_as_they_were(
book, monkeypatch, said):
garden = wx.Garden()
garden.add(reading(at=time.time()))
answers(monkeypatch, *said)
console = Console(width=120, force_terminal=False)
with console.capture():
named = wx._name_one(console, FakeKeys(), FakeLive(), None, book,
garden, wx.WeatherOptions())
assert named is False
assert len(book) == 0
def test_naming_before_anything_is_heard_says_so(book, monkeypatch):
answers(monkeypatch, "1", "back fence")
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
named = wx._name_one(console, FakeKeys(), FakeLive(), None, book,
wx.Garden(), wx.WeatherOptions())
assert named is False and "nothing heard yet" in cap.get()
def test_pressing_n_while_listening_is_what_starts_the_naming(tmp_path,
monkeypatch):
"""The whole point of the key, wired end to end through the listen loop."""
import bandsaunter.ui as ui
# Nothing for the first few blocks, because there is nothing to name
# until a sensor has spoken, and then the key.
keys = FakeKeys(presses=[None] * 6 + ["n"])
monkeypatch.setattr(ui, "KeyReader", lambda *a, **kw: keys)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(40))
monkeypatch.setattr(wx, "_open_display",
lambda *a, **kw: (FakeDisplay(), FakeLive()))
answers(monkeypatch, "1", "back fence")
book = SensorBook(path=tmp_path / "sensors.yaml")
console = Console(width=120, force_terminal=False)
options = wx.WeatherOptions(rate=RATE, offset=OFFSET)
with console.capture():
heard = wx.listen(console, options, str(tmp_path), book=book)
assert heard.named == 1
assert any(s.named for s in book.ordered())
def test_pressing_q_while_listening_stops_it(tmp_path, monkeypatch):
import bandsaunter.ui as ui
device = Garden6(600)
monkeypatch.setattr(ui, "KeyReader",
lambda *a, **kw: FakeKeys(presses=["q"]))
monkeypatch.setattr(wx, "open_device", lambda console, opts: device)
monkeypatch.setattr(wx, "_open_display",
lambda *a, **kw: (FakeDisplay(), FakeLive()))
console = Console(width=120, force_terminal=False)
with console.capture():
wx.listen(console, wx.WeatherOptions(rate=RATE, offset=OFFSET),
str(tmp_path), book=SensorBook(path=tmp_path / "s.yaml"))
assert device.left > 590, "q did not stop it within a couple of blocks"
def test_a_name_given_before_the_sensor_was_heard_moves_across_when_it_is(
book):
"""Naming from the command line and then listening is the obvious order.
The name is filed under the identity alone until the first message says
which model it is; this is the moment it moves. Getting it wrong leaves
the shed and the sensor on the shed listed as two separate things, one
named and never heard, the other heard and never named.
"""
from bandsaunter.cli import _name_sensors
_name_sensors(book, ["1A2B=back fence"], quiet=True)
assert "?/1A2B" in book
book.heard(reading(at=1_000.0))
assert "?/1A2B" not in book
assert book.name_for("tower/1A2B") == "back fence"
assert [s.key for s in book.ordered()] == ["tower/1A2B"]
assert book.get("tower/1A2B").messages == 1
def test_a_waiting_name_survives_being_saved_and_read_back(book):
from bandsaunter.cli import _name_sensors
_name_sensors(book, ["5C=shed"], quiet=True)
again = SensorBook(path=book.path)
again.heard(reading("tower", sensor=0x5C, at=1_000.0))
assert again.name_for("tower/005C") == "" # a different identity
again.heard(a_609_reading())
assert again.name_for("609/5C") == "shed"
def a_609_reading(at=1_000.0):
got = a.decode(a.frame_609(0x5C, 4.2, 80), confirm=False)
got.at = at
return got
def test_a_waiting_name_does_not_get_handed_to_the_wrong_sensor(book):
"""Only an identity that matches claims it, not the next thing heard."""
from bandsaunter.cli import _name_sensors
_name_sensors(book, ["1A2B=back fence"], quiet=True)
book.heard(reading(sensor=0x0C41, at=1_000.0))
assert book.name_for("tower/0C41") == ""
assert book.name_for("?/1A2B") == "back fence"
# ---------------------------------------------------------------------------
# Saying what is arriving, for when nothing is
# ---------------------------------------------------------------------------
def test_the_diagnosis_says_what_each_second_of_band_looked_like(tmp_path,
monkeypatch):
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, diagnose=True,
log=False)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(30))
console = Console(width=140, force_terminal=False)
with console.capture() as cap:
wx.listen(console, options, str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"))
out = cap.get()
assert "noise" in out and "gate" in out and "the noise" in out
assert "pulses" in out and "gaps" in out
# The pulse lengths are the useful part: two or three, and nothing
# in between.
assert "×" in out
assert "Tower 592TXR" in out
def test_the_diagnosis_turns_the_live_table_off(tmp_path):
"""The two cannot share a screen: one redraws in place, one scrolls."""
console = Console(width=120, force_terminal=True)
display, live = wx._open_display(
console, wx.WeatherOptions(diagnose=True), None, time.time())
assert (display, live) == (None, None)
def test_raw_samples_can_be_captured_for_working_out_why(tmp_path,
monkeypatch):
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, messages=True,
log=False)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(3))
where = tmp_path / "band.cf32"
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
heard = wx.listen(console, options, str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"),
save_iq=str(where))
assert heard.iq_path == where and where.exists()
kept = np.fromfile(where, dtype=np.complex64)
assert kept.size == 3 * int(RATE)
# And it says how fast that fills, because it fills fast.
assert "MB a second" in cap.get()
def test_capturing_to_somewhere_unwritable_is_a_message_not_a_crash(
tmp_path, monkeypatch):
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, messages=True,
log=False)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Silence(2))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
heard = wx.listen(console, options, str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"),
save_iq="/proc/nowhere/band.cf32")
assert heard.iq_path is None
assert "cannot write" in cap.get()
def test_an_offset_too_big_for_the_sample_rate_is_refused_with_a_reason():
"""Half the sample rate is all the band there is to move a signal within."""
errs = wx.WeatherOptions(rate=400_000.0, offset=250_000.0).validate()
assert any("offset" in e and "sample rate" in e for e in errs)
assert wx.WeatherOptions(rate=400_000.0, offset=100_000.0).validate() == []
def test_hearing_nothing_points_at_the_diagnosis(tmp_path, monkeypatch):
from bandsaunter.cli import build_parser, cmd_weather
import bandsaunter.cli as cli
monkeypatch.setattr(wx, "open_device", lambda console, opts: Silence(2))
monkeypatch.setattr(wx, "load_options",
lambda *a, **kw: wx.WeatherOptions(rate=RATE,
offset=OFFSET,
messages=True,
log=False))
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["weather"])
with console.capture() as cap:
assert cmd_weather(args) == 1
assert "--diagnose" in cap.get()
@pytest.mark.parametrize("flags,key,value", [
(["--diagnose"], "diagnose", True),
(["--no-diagnose"], "diagnose", False),
])
def test_the_diagnosis_flags_reach_the_option(flags, key, value):
from bandsaunter.cli import _weather_options, build_parser
args = build_parser().parse_args(["weather"] + flags)
assert getattr(_weather_options(args, wx.WeatherOptions()), key) == value
def test_save_iq_is_a_path_on_the_command_line_and_not_a_saved_setting():
"""It is a one-off capture, not something to carry between runs."""
from bandsaunter.cli import build_parser
args = build_parser().parse_args(["weather", "--save-iq", "/tmp/x.cf32"])
assert args.save_iq == "/tmp/x.cf32"
assert not hasattr(wx.WeatherOptions(), "save_iq")
# ---------------------------------------------------------------------------
# The verdict: which of the four faults it was
# ---------------------------------------------------------------------------
def verdict_of(**tally):
heard = wx.Heard()
heard.survey.update(tally)
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
wx._verdict(console, heard)
return " ".join(cap.get().split())
@pytest.mark.parametrize("tally,says", [
# Nothing arrived at all: not the decoder's fault, and not the aerial's.
(dict(blocks=30, dead=30), "empty samples"),
# Nothing above the noise: the aerial.
(dict(blocks=30), "aerial"),
# Something there, never keyed: not one of these sensors.
(dict(blocks=30, loud=12), "continuously"),
# Sliced and never framed: the radio works, the protocol is not one of
# the five. This is the one where the pulse lengths matter.
(dict(blocks=30, loud=12, bursts=40), "cannot read"),
# Framed and never corroborated: a stronger signal, not a new decoder.
(dict(blocks=30, loud=12, bursts=40, framed=9), "second copy"),
(dict(blocks=30, loud=12, bursts=40, framed=9, reported=6), "Working"),
])
def test_the_verdict_tells_the_four_faults_apart(tally, says):
assert says in verdict_of(**tally)
def test_the_verdict_says_nothing_when_there_was_nothing_to_judge():
assert verdict_of() == ""
def test_the_verdict_is_given_even_when_nothing_was_heard(tmp_path,
monkeypatch):
"""Which is exactly the run whose verdict is worth reading."""
monkeypatch.setattr(wx, "open_device", lambda console, opts: Silence(3))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
heard = wx.listen(console, wx.WeatherOptions(rate=RATE, offset=OFFSET,
diagnose=True, log=False),
str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"))
assert heard.sensors == 0
assert "what that came to" in cap.get()
assert "empty samples" in cap.get()
def test_a_working_run_is_counted_as_one(tmp_path, monkeypatch):
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(30))
console = Console(width=120, force_terminal=False)
with console.capture() as cap:
heard = wx.listen(console, wx.WeatherOptions(rate=RATE, offset=OFFSET,
diagnose=True, log=False,
report=False),
str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"))
assert "Working" in cap.get()
assert heard.survey["reported"] == heard.messages
assert heard.survey["blocks"] == 30
# ---------------------------------------------------------------------------
# Working on a capture instead of on the air
# ---------------------------------------------------------------------------
def captured(tmp_path, seconds=3, rate=RATE, offset=OFFSET):
"""A recording of the invented garden, with its settings beside it."""
import json
sky = a.SimulatedSensors(sample_rate=rate, offset=offset, seed=3)
where = tmp_path / "band.cf32"
with open(where, "wb") as fh:
for _ in range(seconds):
sky.read_samples(int(rate)).astype("complex64").tofile(fh)
where.with_suffix(".cf32.json").write_text(json.dumps({
"sample_rate": rate, "frequency": a.ACURITE_HZ, "offset": offset,
"format": "complex64"}))
return where
def test_a_capture_answers_the_way_the_receiver_that_made_it_did(tmp_path):
where = captured(tmp_path, seconds=4)
replay = wx.Replay(where)
assert replay.rate == RATE and replay.offset == OFFSET
assert replay.frequency == a.ACURITE_HZ
assert len(replay) == 4 * int(RATE)
assert replay.seconds == pytest.approx(4.0)
assert replay.read_samples(int(RATE)).size == int(RATE)
def test_a_capture_runs_out_rather_than_repeating_itself(tmp_path):
replay = wx.Replay(captured(tmp_path, seconds=2))
assert replay.read_samples(int(RATE)).size == int(RATE)
assert replay.read_samples(int(RATE)).size == int(RATE)
assert replay.read_samples(int(RATE)).size == 0
def test_the_settings_are_written_beside_a_capture(tmp_path, monkeypatch):
"""A file of samples with no idea what rate it was taken at is nothing.
Read back at the wrong rate every pulse in it is the wrong length, and
nothing will ever decode however good the decoder is.
"""
import json
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, messages=True,
log=False)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(2))
where = tmp_path / "band.cf32"
console = Console(width=120, force_terminal=False)
with console.capture():
wx.listen(console, options, str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"),
save_iq=str(where))
beside = json.loads(where.with_suffix(".cf32.json").read_text())
assert beside["sample_rate"] == RATE and beside["offset"] == OFFSET
assert beside["format"] == "complex64"
assert wx.Replay(where).rate == RATE
def test_a_capture_with_no_settings_beside_it_falls_back_rather_than_failing(
tmp_path):
where = captured(tmp_path, seconds=2)
where.with_suffix(".cf32.json").unlink()
replay = wx.Replay(where, rate=RATE, offset=OFFSET)
assert replay.rate == RATE and replay.settings == {}
def test_the_same_readings_come_out_of_a_capture_as_off_the_air(tmp_path,
monkeypatch):
"""The only thing stood in for is the dongle, so it has to be the same.
That is the whole use of a capture: a recording that yields nothing here
yields nothing for anybody and the fault is in this program, and one that
yields readings here and not on the air is a setting.
"""
where = captured(tmp_path, seconds=6)
console = Console(width=120, force_terminal=False)
options = wx.WeatherOptions(rate=RATE, offset=OFFSET, messages=True,
log=False, report=False)
monkeypatch.setattr(wx, "open_device", lambda console, opts: Garden6(6))
with console.capture():
live = wx.listen(console, options, str(tmp_path),
book=SensorBook(path=tmp_path / "a.yaml"))
with console.capture():
back = wx.listen(console, options, str(tmp_path), device=wx.Replay(where),
book=SensorBook(path=tmp_path / "b.yaml"))
assert {s.key for s in back.garden.all()} == {s.key for s in live.garden.all()}
assert back.messages == live.messages
def test_a_capture_can_be_replayed_from_the_command_line(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_weather
import bandsaunter.cli as cli
where = captured(tmp_path, seconds=6)
monkeypatch.setattr(wx, "load_options",
lambda *a, **kw: wx.WeatherOptions(messages=True,
log=False,
report=False))
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["weather", "--from-iq", str(where)])
with console.capture() as cap:
assert cmd_weather(args) == 0
out = cap.get()
assert "replaying" in out and "Tower 592TXR" in out
def test_replaying_something_that_is_not_a_capture_is_a_message(tmp_path,
monkeypatch):
from bandsaunter.cli import build_parser, cmd_weather
import bandsaunter.cli as cli
empty = tmp_path / "nothing.cf32"
empty.write_bytes(b"")
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["weather", "--from-iq", str(empty)])
with console.capture() as cap:
assert cmd_weather(args) == 1
assert "no samples" in cap.get()
def test_the_capture_settings_win_over_whatever_was_saved(tmp_path,
monkeypatch):
"""Read at the wrong rate, every pulse in it is the wrong length."""
from bandsaunter.cli import build_parser, cmd_weather
import bandsaunter.cli as cli
where = captured(tmp_path, seconds=3)
monkeypatch.setattr(wx, "load_options",
lambda *a, **kw: wx.WeatherOptions(
rate=1_024_000.0, offset=250_000.0,
messages=True, log=False, report=False))
console = Console(width=120, force_terminal=False)
monkeypatch.setattr(cli, "console", console)
args = build_parser().parse_args(["weather", "--from-iq", str(where)])
with console.capture() as cap:
cmd_weather(args)
assert f"{RATE / 1e6:g} MS/s" in cap.get()
# ---------------------------------------------------------------------------
# Finding out where the sensors actually are
# ---------------------------------------------------------------------------
# Above about half a megasample a second there is room to filter, and a
# filter is applied -- which is the only circumstance in which the tuning
# offset does anything at all. At the default rate nothing after the mixer
# is narrower than the band, and turning a number does not change its size.
WIDE = 1_024_000.0
STEP = WIDE / 4.0
def test_the_tuning_offset_does_nothing_at_the_default_sample_rate():
"""Which is worth a test, because the option says so and people read it."""
iq = a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), RATE, offset=0.0,
noise=0.02)
straight = [r.bits for r in a.readings_from(iq, RATE, 0.0)]
shifted = [r.bits for r in a.readings_from(iq, RATE, RATE / 4.0)]
assert straight and straight == shifted
def test_the_diagnosis_finds_the_offset_that_would_have_worked():
"""Rather than asking somebody to try four combinations and report back.
Which way round a dongle presents its samples is not knowable from here
and is perfectly findable, so when a second plainly held something and
the configured setting read none of it, the others are tried.
"""
iq = a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), WIDE, offset=STEP,
noise=0.02)
looking_ahead = wx.WeatherOptions(rate=WIDE, offset=0.0)
assert a.readings_from(iq, WIDE, 0.0) == []
assert wx._other_settings(iq, looking_ahead) == f"--offset {STEP:.0f}"
def test_it_does_not_go_looking_when_the_setting_is_already_working():
iq = a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), WIDE, offset=STEP,
noise=0.02)
options = wx.WeatherOptions(rate=WIDE, offset=STEP)
assert a.readings_from(iq, WIDE, STEP) != []
def test_the_verdict_says_which_offset_to_use(tmp_path, monkeypatch):
step = STEP
class Misplaced:
"""A garden that is not where the receiver is looking."""
def __init__(self, blocks):
self.sky = a.SimulatedSensors(sample_rate=WIDE, offset=step,
seed=3)
self.left = blocks
def tune(self, hz, settle=True):
return int(hz)
def read_samples(self, count, flush=False):
if self.left <= 0:
return np.zeros(0, dtype=np.complex64)
self.left -= 1
return self.sky.read_samples(count)
def close(self):
return None
monkeypatch.setattr(wx, "open_device", lambda console, opts: Misplaced(25))
console = Console(width=140, force_terminal=False)
with console.capture() as cap:
heard = wx.listen(console, wx.WeatherOptions(rate=WIDE, offset=0.0,
diagnose=True, log=False,
report=False),
str(tmp_path),
book=SensorBook(path=tmp_path / "s.yaml"))
out = cap.get()
assert heard.sensors == 0
assert "try this" in out
assert f"--offset {step:.0f}" in out
# ---------------------------------------------------------------------------
# The invented garden keeps time
# ---------------------------------------------------------------------------
def test_the_invented_garden_hands_back_a_second_once_a_second_has_passed():
"""It stands in for a dongle, so it waits like one.
Without the wait a block comes back the moment it is asked for, the
garden lives several times faster than the clock its readings are
stamped with, and a capture bounded by --seconds comes out however many
times too large the machine happens to be worth.
"""
import time as clock
sky = a.SimulatedSensors(sample_rate=RATE, realtime=True)
began = clock.monotonic()
for _ in range(3):
sky.read_samples(int(RATE / 10)) # a tenth of a second each
took = clock.monotonic() - began
assert 0.15 < took < 0.6, f"three tenths of a second took {took:.2f}s"
def test_a_test_garden_does_not_wait_at_all():
import time as clock
sky = a.SimulatedSensors(sample_rate=RATE, realtime=False)
began = clock.monotonic()
for _ in range(3):
sky.read_samples(int(RATE))
assert clock.monotonic() - began < 2.0
def test_the_digital_gain_control_after_the_tuner_is_left_off(monkeypatch):
"""The tuner's own control does its job; the one after it pumps.
It winds the gain up through the silence between one burst and the next,
which lifts the noise towards the signal and squeezes the very difference
the burst detector works on. It matters here in a way it does not for
aircraft, where a frame is found by correlating a preamble over a few
microseconds rather than by comparing a burst with the quiet around it.
The established tools for this band leave it off, and so does this.
"""
import bandsaunter.device as device
asked = {}
class Fake:
def __init__(self, **kw):
asked.update(kw)
def open(self):
return self
monkeypatch.setattr(device, "RtlSdrDevice", Fake)
console = Console(width=120, force_terminal=False)
for gain in ("auto", "40"):
wx.open_device(console, wx.WeatherOptions(gain=gain))
assert asked["agc"] is False, f"digital AGC on with gain {gain!r}"
assert asked["gain"] == gain # the tuner's own is still asked for
def test_the_defaults_are_the_ones_the_established_tools_use():
"""250 kS/s, straight at 433.92 MHz, no offset.
Not a matter of taste: this is the configuration known to receive these
sensors on hardware this program has never run on, and differing from it
bought nothing and cost everything.
"""
options = wx.WeatherOptions()
assert options.rate == 250_000.0
assert options.offset == 0.0
assert options.frequency == a.ACURITE_HZ
Reference in a new issue View git blame Copy permalink