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]>
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from smokescreen.models import ProbeReading
|
|
from smokescreen.router import ProbeRouter
|
|
|
|
|
|
def reading(probe_id: str, tip_c: float = 60.0) -> ProbeReading:
|
|
return ProbeReading(probe_id=probe_id, tip_c=tip_c, source="test")
|
|
|
|
|
|
def test_pinned_probe_drives_its_jack():
|
|
router = ProbeRouter(jacks=[1, 2], pinned={2: "meater:aa"})
|
|
assert router.observe(reading("meater:aa")) == 2
|
|
|
|
|
|
def test_auto_assignment_fills_free_jacks_in_order():
|
|
router = ProbeRouter(jacks=[1, 2])
|
|
assert router.observe(reading("probe:a")) == 1
|
|
assert router.observe(reading("probe:b")) == 2
|
|
|
|
|
|
def test_auto_assignment_skips_pinned_jacks():
|
|
router = ProbeRouter(jacks=[1, 2], pinned={1: "meater:aa"})
|
|
assert router.observe(reading("probe:b")) == 2
|
|
|
|
|
|
def test_binding_is_sticky_across_a_dropout():
|
|
"""A probe that reconnects must not land on a different jack mid-cook."""
|
|
router = ProbeRouter(jacks=[1, 2])
|
|
assert router.observe(reading("probe:a")) == 1
|
|
assert router.observe(reading("probe:b")) == 2
|
|
assert router.observe(reading("probe:a", 70.0)) == 1
|
|
|
|
|
|
def test_extra_probes_have_nowhere_to_go():
|
|
router = ProbeRouter(jacks=[1, 2])
|
|
router.observe(reading("probe:a"))
|
|
router.observe(reading("probe:b"))
|
|
assert router.observe(reading("probe:c")) is None
|
|
assert [r.probe_id for r in router.unassigned()] == ["probe:c"]
|
|
|
|
|
|
def test_reading_for_returns_latest_sample():
|
|
router = ProbeRouter(jacks=[1])
|
|
router.observe(reading("probe:a", 50.0))
|
|
router.observe(reading("probe:a", 75.0))
|
|
assert router.reading_for(1).tip_c == 75.0
|
|
|
|
|
|
def test_reading_for_unbound_jack_is_none():
|
|
assert ProbeRouter(jacks=[1, 2]).reading_for(2) is None
|
|
|
|
|
|
def test_pin_to_unknown_jack_is_rejected():
|
|
with pytest.raises(ValueError, match="unknown jacks"):
|
|
ProbeRouter(jacks=[1, 2], pinned={3: "probe:a"})
|
|
|
|
|
|
def test_probe_pinned_twice_is_rejected():
|
|
with pytest.raises(ValueError, match="more than one jack"):
|
|
ProbeRouter(jacks=[1, 2], pinned={1: "probe:a", 2: "probe:a"})
|