Cleaning reminders

src/gaggibot/cli.py

@@ -170,12 +170,20 @@ async def pump_events():
                         log.exception("event handling failed")
 
             async def pump_shots():
+                import re as _re
+
                 async def tee(source):
                     async for frame in source:
                         cache_frame(frame)
                         await router.on_frame(frame)
                         yield frame
 
+                async def on_utility(profile):
+                    # only actual cleaning runs reset the counter, not water flushes
+                    if _re.search(r"(?i)backflush|descale|clean", profile or ""):
+                        state.set("shots_since_clean", 0)
+                        log.info("cleaning run detected (%s); counter reset", profile)
+
                 frames = tee(
                     replay_frames(replay) if replay else client.status_stream()
                 )
@@ -184,8 +192,19 @@ async def tee(source):
                     min_duration_s=config.min_shot_s,
                     ignore_profiles=config.ignore_profiles,
                     last_known_id=state.get("last_shot_id", -1),
+                    on_utility=on_utility,
                 )
                 async for shot in watcher.shots(frames):
+                    since_clean = state.get("shots_since_clean", 0) + 1
+                    state.set("shots_since_clean", since_clean)
+                    if config.clean_every and since_clean % config.clean_every == 0:
+                        try:
+                            await messenger.send(
+                                f"🧽 {since_clean} espresso shots since the last backflush — "
+                                "the group head would appreciate a round with the blind basket."
+                            )
+                        except Exception:  # noqa: BLE001
+                            log.exception("cleaning reminder failed")
                     state.update(
                         last_shot_id=shot.entry.id,
                         last_shot={

src/gaggibot/commands.py

@@ -93,6 +93,9 @@ async def _cmd_status(self) -> None:
             line += f" → target {tt:.0f}°C"
         if frame.get("wl") is not None:
             line += f" · water {frame['wl']}%"
+        since_clean = self.state.get("shots_since_clean", 0)
+        if since_clean:
+            line += f"\n🧽 {since_clean} shots since the last backflush"
         await self.messenger.send(line)
 
     async def _cmd_last(self) -> None:

src/gaggibot/config.py

@@ -28,6 +28,7 @@
     "journal_url": "GAGGIBOT_JOURNAL_URL",
     "hints_enabled": "GAGGIBOT_HINTS",
     "digest_enabled": "GAGGIBOT_DIGEST",
+    "clean_every": "GAGGIBOT_CLEAN_EVERY",
 }
 
 
@@ -53,6 +54,7 @@ class Config:
     journal_url: str = ""  # public journal base URL, used for /last deep links
     hints_enabled: bool = True  # dial-in suggestions after sour/bitter/low-rated shots
     digest_enabled: bool = True  # weekly summary on Sunday evening
+    clean_every: int = 40  # backflush reminder every N espresso shots; 0 = off
 
     @classmethod
     def load(cls, path: str | Path | None = None) -> Config:
@@ -70,6 +72,8 @@ def load(cls, path: str | Path | None = None) -> Config:
                 continue
             if f.name == "min_shot_s":
                 value = float(value)
+            elif f.name == "clean_every":
+                value = int(value)
             elif f.name in ("sync_enabled", "hints_enabled", "digest_enabled"):
                 value = str(value).lower() in ("1", "true", "yes", "on")
             kwargs[f.name] = value

src/gaggibot/watcher.py

@@ -40,11 +40,13 @@ def __init__(
         min_duration_s: float = 10.0,
         ignore_profiles: str = r"(?i)backflush|descale|flush|clean",
         last_known_id: int = -1,
+        on_utility=None,
     ) -> None:
         self.client = client
         self.min_duration_ms = min_duration_s * 1000
         self.ignore_re = re.compile(ignore_profiles) if ignore_profiles else None
         self.last_known_id = last_known_id
+        self.on_utility = on_utility  # async callback(profile) for ignored utility runs
 
     def _frame_is_shot_end(self, prev: dict | None, frame: dict) -> bool:
         if prev is None:
@@ -107,6 +109,13 @@ async def shots(self, frames: AsyncIterator[dict]) -> AsyncIterator[FinishedShot
                 profile = prev.get("p", "")
                 duration = (prev.get("process") or {}).get("e", 0)
                 log.info("shot ended: profile=%r duration=%.1fs", profile, duration / 1000)
+                is_utility = (
+                    duration >= self.min_duration_ms
+                    and self.ignore_re
+                    and self.ignore_re.search(profile or "")
+                )
+                if is_utility and self.on_utility:
+                    await self.on_utility(profile)
                 if self._accept(profile, duration):
                     entry = await self._resolve_new_entry()
                     if entry is None:

tests/test_cleaning.py

@@ -0,0 +1,22 @@
+import pathlib
+
+import pytest
+
+from gaggibot.watcher import ShotWatcher, replay_frames
+from tests.test_watcher import FakeClient
+
+FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "status_frames.jsonl"
+
+
+@pytest.mark.asyncio
+async def test_on_utility_fires_for_backflush_only():
+    seen = []
+
+    async def on_utility(profile):
+        seen.append(profile)
+
+    watcher = ShotWatcher(FakeClient(), min_duration_s=10, on_utility=on_utility)
+    _ = [s async for s in watcher.shots(replay_frames(FIXTURE))]
+    # fixture contains one 40s backflush (fires) and one 5s short brew (too
+    # short, never counts as a utility run)
+    assert seen == ["[Utility] Backflush"]