706c632f47
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
943 lines
41 KiB
Python
943 lines
41 KiB
Python
"""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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Timings other than the ones this was written against
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def keyed_at(bits, marks_gaps, copies=3, reset_us=8_000.0, amplitude=1.0,
|
|
rate=250_000.0, lead_us=3_000.0):
|
|
"""A sensor keyed with whatever timings, one (mark, gap) pair per bit."""
|
|
per_us = rate / 1e6
|
|
parts = [np.zeros(int(lead_us * per_us), dtype=np.float32)]
|
|
|
|
def push(mark, gap):
|
|
parts.append(np.full(int(round(mark * per_us)), amplitude,
|
|
dtype=np.float32))
|
|
if gap:
|
|
parts.append(np.zeros(int(round(gap * per_us)), dtype=np.float32))
|
|
|
|
for _ in range(copies):
|
|
for mark, gap in marks_gaps:
|
|
push(mark, gap)
|
|
push(500.0, reset_us)
|
|
return a._to_air(np.concatenate(parts), rate, 0.0, 0.02, 0)
|
|
|
|
|
|
def pwm_pairs(bits, one=408.0, zero=220.0, gap=None, syncs=4,
|
|
sync_mark=600.0, sync_gap=600.0):
|
|
"""Pulse-width keying: the bit is the pulse. ``gap=None`` completes the
|
|
bit period, which is one of the two conventions; a number is a spacer,
|
|
which is the other."""
|
|
out = [(sync_mark, sync_gap)] * syncs
|
|
for bit in bits:
|
|
mark = one if bit == "1" else zero
|
|
out.append((mark, (one + zero) - mark if gap is None else gap))
|
|
return out
|
|
|
|
|
|
def ppm_pairs(bits, mark=500.0, short=1_000.0, long=2_000.0):
|
|
"""Pulse-position keying: the bit is the gap."""
|
|
return [(mark, long if bit == "1" else short) for bit in bits]
|
|
|
|
|
|
@pytest.mark.parametrize("name,pairs,reset", [
|
|
# The two pulse-width conventions, at the published widths.
|
|
("pwm, gap completes the bit", pwm_pairs, 8_000.0),
|
|
# ...and with the pulses stretched and squeezed, because these
|
|
# transmitters are unlocked and drift with the temperature.
|
|
("pwm 20% slow", lambda b: [(m * 1.2, g * 1.2)
|
|
for m, g in pwm_pairs(b)], 9_600.0),
|
|
("pwm 15% fast", lambda b: [(m * 0.85, g * 0.85)
|
|
for m, g in pwm_pairs(b)], 6_800.0),
|
|
# A fixed spacer rather than a complement, at three widths.
|
|
("pwm, 200 us spacer", lambda b: pwm_pairs(b, gap=200.0), 8_000.0),
|
|
("pwm, 400 us spacer", lambda b: pwm_pairs(b, gap=400.0), 8_000.0),
|
|
("pwm, 600 us spacer", lambda b: pwm_pairs(b, gap=600.0), 8_000.0),
|
|
# Different numbers of sync pulses, since nothing here counts them.
|
|
("pwm, no sync", lambda b: pwm_pairs(b, syncs=0), 8_000.0),
|
|
("pwm, 8 sync pulses", lambda b: pwm_pairs(b, syncs=8), 8_000.0),
|
|
# And the short pulse meaning one.
|
|
("pwm inverted", lambda b: pwm_pairs(b, one=220.0, zero=408.0,
|
|
gap=200.0), 8_000.0),
|
|
])
|
|
def test_a_tower_sensor_is_read_at_timings_other_than_the_expected_ones(
|
|
name, pairs, reset):
|
|
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
|
|
build = pairs if callable(pairs) else (lambda b: pairs)
|
|
iq = keyed_at(bits, build(bits), reset_us=reset)
|
|
got = a.readings_from(iq, 250_000.0)
|
|
assert [r.sensor for r in got] == ["1A2B"], f"{name}: heard {got}"
|
|
|
|
|
|
@pytest.mark.parametrize("name,short,long,reset", [
|
|
# The 609TXC's published gaps.
|
|
("609 gaps", 1_000.0, 2_000.0, 10_000.0),
|
|
# The 606TX's, which are twice as wide -- wider than any grouping tight
|
|
# enough to keep two sensors apart, which is why the pieces of a message
|
|
# are put back together after being sliced rather than before.
|
|
("606 gaps", 2_000.0, 4_000.0, 8_000.0),
|
|
("half again as wide", 3_000.0, 6_000.0, 14_000.0),
|
|
])
|
|
def test_a_gap_keyed_sensor_is_read_however_wide_its_gaps_are(name, short,
|
|
long, reset):
|
|
bits = a.frame_609(0x5C, 4.2, 80)
|
|
iq = keyed_at(bits, ppm_pairs(bits, short=short, long=long),
|
|
reset_us=reset)
|
|
got = a.readings_from(iq, 250_000.0)
|
|
assert [r.sensor for r in got] == ["5C"], f"{name}: heard {got}"
|
|
|
|
|
|
def test_three_copies_run_together_are_still_three_copies():
|
|
"""A burst too long to be one message is cut at its largest gaps.
|
|
|
|
The gaps inside a 606 message and the wait between its copies are only a
|
|
factor of two apart, which is too close to tell from a ratio -- so the
|
|
joining is allowed to be greedy and the result is divided by counting
|
|
instead.
|
|
"""
|
|
bits = a.frame_606(0x93, -3.5)
|
|
iq = keyed_at(bits, ppm_pairs(bits, short=2_000.0, long=4_000.0),
|
|
reset_us=8_000.0)
|
|
envelope, rate = a.baseband(iq, 250_000.0, 0.0)
|
|
found = a.bursts(envelope, rate)
|
|
assert len(found) == 3, [b.pulses for b in found]
|
|
assert all(b.pulses <= a.MAX_PULSES for b in found)
|
|
assert [r.sensor for r in a.readings_from(iq, 250_000.0)] == ["93"]
|
|
|
|
|
|
def test_a_burst_no_longer_than_a_message_is_left_whole():
|
|
burst = a.Burst(marks=(400.0,) * 40, spaces=(220.0,) * 39)
|
|
assert a._divide([burst]) == [burst]
|
|
|
|
|
|
def test_a_capture_that_is_mostly_burst_is_not_thrown_away_as_empty():
|
|
"""The noise is read off the bottom of a block, never off the middle.
|
|
|
|
Taken from the median, a block that is largely signal measures its own
|
|
burst against its own burst, finds no difference, and is discarded as
|
|
silence. Which is what happened to any short capture: the shorter it
|
|
was, the larger the share of it that was the thing being looked for.
|
|
"""
|
|
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
|
|
# Short gaps, so the carrier is on for most of the recording: this is a
|
|
# real timing, and it is the one that showed the fault.
|
|
iq = keyed_at(bits, pwm_pairs(bits, gap=150.0), copies=1,
|
|
reset_us=1_000.0, lead_us=500.0)
|
|
envelope, rate = a.baseband(iq, 250_000.0, 0.0)
|
|
smooth = a._smoothed(envelope, rate)
|
|
# More than half of it is the sensor talking, which is what puts the
|
|
# median inside the signal and the block in danger of being discarded.
|
|
assert (smooth > smooth.max() / 2).mean() > 0.5
|
|
assert np.median(smooth) * 1.8 > np.percentile(smooth, 99.99)
|
|
assert [r.sensor for r in a.readings_from(iq, 250_000.0)] == ["1A2B"]
|
|
|
|
|
|
def test_a_bit_drawn_as_a_much_longer_gap_does_not_end_the_burst():
|
|
"""Where a one is three times a zero and most bits are zeroes, the middle
|
|
gap is the short one -- so a rule keyed on the middle of the gaps cuts
|
|
the message at every one-bit and hands back pieces of three pulses."""
|
|
bits = a.frame_609(0x5C, 4.2, 80)
|
|
iq = keyed_at(bits, ppm_pairs(bits, short=800.0, long=2_400.0),
|
|
reset_us=10_000.0)
|
|
envelope, rate = a.baseband(iq, 250_000.0, 0.0)
|
|
found = a.bursts(envelope, rate)
|
|
assert found and max(b.pulses for b in found) >= 40
|
|
assert [r.sensor for r in a.readings_from(iq, 250_000.0)] == ["5C"]
|
|
|
|
|
|
@pytest.mark.parametrize("one,zero,gap", [
|
|
(408.0, 220.0, None), (408.0, 220.0, 200.0), (500.0, 250.0, 250.0),
|
|
(300.0, 150.0, 150.0), (600.0, 300.0, 300.0), (400.0, 200.0, 400.0),
|
|
])
|
|
def test_the_envelope_of_pulse_widths_it_will_read(one, zero, gap):
|
|
bits = a.tower_frame(0x1A2B, 21.5, 48, "A")
|
|
iq = keyed_at(bits, pwm_pairs(bits, one=one, zero=zero, gap=gap))
|
|
assert [r.sensor for r in a.readings_from(iq, 250_000.0)] == ["1A2B"]
|
|
|
|
|
|
@pytest.mark.parametrize("duty", [0.05, 0.3, 0.5, 0.62, 0.75, 0.85])
|
|
def test_the_gate_sits_between_the_quiet_and_the_loudest_at_any_duty(duty):
|
|
"""Two ways to get this wrong, and both report a silence that was not.
|
|
|
|
The scatter it is built from has to be measured inside the quiet part.
|
|
Measured across the edge between off and on -- which is where a
|
|
percentile lands once the carrier is on for much of the block -- it
|
|
measures the signal, and six times the signal is a gate above everything
|
|
in the block.
|
|
|
|
And whatever the arithmetic comes to, a gate above the loudest thing
|
|
there is cannot be right: it is the one setting that guarantees nothing
|
|
is found on a second that plainly had something in it.
|
|
"""
|
|
rng = np.random.default_rng(4)
|
|
n = 60_000
|
|
envelope = np.abs(rng.standard_normal(n)
|
|
+ 1j * rng.standard_normal(n)).astype(np.float32) * 0.02
|
|
# The carrier keyed on for `duty` of the block, in pulses rather than
|
|
# one lump, which is what a message looks like.
|
|
period = 400
|
|
for start in range(0, n - period, period):
|
|
envelope[start:start + int(period * duty)] += 1.0
|
|
smooth = a._smoothed(envelope, 250_000.0)
|
|
quiet = float(np.percentile(smooth, 20))
|
|
peak = float(np.percentile(smooth, 99.99))
|
|
gate = a._noise_gate(smooth, quiet, peak)
|
|
assert gate < peak, f"duty {duty}: gate {gate:.3f} is above the peak {peak:.3f}"
|
|
assert gate > quiet, f"duty {duty}: gate {gate:.3f} is at or below the noise"
|
|
|
|
|
|
def test_the_gate_still_keeps_noise_out_when_there_is_no_signal():
|
|
"""The other half of it: a quiet band must find nothing at all."""
|
|
rng = np.random.default_rng(11)
|
|
for _ in range(8):
|
|
noise = np.abs(rng.standard_normal(60_000)
|
|
+ 1j * rng.standard_normal(60_000)).astype(np.float32)
|
|
smooth = a._smoothed(noise, 250_000.0)
|
|
quiet = float(np.percentile(smooth, 20))
|
|
peak = float(np.percentile(smooth, 99.99))
|
|
over = (smooth > a._noise_gate(smooth, quiet, peak)).mean()
|
|
assert over < 0.01, f"{over:.1%} of pure noise cleared the gate"
|