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]>
56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
"""Core data types passed between probe sources and grill outputs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ProbeReading:
|
|
"""One sample from one probe channel.
|
|
|
|
Every source normalises to this shape, so the router and the PT1000
|
|
emulator never need to know what hardware produced the number.
|
|
|
|
Temperatures are always Celsius -- convert at the display edge, not here.
|
|
"""
|
|
|
|
probe_id: str
|
|
"""Stable unique id, e.g. a BLE MAC or serial. Used for config mapping."""
|
|
|
|
tip_c: float
|
|
"""Internal / food temperature."""
|
|
|
|
source: str
|
|
"""Driver name that produced this, e.g. "meater"."""
|
|
|
|
ambient_c: float | None = None
|
|
"""Ambient / surface temperature, if the probe reports one."""
|
|
|
|
battery_pct: int | None = None
|
|
rssi: int | None = None
|
|
display_name: str | None = None
|
|
timestamp: float = field(default_factory=time.monotonic)
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.display_name or self.probe_id
|
|
|
|
def age_s(self, now: float | None = None) -> float:
|
|
return (time.monotonic() if now is None else now) - self.timestamp
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ChannelState:
|
|
"""What we are currently driving onto one grill probe jack."""
|
|
|
|
jack: int
|
|
target_temp_c: float | None = None
|
|
applied_ohms: float | None = None
|
|
reading: ProbeReading | None = None
|
|
stale: bool = True
|
|
error: str | None = None
|