The 60-Second Dwell Gate: Debouncing YouTube Live Viewer Alerts with Sliding-Window Filters

📡 YouTube Watch: LIVE=yes | video=JYa5jrSi4FU | viewers=2 (alert>=3) | chat_items=0 | daemon=RUNNING | title='When t he Sheet Music Jams the Player Piano: Disarming Backtick Cues, Right-Split Unpacking, and...'
🔭 Watching https://www.youtube.com/@MikeLevin/live every 30s...
💡 [HOW TO EXIT TMUX WITHOUT STOPPING]: Press 'q' OR press Ctrl+b, release, then press d
👀 [VIEWER] 3 watching now (min=3) — When the Sheet Music Jams the Player Piano: Disarming Backtick Cues, Right-Split Unpacking, and...
🔔 Sent ntfy push (HTTP 200 -> pipulate-hb-e12...): [👀 Honeybot Live Viewer] 3 watching now (min=3) — When the Sheet Music Jams the Player Piano: Disarming Backtick Cues, Right-Split Unpacking, and...

Raw Viewers:   2 ──► 3 ──► 2          2 ──► 3 ──► 3 ──► 3
                     │     │                │     │     │
Dwell Clock:        0s   reset             0s   30s   60s >= DWELL
                     │     │                            │
Output:           [⏳ hold] [💨 suppressed]          [👀 ALERT + 🔔 ntfy]

.venv/bin/python scripts/articles/common.py --broadcast watch --dry-run
.venv/bin/python -c "import sys; sys.path.insert(0, 'scripts/articles'); import common; fn = getattr(common, '_sustained_viewer_count', None); print('dwell=', getattr(common, 'YT_VIEWER_ALERT_DWELL', None), 'spike_30s=', fn([(0.0, 3), (30.0, 3)], 30.0, 60) if fn else None, 'sustained_60s=', fn([(0.0, 3), (30.0, 3), (61.0, 3)], 61.0, 60) if fn else None)"

context.txt
scripts/articles/common.py
! .venv/bin/python scripts/articles/common.py --broadcast watch --dry-run
! .venv/bin/python -c "import sys; sys.path.insert(0, 'scripts/articles'); import common; fn = getattr(common, '_sustained_viewer_count', None); print('dwell=', getattr(common, 'YT_VIEWER_ALERT_DWELL', None), 'spike_30s=', fn([(0.0, 3), (30.0, 3)], 30.0, 60) if fn else None, 'sustained_60s=', fn([(0.0, 3), (30.0, 3), (61.0, 3)], 61.0, 60) if fn else None)"

Target: scripts/articles/common.py
[[[SEARCH]]]
YT_VIEWER_ALERT_MIN = int(os.environ.get("PIPULATE_YT_VIEWER_MIN", "3"))
YT_VIEWER_ALERT_COOLDOWN = int(os.environ.get("PIPULATE_YT_VIEWER_COOLDOWN", "300"))  # Seconds between any two viewer alerts
[[[DIVIDER]]]
YT_VIEWER_ALERT_MIN = int(os.environ.get("PIPULATE_YT_VIEWER_MIN", "3"))
YT_VIEWER_ALERT_DWELL = int(os.environ.get("PIPULATE_YT_VIEWER_DWELL", "60"))  # Seconds viewers must stay >= min before alerting
YT_VIEWER_ALERT_COOLDOWN = int(os.environ.get("PIPULATE_YT_VIEWER_COOLDOWN", "300"))  # Seconds between any two viewer alerts
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
def _save_ytwatch_state(state_obj):
    """Persist ytwatch state to ~/.config/pipulate/ytwatch_state.json."""
    try:
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        YTWATCH_STATE_FILE.write_text(json.dumps(state_obj, indent=2) + "\n", encoding="utf-8")
    except OSError:
        pass
[[[DIVIDER]]]
def _save_ytwatch_state(state_obj):
    """Persist ytwatch state to ~/.config/pipulate/ytwatch_state.json."""
    try:
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        YTWATCH_STATE_FILE.write_text(json.dumps(state_obj, indent=2) + "\n", encoding="utf-8")
    except OSError:
        pass

def _sustained_viewer_count(dwell_window, now_ts, dwell_secs):
    """Return the viewer count continuously sustained for at least dwell_secs, or 0."""
    if not dwell_window:
        return 0
    if dwell_secs <= 0:
        return int(dwell_window[-1][1])
    run_min = int(dwell_window[-1][1])
    for ts, v in reversed(dwell_window):
        run_min = min(run_min, int(v))
        if (now_ts - float(ts)) >= dwell_secs:
            return run_min
    return 0
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
    if dry_run:
        sample = 'var ytInitialPlayerResponse = {"videoDetails":{"videoId":"test1234567","isLive":true}};'
        parsed = _extract_yt_json(sample, "ytInitialPlayerResponse =")
        ok = (parsed.get("videoDetails") or {}).get("videoId") == "test1234567"
        t_label = "configured" if resolved_topic else "unconfigured"
        print(f"📡 [DRY RUN] YouTube Watch: channel={resolved_channel} | min_viewers={YT_VIEWER_ALERT_MIN} | ntfy={t_label} | parser_ok={ok}")
        return 0 if ok else 1
    try:
        state = fetch_youtube_live_state(resolved_channel)
    except Exception as exc:
        print(f"❌ YouTube Watch fetch failed for {resolved_channel}: {exc}")
        return 1
    persisted = _load_ytwatch_state()
    same_video = (persisted.get("video_id") == state["video_id"] and bool(state["video_id"]))
    seen_ids = set(persisted.get("seen_ids") or []) if same_video else set()
    had_prior_snapshot = bool(same_video and persisted.get("seen_ids") is not None)
    prev_viewers = int(persisted.get("concurrent_viewers") or 0) if same_video else state["concurrent_viewers"]
    peak_viewers = int(persisted.get("peak_viewers") or 0) if same_video else state["concurrent_viewers"]
    prev_comments = persisted.get("comment_count") if same_video else state["comment_count"]
    last_viewer_alert = float(persisted.get("last_viewer_alert") or 0.0) if same_video else 0.0
    daemon_running = (
        subprocess.run(
            ["tmux", "has-session", "-t", "ytwatch"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        ).returncode == 0
    )
    c_str = f" | comments={state['comment_count']}" if state["comment_count"] is not None else ""
    d_str = " | daemon=RUNNING" if daemon_running else ""
    print(
        f"📡 YouTube Watch: LIVE={'yes' if state['is_live'] else 'no'} | "
        f"video={state['video_id'] or 'NONE'} | viewers={state['concurrent_viewers']} (alert>={YT_VIEWER_ALERT_MIN}) | "
        f"chat_items={len(state['chat_messages'])}{c_str}{d_str} | title={state['title']!r}"
    )
[[[DIVIDER]]]
    if dry_run:
        sample = 'var ytInitialPlayerResponse = {"videoDetails":{"videoId":"test1234567","isLive":true}};'
        parsed = _extract_yt_json(sample, "ytInitialPlayerResponse =")
        ok = (parsed.get("videoDetails") or {}).get("videoId") == "test1234567"
        t_label = "configured" if resolved_topic else "unconfigured"
        print(f"📡 [DRY RUN] YouTube Watch: channel={resolved_channel} | min_viewers={YT_VIEWER_ALERT_MIN} | dwell={YT_VIEWER_ALERT_DWELL}s | ntfy={t_label} | parser_ok={ok}")
        return 0 if ok else 1
    try:
        state = fetch_youtube_live_state(resolved_channel)
    except Exception as exc:
        print(f"❌ YouTube Watch fetch failed for {resolved_channel}: {exc}")
        return 1
    persisted = _load_ytwatch_state()
    same_video = (persisted.get("video_id") == state["video_id"] and bool(state["video_id"]))
    seen_ids = set(persisted.get("seen_ids") or []) if same_video else set()
    had_prior_snapshot = bool(same_video and persisted.get("seen_ids") is not None)
    peak_viewers = int(persisted.get("peak_viewers") or 0) if same_video else 0
    prev_comments = persisted.get("comment_count") if same_video else state["comment_count"]
    last_viewer_alert = float(persisted.get("last_viewer_alert") or 0.0) if same_video else 0.0
    alerted_in_run = bool(persisted.get("alerted_in_run")) if (same_video and state["concurrent_viewers"] >= YT_VIEWER_ALERT_MIN) else False
    daemon_running = (
        subprocess.run(
            ["tmux", "has-session", "-t", "ytwatch"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        ).returncode == 0
    )
    c_str = f" | comments={state['comment_count']}" if state["comment_count"] is not None else ""
    d_str = " | daemon=RUNNING" if daemon_running else ""
    print(
        f"📡 YouTube Watch: LIVE={'yes' if state['is_live'] else 'no'} | "
        f"video={state['video_id'] or 'NONE'} | viewers={state['concurrent_viewers']} (alert>={YT_VIEWER_ALERT_MIN} for {YT_VIEWER_ALERT_DWELL}s) | "
        f"chat_items={len(state['chat_messages'])}{c_str}{d_str} | title={state['title']!r}"
    )
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
        _save_ytwatch_state({
            "video_id": state["video_id"],
            "concurrent_viewers": state["concurrent_viewers"],
            "peak_viewers": max(peak_viewers, state["concurrent_viewers"]),
            "comment_count": state["comment_count"],
            "last_viewer_alert": last_viewer_alert,
            "seen_ids": all_ids,
            "updated_at": int(time.time()),
        })
        return 0
    for m in state["chat_messages"]:
        seen_ids.add(m["id"])
    _save_ytwatch_state({
        "video_id": state["video_id"],
        "concurrent_viewers": state["concurrent_viewers"],
        "peak_viewers": max(peak_viewers, state["concurrent_viewers"]),
        "comment_count": state["comment_count"],
        "last_viewer_alert": last_viewer_alert,
        "seen_ids": list(seen_ids)[-500:],
        "updated_at": int(time.time()),
    })
    print(f"🔭 Watching {resolved_channel} every {interval}s...")
    print("💡 [HOW TO EXIT TMUX WITHOUT STOPPING]: Press 'q' OR press Ctrl+b, release, then press d")
    try:
        while True:
            time.sleep(max(5, int(interval)))
            try:
                state = fetch_youtube_live_state(resolved_channel)
            except Exception as exc:
                print(f"⚠️ Poll error: {exc}")
                continue
            viewers = state["concurrent_viewers"]
            now = time.time()
            cooldown_ok = (now - last_viewer_alert) > YT_VIEWER_ALERT_COOLDOWN
            crossed_min = prev_viewers < YT_VIEWER_ALERT_MIN and viewers >= YT_VIEWER_ALERT_MIN
            new_peak = viewers >= YT_VIEWER_ALERT_MIN and viewers > peak_viewers
            if (crossed_min or new_peak) and cooldown_ok:
                peak_viewers = max(peak_viewers, viewers)
                last_viewer_alert = now
                alert_msg = f"{viewers} watching now (min={YT_VIEWER_ALERT_MIN}) — {state['title']}"
                print(f"👀 [VIEWER] {alert_msg}", flush=True)
                spoken_v = SPOKEN_ALERT_TABLE["viewers_threshold"].format(
                    count=viewers, min_viewers=YT_VIEWER_ALERT_MIN
                )
                _play_local_ding(ding, spoken_text=spoken_v)
                if resolved_topic:
                    send_ntfy_notification(
                        alert_msg,
                        title="👀 Honeybot Live Viewer",
                        priority="default",
                        tags="eyes",
                        topic=resolved_topic,
                    )
            prev_viewers = viewers
[[[DIVIDER]]]
        _save_ytwatch_state({
            "video_id": state["video_id"],
            "concurrent_viewers": state["concurrent_viewers"],
            "peak_viewers": peak_viewers,
            "alerted_in_run": alerted_in_run,
            "comment_count": state["comment_count"],
            "last_viewer_alert": last_viewer_alert,
            "seen_ids": all_ids,
            "updated_at": int(time.time()),
        })
        return 0
    for m in state["chat_messages"]:
        seen_ids.add(m["id"])
    init_ts = time.time()
    dwell_window = [(init_ts, state["concurrent_viewers"])] if state["concurrent_viewers"] >= YT_VIEWER_ALERT_MIN else []
    _save_ytwatch_state({
        "video_id": state["video_id"],
        "concurrent_viewers": state["concurrent_viewers"],
        "peak_viewers": peak_viewers,
        "alerted_in_run": alerted_in_run,
        "comment_count": state["comment_count"],
        "last_viewer_alert": last_viewer_alert,
        "seen_ids": list(seen_ids)[-500:],
        "updated_at": int(init_ts),
    })
    print(f"🔭 Watching {resolved_channel} every {interval}s (dwell={YT_VIEWER_ALERT_DWELL}s)...")
    print("💡 [HOW TO EXIT TMUX WITHOUT STOPPING]: Press 'q' OR press Ctrl+b, release, then press d")
    try:
        while True:
            time.sleep(max(5, int(interval)))
            poll_ts = time.time()
            try:
                state = fetch_youtube_live_state(resolved_channel)
            except Exception as exc:
                print(f"⚠️ Poll error: {exc}")
                continue
            viewers = state["concurrent_viewers"]
            now = time.time()
            if viewers >= YT_VIEWER_ALERT_MIN:
                dwell_window.append((poll_ts, viewers))
                cutoff_ts = poll_ts - max(YT_VIEWER_ALERT_DWELL * 3, int(interval) * 4)
                dwell_window = [(ts, v) for ts, v in dwell_window if ts >= cutoff_ts]
                if len(dwell_window) == 1 and YT_VIEWER_ALERT_DWELL > 0:
                    print(
                        f"⏳ [VIEWER DWELL] {viewers} watching (min={YT_VIEWER_ALERT_MIN}) — "
                        f"holding {YT_VIEWER_ALERT_DWELL}s to confirm they stick around...",
                        flush=True,
                    )
                sustained = _sustained_viewer_count(dwell_window, poll_ts, YT_VIEWER_ALERT_DWELL)
                cooldown_ok = (now - last_viewer_alert) > YT_VIEWER_ALERT_COOLDOWN
                if sustained >= YT_VIEWER_ALERT_MIN and (not alerted_in_run or sustained > peak_viewers) and cooldown_ok:
                    alerted_in_run = True
                    peak_viewers = max(peak_viewers, sustained)
                    last_viewer_alert = now
                    dwell_elapsed = int(round(poll_ts - dwell_window[0][0]))
                    alert_msg = (
                        f"{viewers} watching now (sustained >={dwell_elapsed}s, min={YT_VIEWER_ALERT_MIN}) — "
                        f"{state['title']}"
                    )
                    print(f"👀 [VIEWER] {alert_msg}", flush=True)
                    spoken_v = SPOKEN_ALERT_TABLE["viewers_threshold"].format(
                        count=viewers, min_viewers=YT_VIEWER_ALERT_MIN
                    )
                    _play_local_ding(ding, spoken_text=spoken_v)
                    if resolved_topic:
                        send_ntfy_notification(
                            alert_msg,
                            title="👀 Honeybot Live Viewer",
                            priority="default",
                            tags="eyes",
                            topic=resolved_topic,
                        )
            else:
                if dwell_window and not alerted_in_run and YT_VIEWER_ALERT_DWELL > 0:
                    dwell_elapsed = int(round(poll_ts - dwell_window[0][0]))
                    print(
                        f"💨 [VIEWER DWELL] Dropped to {viewers} after {dwell_elapsed}s "
                        f"(<{YT_VIEWER_ALERT_DWELL}s dwell) — notification suppressed.",
                        flush=True,
                    )
                dwell_window.clear()
                alerted_in_run = False
[[[REPLACE]]]

Target: scripts/articles/common.py
[[[SEARCH]]]
            _save_ytwatch_state({
                "video_id": state["video_id"],
                "concurrent_viewers": viewers,
                "peak_viewers": peak_viewers,
                "comment_count": cur_comments,
                "last_viewer_alert": last_viewer_alert,
                "seen_ids": list(seen_ids)[-500:],
                "updated_at": int(now),
            })
[[[DIVIDER]]]
            _save_ytwatch_state({
                "video_id": state["video_id"],
                "concurrent_viewers": viewers,
                "peak_viewers": peak_viewers,
                "alerted_in_run": alerted_in_run,
                "comment_count": cur_comments,
                "last_viewer_alert": last_viewer_alert,
                "seen_ids": list(seen_ids)[-500:],
                "updated_at": int(now),
            })
[[[REPLACE]]]

ytwatch restart

Verify the AFTER receipts for the 60-second `YT_VIEWER_ALERT_DWELL` dampening filter in `scripts/articles/common.py` and confirm `ytwatch` status.

# Content 1
# context.txt
# flake.nix
# 
# # HOME-HOSTED (NOT BEHIND CDN) NGINX WEBSERVER HONEYBOT AI OBSERVATORY
# nixops.sh
# remotes/honeybot/hooks/post-receive
# remotes/honeybot/nixos/configuration.nix
# remotes/honeybot/scripts/score.py
# remotes/honeybot/scripts/card.py
# remotes/honeybot/scripts/forest.py
# remotes/honeybot/scripts/test_forest.py
# remotes/honeybot/scripts/logs.py
# remotes/honeybot/scripts/content_loader.py
# remotes/honeybot/scripts/db.py
# 
# scripts/articles/common.py

# Context 2
context.txt
scripts/articles/common.py
! .venv/bin/python scripts/articles/common.py --broadcast watch --dry-run
! .venv/bin/python -c "import sys; sys.path.insert(0, 'scripts/articles'); import common; fn = getattr(common, '_sustained_viewer_count', None); print('dwell=', getattr(common, 'YT_VIEWER_ALERT_DWELL', None), 'spike_30s=', fn([(0.0, 3), (30.0, 3)], 30.0, 60) if fn else None, 'sustained_60s=', fn([(0.0, 3), (30.0, 3), (61.0, 3)], 61.0, 60) if fn else None)"

context.txt
flake.nix
nixops.sh
remotes/honeybot/hooks/post-receive
remotes/honeybot/nixos/configuration.nix
remotes/honeybot/scripts/score.py
remotes/honeybot/scripts/card.py
remotes/honeybot/scripts/forest.py
remotes/honeybot/scripts/test_forest.py
remotes/honeybot/scripts/logs.py
remotes/honeybot/scripts/content_loader.py
remotes/honeybot/scripts/db.py
scripts/articles/common.py

Brief viewer spikes shouldn't ring the bell. A 60-second sliding-window dwell filter in Python confirms viewers stay before firing TTS or ntfy.
https://mikelev.in/futureproof/dwell-gate-debouncing-ytwatch-viewer-alerts/
#Python #Linux #Automation

A wide 16:9 widescreen graphic-novel ink-and-wash illustration set in a deep obsidian-slate (#111111) environment, featuring a precision brass mechanical sliding-window gate and glowing amber hourglass vertically centered in the composition. From the left, jagged cyan telemetry waves representing erratic signal spikes enter a series of three brass chronometer rings; transient spikes dissipate into wisps of silver vapor at the first ring, while a steady, continuous golden beam passes cleanly through all three aligned rings to strike a resonant copper bell on the right, sending warm amber acoustic ripples outward into the dark slate workshop. High-contrast chiaroscuro lighting, crisp architectural linework, and intricate analog-digital instrument textures extend continuously to every edge of the scene. Full-bleed edge-to-edge composition filling 100% of the canvas with no borders, no frames, no matting, no margins, and no text overlays or lettering.