dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
bandsaunter/tests/test_acurite.py
The Dust Council 335d83f8a0 Hear every sensor in the garden, not only the loudest one 
Reported as reading nothing at all with several sensors in range.  Two faults,
either of which is enough on its own, and both of them things I assumed rather
than checked -- this was written against messages I generated myself and never
against a sensor.

The first is the threshold.  Bursts were found by setting one level per second
of band, halfway between the noise floor and the loudest thing in that second.
That is the obvious way to write it and it is wrong: a sensor on the windowsill
and a sensor at the end of the garden differ by forty decibels, so a level set
halfway to the near one sits above everything the far one ever does.  The far
ones do not come through weakly, they vanish -- and vanish only while the near
one is transmitting, which is as confusing a symptom as radio produces.  A
block with one loud sensor in it yielded exactly one sensor however many were
out there.

Finding bursts is now two passes.  The first asks only where anything happened
at all and asks it against the noise -- the bottom fifth of the second, which
is noise however busy the rest was, and which does not move when something
loud arrives.  Whatever clears that is grouped into regions, and the second
pass re-thresholds each region against its own high and low.  Every sensor is
sliced at its own amplitude.  Six sensors spanning eighty times in strength
now all come back from one second of band.

The second fault is that the slicer knew how a bit is drawn.  It read a pulse
by comparing it with the gap that followed, which is right when the gap is the
complement of the pulse so that every bit takes the same time, and wrong when
the gap is a fixed spacer: a two-hundred-and-twenty microsecond pulse against
a two-hundred microsecond spacer is the longer of the two and reads as a one,
which is the wrong bit, and then every message fails its checksum having said
nothing about why.  Nothing is assumed now -- not which of the pulse and the
gap carries the bit, not whether the gap is a complement or a spacer, not
which of long and short means one.  The same burst is read half a dozen ways
and the checksums say which reading it was, at most one being able to satisfy
one.  Copies are counted per message rather than per reading, or two readings
of one burst would corroborate each other and the rule protecting the two
thinly-checked models would protect nothing.

Both were caught the same way: by measuring, rather than by reading the code
again.  A thousand seconds of the invented garden still yields no sensor that
is not there, and reception of the ones that are is up by a quarter, because
bursts that used to be masked now decode.

And, because none of the above should have needed me: `bandsaunter weather
--diagnose` prints each second taken apart stage by stage -- the noise, the
level a burst must clear, the loudest thing in the block, then every burst
with the lengths of its pulses and gaps and whatever was made of them.  Those
lengths are the useful part: a real message has two or three of them and
nothing in between, which says at a glance whether the trouble is the radio or
the arithmetic.  At the end it says which of five things it was: nothing
arriving, nothing above the noise, something never keyed, bursts that framed
as nothing, or messages that framed and arrived only once.  `--save-iq FILE`
keeps the raw samples for whatever that cannot settle.

Full suite 2286 passed; the new work checked against six deliberately broken
builds.  Built as 2026-09-07_02.

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

744 lines
32 KiB
Python
Raw Blame History

"""The weather sensors: the messages, the checks, and getting them off the air.
Every format here is implemented from a published description, and every test
puts a reading in through the encoder and takes the same one out through the
decoder. That proves the framing, the parity, the checksums and the
arithmetic, which is what can be proved without owning one of each of these.
The other half of this file is about what must *not* be read: a run of noise,
a message with a bit wrong, a short model found inside a long one, and a
temperature no thermometer of this kind could report. On a band shared with
doorbells, car keys and tyre-pressure sensors, that half matters more.
"""
import numpy as np
import pytest
from bandsaunter import acurite as a
# A rate and an offset that keep these tests quick. Everything here works
# at 250 kS/s upwards; the program itself defaults to 1.024 MS/s, which is
# exercised by the round trip through the simulator at the bottom.
RATE = 400_000.0
OFFSET = 100_000.0
def heard(bits, coding="pwm", rate=RATE, offset=OFFSET, noise=0.05,
amplitude=1.0, seed=0):
"""One message, put on the air and taken off it again."""
iq = a.modulate(bits, rate, coding=coding, offset=offset, noise=noise,
amplitude=amplitude, seed=seed)
return a.readings_from(iq, rate, offset=offset)
# ---------------------------------------------------------------------------
# The tower sensor: what most people have
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("sensor,celsius,humidity,channel", [
(0x1A2B, 21.5, 48, "A"), (0x0001, -20.0, 5, "C"), (0x3FFF, 45.3, 100, "B"),
(0x2AAA, 0.0, 50, "C"), (0x0555, -39.9, 1, "A"),
])
def test_a_tower_reading_comes_back_as_it_was_sent(sensor, celsius, humidity,
channel):
got = a.decode(a.tower_frame(sensor, celsius, humidity, channel))
assert got is not None
assert got.family == "tower"
assert got.sensor == f"{sensor:04X}"
assert got.channel == channel
assert got.value("temperature") == pytest.approx(celsius, abs=0.05)
assert got.value("humidity") == humidity
def test_the_channel_switch_is_encoded_in_the_order_the_sensor_uses():
"""A is 3, B is 2 and C is 0, which is not the order anyone would guess."""
assert a.CHANNELS[3] == "A" and a.CHANNELS[2] == "B" and a.CHANNELS[0] == "C"
for channel in "ABC":
assert a.decode(a.tower_frame(1, 10.0, 50, channel)).channel == channel
def test_a_flat_battery_is_reported_and_a_good_one_is_not():
good = a.decode(a.tower_frame(0x1234, 20.0, 50, "A", battery_low=False))
flat = a.decode(a.tower_frame(0x1234, 20.0, 50, "A", battery_low=True))
assert good.battery_low is False
assert flat.battery_low is True
def test_a_tower_message_is_found_wherever_in_the_burst_it_starts():
bits = "1011" + a.tower_frame(0x0ABC, 12.3, 77, "B")
assert a.decode(bits).sensor == "0ABC"
# ---------------------------------------------------------------------------
# The 5-in-1, which says half of what it knows at a time
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("kph,degrees,counter", [
(0.0, 0.0, 0), (11.0, 90.0, 1284), (48.5, 337.5, 16383), (2.0, 180.0, 7),
])
def test_the_wind_message_comes_back_as_it_was_sent(kph, degrees, counter):
got = a.decode(a.five_in_one_wind_rain(0x777, kph, degrees, counter))
assert got is not None and got.family == "5n1"
assert got.value("wind") == pytest.approx(kph, abs=0.9)
assert got.value("wind from") == degrees
assert got.value("rain") == pytest.approx(counter * 0.254, abs=0.01)
def test_the_rain_counter_survives_as_well_as_the_millimetres():
"""It is a tipping bucket: the count is the evidence, the depth an opinion."""
got = a.decode(a.five_in_one_wind_rain(0x777, 5.0, 90.0, 1284))
rain = next(m for m in got.measures if m.name == "rain")
assert rain.raw == 1284
assert rain.unit == "mm"
def test_every_one_of_the_sixteen_wind_directions_comes_back():
seen = set()
for point in a.WIND_POINTS:
got = a.decode(a.five_in_one_wind_rain(0x777, 10.0, point, 0))
assert got.value("wind from") == point
seen.add(point)
assert len(seen) == 16
def test_the_weather_message_comes_back_as_it_was_sent():
got = a.decode(a.five_in_one_weather(0x777, 11.0, 16.8, 71))
assert got.message == 0x38
assert got.value("temperature") == pytest.approx(16.8, abs=0.1)
assert got.value("humidity") == 71
assert got.value("wind") == pytest.approx(11.0, abs=0.9)
def test_the_two_halves_of_a_5n1_are_different_messages_from_one_sensor():
wind = a.decode(a.five_in_one_wind_rain(0x777, 11.0, 90.0, 12))
weather = a.decode(a.five_in_one_weather(0x777, 11.0, 16.8, 71))
assert wind.key == weather.key == "5n1/0777"
assert wind.message != weather.message
def test_a_stopped_anemometer_reads_as_nothing_and_not_as_a_breeze():
"""The published conversion has an offset, so zero has to be a special case."""
assert a.decode(a.five_in_one_wind_rain(0x777, 0.0, 0.0, 0)).value("wind") == 0.0
# ---------------------------------------------------------------------------
# The lightning detector
# ---------------------------------------------------------------------------
def test_the_lightning_detector_reports_the_weather_and_the_storm():
got = a.decode(a.lightning_frame(0x311, 19.1, 58, strikes=7, miles=12))
assert got.family == "6045"
assert got.value("temperature") == pytest.approx(19.1, abs=0.1)
assert got.value("humidity") == 58
assert got.value("strikes") == 7
assert got.value("storm") == pytest.approx(12 * 1.609344, abs=0.1)
def test_a_storm_out_of_range_is_not_reported_as_a_distance():
"""Thirty-one means "further off than this can tell", not thirty-one miles."""
got = a.decode(a.lightning_frame(0x311, 19.1, 58, strikes=1, miles=31))
assert got.value("storm") is None
assert got.value("strikes") == 1
def test_interference_is_reported_because_a_strike_count_under_it_is_not_real():
quiet = a.decode(a.lightning_frame(0x311, 19.1, 58, interference=False))
noisy = a.decode(a.lightning_frame(0x311, 19.1, 58, interference=True))
assert "interference" not in quiet.checks
assert "interference" in noisy.checks
# ---------------------------------------------------------------------------
# The two older ones, which carry one byte of check between them
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("celsius", [-39.5, -0.1, 0.0, 12.3, 45.0])
def test_a_609_reading_comes_back_as_it_was_sent(celsius):
got = a.decode(a.frame_609(0x5C, celsius, 80), confirm=False)
assert got.family == "609" and got.sensor == "5C"
assert got.value("temperature") == pytest.approx(celsius, abs=0.05)
assert got.value("humidity") == 80
@pytest.mark.parametrize("celsius", [-39.5, -0.1, 0.0, 12.3, 45.0])
def test_a_606_reading_comes_back_as_it_was_sent(celsius):
got = a.decode(a.frame_606(0x93, celsius), confirm=False)
assert got.family == "606" and got.sensor == "93"
assert got.value("temperature") == pytest.approx(celsius, abs=0.05)
def test_the_thinly_checked_models_are_not_believed_the_first_time():
"""One byte of check is one false message in two hundred and fifty-six.
These sensors send everything three times, so asking for two of them
costs nothing and is the difference between a decoder that can be run on
a shared band and one that cannot. The copies are separate bursts, ten
milliseconds apart, so the counting happens over a whole block.
"""
for bits in (a.frame_609(0x5C, 4.2, 80), a.frame_606(0x93, -3.5)):
once = a.candidates(bits)
assert once and a.confirmed(once) == []
assert len(a.confirmed(once + a.candidates(bits))) == 1
def test_a_thin_message_that_arrives_once_off_the_air_is_not_reported():
"""The same rule, from the antenna rather than from a bit string."""
iq = a.modulate(a.frame_609(0x5C, 4.2, 80), RATE, coding="ppm",
offset=OFFSET, repeats=1, noise=0.02)
assert a.readings_from(iq, RATE, offset=OFFSET) == []
iq = a.modulate(a.frame_609(0x5C, 4.2, 80), RATE, coding="ppm",
offset=OFFSET, repeats=2, noise=0.02)
assert len(a.readings_from(iq, RATE, offset=OFFSET)) == 1
def test_the_well_checked_models_are_believed_the_first_time():
"""Twelve to fourteen bits of check does not need a second opinion."""
for bits in (a.tower_frame(1, 10.0, 50), a.five_in_one_weather(1, 5.0, 10.0, 50),
a.lightning_frame(1, 10.0, 50)):
assert a.decode(bits) is not None
# ---------------------------------------------------------------------------
# What must not be read
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("bits", [
a.tower_frame(0x1234, 21.5, 48, "A"),
a.five_in_one_wind_rain(0x777, 11.0, 90.0, 1284),
a.lightning_frame(0x311, 19.1, 58, 3, 12),
])
def test_a_message_with_a_bit_wrong_is_never_read_as_that_sensor(bits):
"""The property that matters, stated as narrowly as it is true.
A thirteen-bit check refuses about eight thousand messages in eight
thousand and one, and this tries a couple of thousand corruptions, so
"nothing ever gets through" is not something that can honestly be
asserted. What can be, and what a person watching actually depends on,
is that a corrupted message is never attributed to the sensor that sent
it: the temperature on the screen beside "back fence" is either what the
back fence said or nothing at all.
"""
truth = a.decode(bits, confirm=False)
slipped = 0
for i in range(len(bits)):
broken = list(bits)
broken[i] = "1" if broken[i] == "0" else "0"
got = a.decode("".join(broken), confirm=False)
if got is None:
continue
slipped += 1
assert (got.family, got.sensor) != (truth.family, truth.sensor), \
f"bit {i} came back as the same sensor saying something else"
assert slipped <= 2, f"{slipped} of {len(bits)} corruptions framed"
def test_parity_is_what_stops_a_run_of_zeroes_becoming_a_message():
"""A byte of zeroes has even parity, and the payload bytes must be odd."""
assert a.decode("0" * 80) is None
assert a.parity8(0x00) == 0
def test_a_reading_outside_what_the_sensor_can_report_is_refused():
"""A checksum can be satisfied by a message the hardware cannot send."""
boiling = a.tower_frame(0x1234, 130.0, 50, "A") # 130 C on a fence post
assert a.decode(boiling) is None
steam = a.tower_frame(0x1234, 20.0, 120, "A") # 120% humidity
assert a.decode(steam) is None
def test_a_short_message_is_not_read_out_of_the_middle_of_a_long_one():
"""The mistake this guards against, put in on purpose.
A five-byte message inside an eight-byte one satisfies its own eight-bit
sum about once in every two hundred and fifty-six bursts, and would show
up on the display as a sensor that is not there. What tells them apart
is that a real message runs to the end of the burst.
"""
long_one = a.five_in_one_wind_rain(0x777, 11.0, 90.0, 1284)
got = a.decode(long_one, confirm=False)
assert got is not None and got.family == "5n1"
assert [r.family for r in a.candidates(long_one)] == ["5n1"]
def test_two_messages_sharing_bits_do_not_both_survive():
for bits in (a.tower_frame(0x1A2B, 21.5, 48), a.frame_609(0x5C, 4.2, 80),
a.lightning_frame(0x311, 19.1, 58)):
found = a.candidates("0" + bits)
spans = [(r.offset, r.offset + len(r.bits)) for r in found]
for i, (start, end) in enumerate(spans):
for other_start, other_end in spans[i + 1:]:
assert not (start < other_end and other_start < end)
def test_almost_nothing_is_read_out_of_random_bits():
rng = np.random.default_rng(1)
accepted = sum(1 for _ in range(4000)
if a.decode("".join(rng.integers(0, 2, 80).astype(str)),
confirm=False) is not None)
# Five models are tried at every offset of every burst, so the bar is a
# rate rather than zero. What reaches this off the air has also had to
# be a burst of on-off keying with the right shape.
assert accepted <= 20, f"{accepted} of 4000 random runs were believed"
def test_nothing_at_all_is_read_out_of_receiver_noise():
rng = np.random.default_rng(4)
for _ in range(12):
noise = (rng.standard_normal(200_000)
+ 1j * rng.standard_normal(200_000)).astype(np.complex64)
assert a.readings_from(noise * 0.05, RATE, offset=OFFSET) == []
# ---------------------------------------------------------------------------
# Off the air: the slicer
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("bits,coding", [
(a.tower_frame(0x1A2B, 21.5, 48, "A"), "pwm"),
(a.five_in_one_wind_rain(0x777, 11.0, 90.0, 1284), "pwm"),
(a.five_in_one_weather(0x777, 11.0, 16.8, 71), "pwm"),
(a.lightning_frame(0x311, 19.1, 58, 3, 12), "pwm"),
(a.frame_609(0x5C, 4.2, 80), "ppm"),
(a.frame_606(0x93, -3.5), "ppm"),
])
def test_every_model_survives_the_whole_path_from_the_air(bits, coding):
got = heard(bits, coding)
assert len(got) == 1, f"{len(got)} readings, wanted one"
assert got[0].bits == bits
@pytest.mark.parametrize("rate,offset", [
(250_000.0, 60_000.0), (400_000.0, 100_000.0), (1_024_000.0, 250_000.0),
(2_048_000.0, 500_000.0),
])
def test_it_works_at_every_sample_rate_the_options_allow(rate, offset):
got = heard(a.tower_frame(0x1A2B, 21.5, 48, "A"), rate=rate, offset=offset)
assert [r.sensor for r in got] == ["1A2B"]
def test_the_last_bit_of_a_burst_is_recovered():
"""The gap after the final pulse is silence, not part of the bit.
A slicer that reads the bit from that gap loses the last bit of the
checksum, which loses the message -- so this is a message whose final bit
is a one, which is the case that fails if the fallback is not there.
"""
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
assert bits[-1] == "1"
assert [r.sensor for r in heard(bits)] == ["1A2B"]
def test_a_weak_sensor_at_the_end_of_the_garden_is_still_read():
got = heard(a.tower_frame(0x0C41, 20.9, 44, "B"), amplitude=0.08,
noise=0.01)
assert [r.sensor for r in got] == ["0C41"]
def test_the_receivers_own_spike_is_kept_off_the_signal():
"""Tuned straight at an on-off-keyed signal, the spike fills in the gaps.
The spike is a constant added at the tuned frequency, so this puts one
there and checks that tuning to one side and shifting back reads the
sensor while tuning straight at it does not.
"""
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
rate, offset = 1_024_000.0, 250_000.0
clean = a.modulate(bits, rate, offset=offset, amplitude=0.6, noise=0.02)
spike = np.full(clean.size, 4.0, dtype=np.complex64) # at the centre
assert [r.sensor for r in a.readings_from(clean + spike, rate,
offset=offset)] == ["1A2B"]
# The same samples read as if the receiver had been tuned at the sensor:
# the spike is now on top of it and there is nothing to slice.
assert a.readings_from(clean + spike, rate, offset=0.0) == []
def test_a_burst_of_evenly_spaced_gaps_is_not_read_as_pulse_position():
"""In that coding the gap is the bit, so gaps all one length carry none."""
even = a.Burst(marks=(400.0,) * 12, spaces=(400.0,) * 11)
assert a.bits_ppm(even) == ""
assert a.bits_pwm(even) == "0" * 12
def test_a_long_silence_ends_a_burst_and_a_short_one_does_not():
rate = 250_000.0
envelope = np.zeros(int(rate), dtype=np.float32)
per_us = rate / 1e6
def key(at_us, length_us):
lo = int(at_us * per_us)
envelope[lo:lo + int(length_us * per_us)] = 1.0
for i in range(20): # one burst, 600 us apart
key(1_000 + i * 600, 400)
for i in range(20): # another, 20 ms later
key(35_000 + i * 600, 400)
found = a.bursts(envelope, rate)
assert len(found) == 2
assert found[0].pulses == found[1].pulses == 20
def test_one_sample_of_noise_does_not_split_a_pulse_into_three():
rate = 250_000.0
envelope = np.zeros(int(rate * 0.05), dtype=np.float32)
per_us = rate / 1e6
for i in range(20):
lo = int((1_000 + i * 600) * per_us)
envelope[lo:lo + int(400 * per_us)] = 1.0
envelope[int(1_200 * per_us)] = 0.0 # a hole in the middle of one
found = a.bursts(envelope, rate)
assert len(found) == 1 and found[0].pulses == 20
def test_a_burst_too_short_to_be_a_message_is_ignored():
rate = 250_000.0
envelope = np.zeros(int(rate * 0.05), dtype=np.float32)
per_us = rate / 1e6
for i in range(3):
lo = int((1_000 + i * 600) * per_us)
envelope[lo:lo + int(400 * per_us)] = 1.0
assert a.bursts(envelope, rate) == []
def test_a_transmitter_running_ten_per_cent_fast_is_read_anyway():
"""Nothing is measured against a clock, so the drift cannot matter.
Which is just as well: these transmitters are unlocked and change
frequency and rate with the temperature, and an outdoor sensor in
January is not the one that was on the fence in July.
"""
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
marks, spaces = a.pulse_train(bits, "pwm")
per_us = RATE / 1e6
parts = []
for mark, space in zip(marks, spaces + [600.0]):
parts.append(np.full(int(mark * 1.1 * per_us), 1.0, dtype=np.float32))
parts.append(np.zeros(int(space * 1.1 * per_us), dtype=np.float32))
envelope = np.concatenate([np.zeros(int(2000 * per_us),
dtype=np.float32)] + parts)
found = a.bursts(envelope, RATE)
assert len(found) == 1
assert a.decode(a.bits_pwm(found[0])).sensor == "1A2B"
# ---------------------------------------------------------------------------
# Sensors that are not there
# ---------------------------------------------------------------------------
def test_every_invented_sensor_is_heard_within_a_few_minutes():
sky = a.SimulatedSensors(sample_rate=RATE, offset=OFFSET, seed=3)
seen = set()
for i in range(70):
for reading in a.readings_from(sky.read_samples(int(RATE)), RATE,
offset=OFFSET, when=float(i)):
seen.add(reading.key)
assert seen == {s.family + "/" + (f"{s.sensor:04X}" if s.family in
("tower", "5n1", "6045")
else f"{s.sensor:02X}")
for s in a.default_sensors()}
def test_nothing_that_is_not_there_is_heard_either():
"""Every reading over five minutes of an invented garden is a real sensor."""
sky = a.SimulatedSensors(sample_rate=RATE, offset=OFFSET, seed=3)
real = {f"{s.family}/{s.sensor:04X}" if s.family in ("tower", "5n1", "6045")
else f"{s.family}/{s.sensor:02X}" for s in a.default_sensors()}
for i in range(300):
for reading in a.readings_from(sky.read_samples(int(RATE)), RATE,
offset=OFFSET, when=float(i)):
assert reading.key in real, f"{reading.describe()} is not out there"
def test_the_5n1_alternates_its_two_messages():
sky = a.SimulatedSensors(sample_rate=RATE, offset=OFFSET, seed=3)
kinds = set()
for i in range(90):
for reading in a.readings_from(sky.read_samples(int(RATE)), RATE,
offset=OFFSET, when=float(i)):
if reading.family == "5n1":
kinds.add(reading.message)
assert kinds == {0x31, 0x38}
def test_the_same_seed_gives_the_same_garden_twice():
def run():
sky = a.SimulatedSensors(sample_rate=RATE, offset=OFFSET, seed=9)
return [(r.key, r.bits) for i in range(20)
for r in a.readings_from(sky.read_samples(int(RATE)), RATE,
offset=OFFSET, when=float(i))]
assert run() == run()
def test_a_reading_is_stamped_with_the_moment_it_arrived():
"""Not its offset in a buffer: everything downstream of this is a clock."""
sky = a.SimulatedSensors(sample_rate=RATE, offset=OFFSET, seed=3)
for i in range(30):
for reading in a.readings_from(sky.read_samples(int(RATE)), RATE,
offset=OFFSET, when=1_700_000_000.0 + i):
assert 1_700_000_000.0 + i <= reading.at < 1_700_000_001.0 + i
def test_a_message_heard_three_times_comes_back_once_and_says_so():
got = heard(a.tower_frame(0x1A2B, 21.5, 48, "A"))
assert len(got) == 1
assert got[0].copies == 3
# ---------------------------------------------------------------------------
# Saying it out loud
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("value,unit,metric,imperial", [
(21.5, "C", "21.5 C", "70.7 F"),
(48.0, "%", "48%", "48%"),
(16.0, "km/h", "16.0 km/h", "9.9 mph"),
(25.4, "mm", "25.4 mm", "1.00 in"),
(16.0, "km", "16 km", "10 mi"),
])
def test_a_measurement_reads_the_same_in_either_system(value, unit, metric,
imperial):
measure = a.Measure("x", value, unit)
assert a.format_measure(measure, False) == metric
assert a.format_measure(measure, True) == imperial
def test_a_bearing_is_given_a_name_as_well_as_a_number():
assert a.format_measure(a.Measure("wind from", 90.0, "deg")) == "90° E"
assert a.compass(0.0) == "N" and a.compass(359.0) == "N"
assert a.compass(180.0) == "S" and a.compass(247.5) == "WSW"
def test_a_sensor_is_filed_under_its_family_and_its_identity():
"""Not its channel: the switch is on the outside and someone will move it."""
one = a.decode(a.tower_frame(0x1A2B, 20.0, 50, "A"))
two = a.decode(a.tower_frame(0x1A2B, 20.0, 50, "C"))
assert one.key == two.key == "tower/1A2B"
def test_a_message_that_frames_and_is_not_understood_is_still_reported():
"""Knowing something is out there transmitting is worth a line."""
odd = a._txr_bits([0xC0, 0x11, a._status(0x1B, False), 0x01, 0x02, 0x03])
got = a.decode(odd, confirm=False)
assert got is not None
assert got.measures == ()
assert got.sensor == "0011"
assert "not understood" in got.describe()
def test_a_nine_byte_sensor_that_is_not_a_lightning_detector_is_not_read_as_one():
"""The Atlas is nine bytes too, and lays its payload out differently.
Nothing but the message type tells them apart, so a nine-byte message of
any other type comes back with its identity and no weather -- rather than
a temperature read off the wrong bits, which would pass the checksum, pass
the parity, and be wrong.
"""
atlas = a._txr_bits([0xC0, 0x11, a._status(0x06, False),
0x22, 0x11, 0x33, 0x44, 0x55])
got = a.decode(atlas, confirm=False)
assert got is not None
assert got.measures == ()
assert got.sensor == "0011"
assert got.value("temperature") is None
# ---------------------------------------------------------------------------
# A garden with more than one sensor in it
# ---------------------------------------------------------------------------
def block_of(*bursts_in, seconds: float = 1.0, noise: float = 0.02,
rate: float = RATE, offset: float = OFFSET, seed: int = 0):
"""One second of band with whatever was handed in placed about in it."""
rng = np.random.default_rng(seed)
n = int(rate * seconds)
block = ((rng.standard_normal(n) + 1j * rng.standard_normal(n))
* noise).astype(np.complex64)
for at, part in bursts_in:
start = int(at * rate)
room = min(part.size, max(0, n - start))
block[start:start + room] += part[:room]
return block
def keyed(bits, amplitude: float = 1.0, rate: float = RATE,
offset: float = OFFSET):
return a.modulate(bits, rate, offset=offset, amplitude=amplitude,
lead_us=0.0, noise=0.0)
def test_a_sensor_by_the_aerial_does_not_hide_the_rest_of_the_garden():
"""The fault that had this reading one sensor out of six.
A threshold set halfway between the noise and the loudest thing in the
block is halfway to whichever sensor happens to be nearest, and every
quieter sensor is then below it -- so they disappear, and disappear only
while the near one is transmitting, which is as confusing a symptom as
radio produces. Thirty-six decibels between these two.
"""
loud = keyed(a.tower_frame(0x1A2B, 21.5, 48, "A"), amplitude=8.0)
faint = keyed(a.tower_frame(0x0C41, 3.2, 91, "B"), amplitude=0.12)
block = block_of((0.05, loud), (0.5, faint))
heard_now = {r.sensor for r in a.readings_from(block, RATE, offset=OFFSET)}
assert heard_now == {"1A2B", "0C41"}
@pytest.mark.parametrize("apart", [4.0, 20.0, 80.0])
def test_two_sensors_are_both_read_however_far_apart_in_strength(apart):
loud = keyed(a.tower_frame(0x1A2B, 21.5, 48, "A"), amplitude=0.9)
faint = keyed(a.tower_frame(0x0C41, 3.2, 91, "B"), amplitude=0.9 / apart)
block = block_of((0.05, loud), (0.5, faint), noise=0.9 / apart / 12.0)
got = {r.sensor for r in a.readings_from(block, RATE, offset=OFFSET)}
assert got == {"1A2B", "0C41"}, f"{apart:g}x apart: heard {got}"
def test_six_sensors_in_one_second_all_come_back():
parts = [(0.02 + i * 0.14,
keyed(a.tower_frame(0x100 + i, 10.0 + i, 50, "A"),
amplitude=0.15 * (i + 1)))
for i in range(6)]
got = {r.sensor for r in a.readings_from(block_of(*parts), RATE,
offset=OFFSET)}
assert got == {f"{0x100 + i:04X}" for i in range(6)}
# ---------------------------------------------------------------------------
# Not assuming how a bit is drawn
# ---------------------------------------------------------------------------
def transmitted(bits, one_mark, zero_mark, gap, sync_mark=600.0,
sync_gap=600.0, syncs=4, copies=3, amplitude=1.0):
"""A sensor keyed with whatever timings, rather than with mine."""
per_us = RATE / 1e6
parts = [np.zeros(int(3_000 * per_us), dtype=np.float32)]
def push(mark, space):
parts.append(np.full(int(round(mark * per_us)), amplitude,
dtype=np.float32))
parts.append(np.zeros(int(round(space * per_us)), dtype=np.float32))
for _ in range(copies):
for _ in range(syncs):
push(sync_mark, sync_gap)
for bit in bits:
push(one_mark if bit == "1" else zero_mark, gap)
parts.append(np.zeros(int(9_000 * per_us), dtype=np.float32))
return a._to_air(np.concatenate(parts), RATE, OFFSET, 0.02, 0)
@pytest.mark.parametrize("name,one,zero,gap", [
# The gap is the complement of the pulse, so every bit takes the same
# time. This is the one it was written against.
("complementary gap", 408.0, 220.0, None),
# The gap is a fixed spacer. Judged against a 200 us spacer a 220 us
# pulse is the longer of the two and reads as a one, which is the wrong
# bit, and every message fails its checksum saying nothing about why.
("short fixed gap", 408.0, 220.0, 200.0),
("long fixed gap", 408.0, 220.0, 500.0),
# And the other way up: the short pulse is the one.
("inverted", 220.0, 408.0, 200.0),
])
def test_a_burst_is_read_whichever_way_the_bits_are_drawn(name, one, zero,
gap):
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
if gap is None:
# complementary: build it a bit at a time so each gap completes its
# own bit period
per_us = RATE / 1e6
parts = [np.zeros(int(3_000 * per_us), dtype=np.float32)]
for _ in range(3):
for mark, space in zip(*a.pulse_train(bits, "pwm")):
parts.append(np.full(int(round(mark * per_us)), 1.0,
dtype=np.float32))
parts.append(np.zeros(int(round(space * per_us)),
dtype=np.float32))
parts.append(np.zeros(int(9_000 * per_us), dtype=np.float32))
iq = a._to_air(np.concatenate(parts), RATE, OFFSET, 0.02, 0)
else:
iq = transmitted(bits, one, zero, gap)
got = a.readings_from(iq, RATE, offset=OFFSET)
assert [r.sensor for r in got] == ["1A2B"], f"{name}: heard {got}"
def test_the_readings_of_a_burst_are_all_different_from_each_other():
"""Half a dozen ways of reading it, and no duplicates among them."""
burst = a.bursts(*a.baseband(
a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), RATE, offset=OFFSET,
noise=0.02), RATE, OFFSET))[0]
tries = a.slicings(burst)
assert len(tries) >= 4
assert len(set(tries)) == len(tries)
assert a.tower_frame(0x1A2B, 21.5, 48, "A") in "".join(tries)
def test_reading_a_burst_several_ways_is_not_the_same_as_hearing_it_twice():
"""Corroboration counts messages, not readings of one message.
The two thinly-checked models are believed when the same message arrives
twice. If two ways of reading one burst each produced it, that would
look like two arrivals and the rule would protect nothing.
"""
iq = a.modulate(a.frame_609(0x5C, 4.2, 80), RATE, coding="ppm",
offset=OFFSET, repeats=1, noise=0.02)
burst = a.bursts(*a.baseband(iq, RATE, OFFSET))[0]
ways = [r for bits in a.slicings(burst) for r in a.candidates(bits)]
assert any(r.family == "609" for r in ways), "it did frame"
assert a.readings_from(iq, RATE, offset=OFFSET) == []
# ---------------------------------------------------------------------------
# Saying what arrived, when nothing decodes
# ---------------------------------------------------------------------------
def test_the_survey_reports_the_same_readings_the_program_acts_on():
"""A diagnostic that disagrees with the thing it diagnoses is worse than
none, so it runs the ordinary path rather than repeating it."""
iq = a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), RATE, offset=OFFSET,
noise=0.05)
look = a.survey(iq, RATE, OFFSET, when=1_000.0)
assert [r.describe() for r in look.readings] == \
[r.describe() for r in a.readings_from(iq, RATE, offset=OFFSET,
when=1_000.0)]
def test_the_survey_separates_nothing_arriving_from_nothing_decoding():
rng = np.random.default_rng(2)
n = int(RATE)
quiet = ((rng.standard_normal(n) + 1j * rng.standard_normal(n))
* 0.02).astype(np.complex64)
nothing = a.survey(quiet, RATE, OFFSET)
assert nothing.seen == [] and nothing.readings == []
assert nothing.loudest < 3.0 # and it says the band was quiet
# A burst of the right shape whose bits are nonsense: it groups, it
# slices, and it frames nothing. A different fault, and it looks it.
rubbish = transmitted("01" * 28, 408.0, 220.0, 220.0)
junk = a.survey(rubbish, RATE, OFFSET)
assert junk.seen and junk.readings == []
assert all(framed is None for _b, _t, framed in junk.seen)
assert junk.loudest > 3.0
def test_the_survey_shows_a_message_that_framed_and_was_not_corroborated():
"""Which is a third fault again, and the one hardest to guess at."""
iq = a.modulate(a.frame_609(0x5C, 4.2, 80), RATE, coding="ppm",
offset=OFFSET, repeats=1, noise=0.02)
look = a.survey(iq, RATE, OFFSET)
assert look.readings == []
assert any(framed is not None and framed.family == "609"
for _b, _t, framed in look.seen)
def test_the_pulse_lengths_of_a_burst_are_reported_as_the_protocol_shape():
burst = a.bursts(*a.baseband(
a.modulate(a.tower_frame(0x1A2B, 21.5, 48, "A"), RATE, offset=OFFSET,
noise=0.02), RATE, OFFSET))[0]
marks = a.timings(burst.marks)
assert len(marks) == 3 # short, long, sync
assert [n for _v, n in marks] == [28, 28, 4]
assert [round(v / 10) * 10 for v, _n in marks] == [220, 400, 600]
assert sum(n for _v, n in marks) == burst.pulses
def test_a_burst_of_one_length_is_reported_as_one_length():
assert a.timings([400.0] * 12) == [(400.0, 12)]
assert len(a.timings([200.0] * 6 + [400.0] * 6)) == 2
Reference in a new issue View git blame Copy permalink