"""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