steveandClaude Opus 5
ba0c3b58b6
Feeds third-party BBQ probes into a pellet grill's wired probe jacks by presenting the resistance a real PT1000 RTD would show at that temperature. The grill sees an ordinary wired probe, so its display, app, target-temp alarms and Keep Warm all work with no protocol reversing involved. Pluggable probe sources: ThermoWorks RFX (via ThermoWorks Cloud), MEATER (community-derived BLE decode), a synthetic simulator for hardware-free development, and a Combustion stub. Two safety invariants are load-bearing: - Stale, missing or implausible readings drive the channel cold, never hot. A stuck-hot channel could convince the grill a cook finished early. - Unimplemented sources raise rather than returning plausible numbers, since the grill acts on these values. Cloud sources are dated by the cloud's own timestamp rather than by fetch time, because ThermoWorks serves a dead probe's last value with a fresh 200. Co-Authored-By: Claude Opus 5 <[email protected]>
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
"""Emulator behaviour, especially the failure modes.
|
|
|
|
The fail-cold policy is a safety property, not a nicety: a stuck-hot channel
|
|
can convince the grill a cook finished early. These tests pin that down.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from smokescreen.models import ProbeReading
|
|
from smokescreen.outputs.pt1000_emulator import (
|
|
ChannelCalibration,
|
|
PT1000Emulator,
|
|
)
|
|
from smokescreen.outputs.rheostat import MockRheostat
|
|
from smokescreen.pt1000 import temp_c_to_resistance
|
|
|
|
|
|
def make_emulator(**kwargs) -> tuple[PT1000Emulator, MockRheostat]:
|
|
channel = MockRheostat(max_code=1023)
|
|
emulator = PT1000Emulator(
|
|
jack=1,
|
|
channel=channel,
|
|
calibration=ChannelCalibration(r_offset_ohms=1000.0, step_ohms=1.0),
|
|
**kwargs,
|
|
)
|
|
return emulator, channel
|
|
|
|
|
|
def reading(tip_c: float, *, age_s: float = 0.0) -> ProbeReading:
|
|
return ProbeReading(
|
|
probe_id="test:1",
|
|
tip_c=tip_c,
|
|
source="test",
|
|
timestamp=time.monotonic() - age_s,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("temp_c", [0.0, 25.0, 60.0, 93.3, 150.0])
|
|
def test_applied_temperature_tracks_request(temp_c):
|
|
emulator, _ = make_emulator()
|
|
assert emulator.apply_temperature(temp_c) == pytest.approx(temp_c, abs=0.2)
|
|
|
|
|
|
def test_wiper_code_matches_the_pt1000_curve():
|
|
emulator, channel = make_emulator()
|
|
emulator.apply_temperature(100.0)
|
|
# r_offset 1000 + code*1 should land on 1385 ohm.
|
|
expected_code = round(temp_c_to_resistance(100.0) - 1000.0)
|
|
assert channel.code == expected_code
|
|
|
|
|
|
def test_temperature_above_hardware_range_clamps_rather_than_wraps():
|
|
emulator, channel = make_emulator()
|
|
applied = emulator.apply_temperature(300.0)
|
|
assert channel.code == channel.max_code
|
|
assert applied < 300.0 # clamped, and the caller can see it
|
|
|
|
|
|
def test_stale_reading_fails_cold():
|
|
emulator, channel = make_emulator(stale_after_s=30.0)
|
|
emulator.update(reading(90.0))
|
|
assert channel.code > 0
|
|
|
|
emulator.update(reading(90.0, age_s=120.0))
|
|
assert channel.code == 0
|
|
assert emulator.state.stale
|
|
assert "stale" in emulator.state.error
|
|
|
|
|
|
def test_missing_probe_fails_cold():
|
|
emulator, channel = make_emulator()
|
|
emulator.update(None)
|
|
assert channel.code == 0
|
|
assert emulator.state.stale
|
|
|
|
|
|
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), 5000.0])
|
|
def test_implausible_reading_fails_cold_instead_of_raising(bad):
|
|
emulator, channel = make_emulator()
|
|
emulator.update(reading(bad))
|
|
assert channel.code == 0
|
|
assert emulator.state.stale
|
|
|
|
|
|
def test_fail_cold_is_never_hot():
|
|
"""Whatever else happens, the fault state must read cold, not hot."""
|
|
emulator, channel = make_emulator()
|
|
emulator.apply_temperature(150.0)
|
|
emulator.fail_cold("test")
|
|
assert channel.code == 0
|
|
assert emulator.min_ohms < temp_c_to_resistance(0.0) + 1e-6
|
|
|
|
|
|
def test_close_drives_cold_before_releasing():
|
|
emulator, channel = make_emulator()
|
|
emulator.apply_temperature(120.0)
|
|
emulator.close()
|
|
assert channel.history[-1] == 0
|
|
|
|
|
|
def test_calibration_solved_from_two_measured_points():
|
|
cal = ChannelCalibration.from_two_points(
|
|
code_a=0, ohms_a=1012.4, code_b=1000, ohms_b=2015.6
|
|
)
|
|
assert cal.r_offset_ohms == pytest.approx(1012.4, abs=0.01)
|
|
assert cal.step_ohms == pytest.approx(1.0032, abs=0.0001)
|
|
assert cal.ohms_for_code(500) == pytest.approx(1514.0, abs=0.1)
|
|
|
|
|
|
def test_calibration_rejects_degenerate_points():
|
|
with pytest.raises(ValueError):
|
|
ChannelCalibration.from_two_points(10, 1000.0, 10, 2000.0)
|