feat: Wrap Coach -- rule-based hold/wrap advice (--coach)

README.md

@@ -254,6 +254,24 @@ Syntax is `[PROBE:]NAME` (bare name → probe 1, e.g. `--probe-name 2:brisket` f
 probe 2). Set once and it's remembered across `--watch` runs (saved to a gitignored
 `.probe_names.json`).
 
+### Wrap Coach — should you wrap now?
+
+Rule-based, mid-cook advice on whether to hold for more bark or wrap to push through —
+no ML, no network, just the same recent-window forecast and stall band already driving
+the ETA:
+
+```bash
+./venv/bin/python poll.py --watch 30 --coach --stage 165:wrap --stage 203:done
+```
+
+```text
+  🟡 the pork butt: 47 minutes into the stall -- normal, bark's still setting. Hold for more bark, or wrap now to push through faster.
+```
+
+It escalates ("🔴 urgent") past ~90 minutes in a single continuous stall, and knows
+the difference between "still stalled, not wrapped yet" and "wrapped and *still*
+stalled" (the latter suggests bumping the grill temp instead of wrapping again).
+
 ---
 
 ## 🔐 Credentials

poll.py

@@ -6,6 +6,7 @@
   ./venv/bin/python poll.py            # one reading, printed + logged
   ./venv/bin/python poll.py --watch 30 # log every 30s until Ctrl-C
   ./venv/bin/python poll.py --watch 30 --speak  # + a spoken update every tick
+  ./venv/bin/python poll.py --watch 30 --coach  # + rule-based hold/wrap advice each tick
   ./venv/bin/python poll.py --watch 30 --speak --stage 165:wrap --stage 205:done \
       --chart cook.html    # set it and forget it: logs, speaks, and keeps a
                             # self-refreshing chart at cook.html up to date --
@@ -25,6 +26,7 @@
 
 import plan
 import probe_names
+import wrap_coach
 from alarms import notify_remote
 from forecast import describe, describe_stages, forecast, forecast_stages
 from traeger_client import Traeger, TraegerError, parse_status
@@ -481,6 +483,32 @@ def print_forecasts(row, stages=None):
                 print(f"  ⏱  P{i} {describe(fc, target, now=now)}")
 
 
+_COACH_ICONS = {"info": "💡", "suggest": "🟡", "urgent": "🔴"}
+
+
+def print_coach(row, stages=None, names=None):
+    """Opt-in (--coach) rule-based advice line per connected probe: hold or
+    wrap, using the same recent-window forecast/stall logic that drives the
+    printed ETA -- see wrap_coach.py. Reuses _eta_samples (print_forecasts
+    must have run first this tick, same requirement as speech_for_probes)."""
+    stages = stages or {}
+    for i in range(1, MAX_PROBES + 1):
+        temp = row.get(f"probe{i}_temp")
+        if not row.get(f"probe{i}_connected") or temp is None:
+            continue
+        samples = _eta_samples.get(i, [])
+        if len(samples) < 2:
+            continue
+        t0 = samples[0][0]
+        mins = [(t - t0).total_seconds() / 60 for t, _ in samples]
+        temps = [v for _, v in samples]
+        tgt = row.get(f"probe{i}_set")
+        rec = wrap_coach.recommend_for_probe(mins, temps, stages.get(i), float(tgt) if tgt else None)
+        label = probe_names.label(i, names)
+        icon = _COACH_ICONS.get(rec["urgency"], "💡")
+        print(f"  {icon} {label}: {rec['advice']}")
+
+
 _STAGE_ACTIONS = {"wrap": "WRAP IT", "done": "DONE — rest it", "rest": "RESTING done"}
 
 
@@ -527,7 +555,8 @@ def check_alarms(row, alarms, names=None):
                 print(f"  🔔 ALARM: probe {probe} crossed {int(thr)}°F")
 
 
-def one_shot(t, alarms=None, stages=None, speak_every_tick=False, chart_path=None, chart_probe=1, names=None):
+def one_shot(t, alarms=None, stages=None, speak_every_tick=False, chart_path=None, chart_probe=1,
+             names=None, coach=False):
     alarms = alarms or {}
     stages = stages or {}
     status = t.poll()
@@ -549,6 +578,8 @@ def one_shot(t, alarms=None, stages=None, speak_every_tick=False, chart_path=Non
         probes_txt = "  ".join(parts) if parts else "no probes"
         print(f"[{row['ts']}] grill {row['grill']}° (set {row['set']}°)  {probes_txt}  [{state}]")
         print_forecasts(row, stages)  # live "when's it done" prediction (stage-aware)
+        if coach:
+            print_coach(row, stages, names)  # opt-in hold/wrap advice
         if speak_every_tick:
             text = speech_for_probes(row, stages, names)  # full sentence, incl. "Update N" + grill temp
             if text:
@@ -723,9 +754,16 @@ def _add_alarm(spec):
     if chart_path:
         print(f"Writing a live chart to {chart_path} every tick -- open it once and leave it.")
 
+    # Wrap Coach: --coach or PELLET_PILOT_COACH=1 -- rule-based hold/wrap advice
+    # each tick, printed alongside the forecast line. Off by default.
+    coach = "--coach" in sys.argv or \
+        os.environ.get("PELLET_PILOT_COACH", "").lower() in ("1", "true", "yes")
+    if coach:
+        print("Wrap Coach enabled (--coach).")
+
     if interval is None:
         one_shot(t, alarms, stages, speak_every_tick,
-                 chart_path=chart_path, chart_probe=chart_probe, names=names)
+                 chart_path=chart_path, chart_probe=chart_probe, names=names, coach=coach)
         return
 
     print(f"Watching every {interval}s. Ctrl-C to stop.")
@@ -734,7 +772,7 @@ def _add_alarm(spec):
         while True:
             try:
                 one_shot(t, alarms, stages, speak_every_tick,
-                         chart_path=chart_path, chart_probe=chart_probe, names=names)
+                         chart_path=chart_path, chart_probe=chart_probe, names=names, coach=coach)
                 consecutive_failures = 0
                 time.sleep(interval)
             except Exception as e:

pyproject.toml

@@ -35,6 +35,7 @@ py-modules = [
     "export",
     "alarms",
     "traeger_client",
+    "wrap_coach",
 ]
 
 [tool.setuptools.package-data]

tests/smoke.py

@@ -8,7 +8,9 @@
 runtime. No network required (SSRF checks use IP literals; auth flows use
 mocked HTTP responses).
 """
+import contextlib
 import datetime as dt
+import io
 import json
 import os
 import re
@@ -35,6 +37,7 @@
 import report  # noqa: E402
 import traeger_client as tc  # noqa: E402
 import trend  # noqa: E402,F401
+import wrap_coach  # noqa: E402
 
 
 def _doc(temps=((150, 203),), grill=225):
@@ -928,6 +931,83 @@ def test_stall_minutes():
     assert report._stall_minutes(climbing, 1) == 0, "a genuine climb shouldn't count as stalled"
 
 
+def test_wrap_coach_current_stall_streak_not_total():
+    # A climb that only later flattens should report the CURRENT streak's
+    # length, not the whole cook's elapsed time.
+    mins = list(range(0, 40, 2))              # 20 readings, 2 min apart
+    temps = [130.0 + 2 * i for i in range(10)] + [160.0] * 10  # climb, then flat at 160 (in-band)
+    stall_min = wrap_coach.current_stall_minutes(mins, temps)
+    assert 0 < stall_min <= 20, stall_min      # only the trailing ~18 min flat stretch, not 38
+
+
+def test_wrap_coach_recommend_categories():
+    mins = list(range(20))
+    assert wrap_coach.recommend(mins[:1], [140.0])["status"] == "insufficient"
+
+    # rate 1 deg/min, still 46 min out at index 19 (159 -> target 205) -> no rush
+    climbing = [140.0 + i for i in range(20)]
+    rec = wrap_coach.recommend(mins, climbing, target=205)
+    assert rec["status"] == "on_track" and rec["urgency"] == "info", rec
+
+    # rate 1 deg/min, only 1 min out (204 -> target 205) -> time to plan the rest
+    near_done = [185.0 + i for i in range(20)]
+    rec = wrap_coach.recommend(mins, near_done, target=205)
+    assert rec["status"] == "on_track" and rec["urgency"] == "suggest", rec
+    assert "rest time" in rec["advice"], rec
+
+    done = wrap_coach.recommend(mins, [200.0 + i for i in range(20)], target=205)
+    assert done["status"] == "done" and "pull it" in done["advice"], done
+
+    flat_170 = wrap_coach.recommend(mins, [170.0] * 20, target=205)
+    assert flat_170["status"] == "stalled" and flat_170["urgency"] == "suggest", flat_170
+
+    flat_100 = wrap_coach.recommend(mins, [100.0] * 20, target=205)
+    assert flat_100["status"] == "not_rising" and flat_100["urgency"] == "suggest", flat_100
+
+
+def test_wrap_coach_long_stall_and_wrapped_escalation():
+    long_stall_mins = list(range(0, 200, 2))
+    long_stall_temps = [160.0] * len(long_stall_mins)
+
+    unwrapped = wrap_coach.recommend(long_stall_mins, long_stall_temps, target=205, wrapped=False)
+    assert unwrapped["urgency"] == "urgent" and "Wrap now" in unwrapped["advice"], unwrapped
+
+    wrapped_long = wrap_coach.recommend(long_stall_mins, long_stall_temps, target=205, wrapped=True)
+    assert wrapped_long["urgency"] == "urgent" and "bumping the grill temp" in wrapped_long["advice"], wrapped_long
+
+    short_stall_mins = list(range(0, 20, 2))
+    short_stall_temps = [160.0] * len(short_stall_mins)
+    wrapped_short = wrap_coach.recommend(short_stall_mins, short_stall_temps, target=205, wrapped=True)
+    assert wrapped_short["urgency"] == "info", wrapped_short
+
+
+def test_wrap_coach_recommend_for_probe_auto_detects_wrapped():
+    # Past the "wrap" stage temp and stalled -> auto-detected as wrapped,
+    # so advice should NOT tell you to wrap something you already wrapped.
+    mins = list(range(0, 200, 2))
+    temps = [170.0] * len(mins)  # past the 165 wrap stage, stalled
+    stages_for_probe = [(165.0, "wrap"), (205.0, "done")]
+    rec = wrap_coach.recommend_for_probe(mins, temps, stages_for_probe)
+    assert "already wrapped" in rec["advice"].lower(), rec
+
+    not_yet_wrapped_temps = [160.0] * len(mins)  # below 165 -> not wrapped yet
+    rec2 = wrap_coach.recommend_for_probe(mins, not_yet_wrapped_temps, stages_for_probe)
+    assert "wrap now" in rec2["advice"].lower(), rec2
+
+
+def test_print_coach_smoke():
+    poll._eta_samples.clear()
+    t0 = dt.datetime(2026, 7, 4, 12, 0, 0)
+    t1 = t0 + dt.timedelta(minutes=10)
+    poll._eta_samples[1] = [(t0, 150.0), (t1, 160.0)]
+    row = {"ts": t1.isoformat(), "probe1_temp": 160, "probe1_connected": True, "probe1_set": 205}
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf):
+        poll.print_coach(row, {}, {1: "pork butt"})
+    out = buf.getvalue()
+    assert "pork butt" in out, out
+
+
 def test_backoff_seconds():
     # RT-4: exponential backoff, capped, so a persistent re-auth failure doesn't
     # hammer Cognito every `interval` seconds forever.

wrap_coach.py

@@ -0,0 +1,111 @@
+"""Wrap Coach -- rule-based, mid-cook advice on whether to hold or wrap.
+
+No ML, no network: a handful of readable thresholds layered on the SAME
+recent-window forecast (forecast.py) and stall band the rest of Pellet Pilot
+already uses for its ETA, so the advice never contradicts what the printed
+prediction is already telling you.
+"""
+from forecast import _FLAT, STALL_HI, STALL_LO, forecast
+
+# How long the CURRENT, ongoing stall has to run before advice escalates.
+_LONG_STALL_MIN = 90.0
+# How close to target before it's time to start thinking about the rest.
+_NEAR_DONE_MIN = 30.0
+
+
+def current_stall_minutes(times_min, temps):
+    """Minutes of the current, ongoing stall streak, walking backward from
+    the most recent sample -- NOT total stall time across the whole cook.
+    Zero as soon as the most recent step leaves the stall band or the rate
+    stops being flat (same threshold forecast.py uses to call it "stalled").
+    """
+    total = 0.0
+    for i in range(len(temps) - 1, 0, -1):
+        v0, v1 = temps[i - 1], temps[i]
+        t0, t1 = times_min[i - 1], times_min[i]
+        if v0 is None or v1 is None:
+            break
+        if not (STALL_LO <= v0 <= STALL_HI and STALL_LO <= v1 <= STALL_HI):
+            break
+        elapsed = t1 - t0
+        if elapsed <= 0:
+            break
+        if (v1 - v0) / elapsed >= _FLAT:
+            break
+        total += elapsed
+    return total
+
+
+def recommend(times_min, temps, target=None, wrapped=False):
+    """One coaching line + an urgency ('info'|'suggest'|'urgent').
+
+    Mirrors forecast()'s own status so advice never disagrees with the
+    printed ETA line for the same tick.
+    """
+    if len(temps) < 2:
+        return {"advice": "Not enough data yet to coach this cook.",
+                "urgency": "info", "status": "insufficient"}
+
+    fc = forecast(times_min, temps, target)
+    status = fc["status"]
+
+    if status in ("insufficient", "no_target"):
+        return {"advice": "Not enough data yet to coach this cook.",
+                "urgency": "info", "status": status}
+
+    if status == "done":
+        return {"advice": "At target -- pull it and rest at least 30-60 min before slicing.",
+                "urgency": "info", "status": status}
+
+    if status == "not_rising":
+        return {"advice": "Not climbing, and it's outside the usual 150-175 stall range -- "
+                           "worth checking the fire, pellets, or lid seal.",
+                "urgency": "suggest", "status": status}
+
+    if status == "stalled":
+        stall_min = current_stall_minutes(times_min, temps)
+        if wrapped:
+            if stall_min >= _LONG_STALL_MIN:
+                return {"advice": f"Already wrapped and still stalled after {stall_min:.0f} min -- "
+                                   "consider bumping the grill temp 10-15 degrees.",
+                        "urgency": "urgent", "status": status}
+            return {"advice": f"Wrapped and holding at the stall ({stall_min:.0f} min so far) -- "
+                               "normal, give it more time.",
+                    "urgency": "info", "status": status}
+        if stall_min >= _LONG_STALL_MIN:
+            return {"advice": f"{stall_min:.0f} minutes into the stall -- that's a long one. "
+                               "Wrap now to push through, unless you're deliberately going low and slow.",
+                    "urgency": "urgent", "status": status}
+        return {"advice": f"{stall_min:.0f} minutes into the stall -- normal, bark's still setting. "
+                           "Hold for more bark, or wrap now to push through faster.",
+                "urgency": "suggest", "status": status}
+
+    # on_track -- rate/eta_min are always real numbers here
+    eta_min = fc["eta_min"]
+    if eta_min is not None and eta_min <= _NEAR_DONE_MIN:
+        return {"advice": f"Getting close (~{eta_min:.0f} min out) -- start planning your rest time.",
+                "urgency": "suggest", "status": status}
+    return {"advice": f"Climbing steadily at {fc['rate']:.1f} deg/min -- no action needed.",
+            "urgency": "info", "status": status}
+
+
+def _wrap_stage_temp(stages_for_probe):
+    for temp, label in stages_for_probe or []:
+        if "wrap" in label.lower():
+            return temp
+    return None
+
+
+def recommend_for_probe(times_min, temps, stages_for_probe=None, target=None):
+    """High-level entry point: auto-detects wrap status from a stage plan
+    (a 'wrap' stage that's already been crossed) if one is given, projecting
+    to the final stage temp; otherwise falls back to a plain probe target
+    with wrapped=False.
+    """
+    cur = temps[-1] if temps else None
+    wrap_temp = _wrap_stage_temp(stages_for_probe)
+    wrapped = bool(wrap_temp is not None and cur is not None and cur >= wrap_temp)
+    if stages_for_probe:
+        final_temp = max(t for t, _ in stages_for_probe)
+        return recommend(times_min, temps, final_temp, wrapped=wrapped)
+    return recommend(times_min, temps, target, wrapped=False)