scripts._irc_watchdog

Reference › scripts › _irc_watchdog

A watchdog for the IRC listener _irc_listen: it checks every 180 seconds and restarts the listener if it has died. No test imports it and it takes no options. main prints its interval, starts the listener if none is running (is_listener_alive, start_listener), then loops forever, sleeping for the check interval and restarting the listener whenever the check finds it dead.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/_irc_watchdog.py

"""Watchdog: checks IRC listener health every 180s, restarts if dead."""

import os
import subprocess
import sys
import time

SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_irc_listen.py")
CHECK_INTERVAL = 180  # seconds


def is_listener_alive():
    """Check if an _irc_listen.py process is running (not this watchdog)."""
    try:
        result = subprocess.run(
            ["pgrep", "-f", "_irc_listen.py"], capture_output=True, text=True
        )
        pids = [
            p for p in result.stdout.strip().split("\n") if p and str(os.getpid()) != p
        ]
        return len(pids) > 0
    except Exception:
        return False


def start_listener():
    """Start the IRC listener in the background."""
    return subprocess.Popen(
        [sys.executable, SCRIPT],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )


def main():
    print(f"irc-watchdog: checking every {CHECK_INTERVAL}s", flush=True)

    # Start listener if not already running
    if not is_listener_alive():
        print("irc-watchdog: starting listener", flush=True)
        start_listener()
    else:
        print("irc-watchdog: listener already running", flush=True)

    while True:
        time.sleep(CHECK_INTERVAL)
        if not is_listener_alive():
            print("irc-watchdog: listener died, restarting", flush=True)
            start_listener()


if __name__ == "__main__":
    main()