Example shareware heartbeat logging service for Mac/Linux

Purpose: Local testing via Mac (and optionally for Linux) for long-time heartbeat stability testing, with logging.

Disclaimer: No responsibility apply for all or any use of this code. Use at own risk!

Brief usage description:

Project: ManagdWALPlus iOS Kiosk App Heartbeat Logging for Testing
Purpose: Receive, display, and log JSON heartbeat POSTs from a managed iOS device to a Mac running as a local heatbeat logging server.

---

Overview

Two versions of heartbeat_logger.py exist, both written in pure Python 3 with no external dependencies. They can be used standalone or combined (see Combining the Scripts below).

---

Version 1 — Simple Logger

What it does

• Starts an HTTP server on a configurable port (default 8765)
• Accepts HTTP POST requests carrying a JSON body on any path (/heartbeat, /ping, /log, etc.)
• Pretty-prints each incoming beat to the terminal with timestamp and client IP, in ANSI colour
• Appends every beat as a single JSON line to heartbeats.log in the same folder as the script
• Returns {"status":"ok"} to the sender
• Responds to GET requests with a plain-text status page (useful for browser verification)

Usage

python3 heartbeat_logger.py           # port 8765 (default)
python3 heartbeat_logger.py 9000      # custom port

Terminal output

Each heartbeat prints a coloured block:

────────────────────────────────────────────────────────────
▶ 2026-04-26 17:24:33  192.168.2.75  →  /heartbeat
  {
    "device": "ManagdWALPlus",
    "status": "ok"
  }
────────────────────────────────────────────────────────────

Log file format

One JSON object per line in heartbeats.log:

{"timestamp": "2026-04-26 17:24:33", "client_ip": "192.168.2.75", "path": "/heartbeat", "payload": {"device": "ManagdWALPlus", "status": "ok"}}

Watching the log file live

tail -f heartbeats.log

Limitations

• No alerting if heartbeats stop arriving
• No stability tracking
• Plain scrolling terminal output only — no dashboard

---

Version 2 — Monitored Logger

What it does

Everything v1 does, plus:

• Two-pane curses terminal — fixed dashboard at top, scrolling beat log below
• Watchdog thread — checks every second whether a beat is overdue; fires an alert when elapsed > interval × (1 + tolerance%)
• macOS native notifications — popup via osascript with sound (“Basso”) when a beat is missed; appears in Notification Centre
• Late beat detection — if a beat arrives but is overdue, it is logged as ⚠ LATE and counted separately from a fully missed beat
• Stability metric — percentage of beats that arrived on time, updated live

Screenshot example (note forced network issue provoking some missed heartbeat transfers):

Usage

python3 heartbeat_logger.py                     # port 8888, 10s interval, 20% tolerance
python3 heartbeat_logger.py 9000                # custom port
python3 heartbeat_logger.py 9000 30             # 30s expected interval
python3 heartbeat_logger.py 9000 30 25          # 25% tolerance (deadline = 37.5s)

Dashboard (top pane)

Field	Description
Total received	Count of all POST requests received
Late arrivals	Beats that arrived after the deadline but before the next interval
Missed entirely	Full intervals with no beat at all
Stability	(total - late - missed) / (total + missed) × 100%
Last heartbeat	Timestamp + seconds elapsed since last beat (turns red when overdue)
Last client	IP address of the most recent sender
Status	● Status: OK (green) or ⚠ ALERT: … (red)

Scrolling log (bottom pane)

Same format as v1, but integrated into the curses window. Late beats are highlighted in red.

Watchdog logic

The deadline is calculated as:

deadline = expected_interval × (1 + tolerance_pct / 100)

For example, with a 10s interval and 20% tolerance: deadline = 12.0s

The watchdog fires once per missed interval. If the device is offline for three consecutive intervals, three separate notifications are sent.

Log file format

Same as v1, with an additional "late" boolean field:

{"timestamp": "2026-04-26 19:56:31", "client_ip": "192.168.2.75", "late": false, "payload": {"device": "ManagdWALPlus", "status": "ok"}}

---

1. WALconfig.plist — configurable endpoint (example):

<key>HeartbeatURL</key>
<string>http://192.168.2.200:8888/heartbeat</string>
<key>HeartbeatIntervalSeconds</key>
<integer>10</integer>

Keeping the URL in WALconfig.plist means MDM admins can change the target server without a recompile.

---

macOS Server Requirements

Firewall: System Settings → Network → Firewall → Options — python3 must be set to Allow incoming connections.

Kill a stuck port (e.g. after Ctrl-Z instead of Ctrl-C):

kill -9 $(lsof -ti :8888)

Verify the server is reachable from the Mac itself:

curl -X POST http://192.168.2.200:8888/heartbeat \
     -H "Content-Type: application/json" \
     -d '{"device":"test","status":"ok"}'

---

Combining the Scripts

The two scripts are not normally run simultaneously — v2 is a superset of v1. However, there are scenarios where running both (on different ports) makes sense:

Scenario A — Separate lightweight file log + rich monitor

Run v1 on port 8765 purely for clean file logging (no curses overhead, easier to tail -f), and v2 on port 8888 for the live dashboard. Configure the iOS app to POST to both endpoints:

// Send to both loggers
let endpoints = [
    "http://192.168.2.200:8765/heartbeat",  // v1 — file log
    "http://192.168.2.200:8888/heartbeat",  // v2 — dashboard
]
for url in endpoints { sendHeartbeat(to: url) }

Scenario B — Multiple devices, one monitor

Run one v2 instance per device on different ports. Each instance maintains its own log file and dashboard. Tile terminal windows side by side.

# Terminal 1 — Device A
python3 heartbeat_logger.py 8881 10 20

# Terminal 2 — Device B
python3 heartbeat_logger.py 8882 10 20

Scenario C — Merge both into a single enhanced script

The cleanest long-term solution is to merge v1’s simplicity into v2 by adding a --no-ui flag that disables curses and falls back to v1-style plain terminal output. This makes a single script deployable in both interactive (dashboard) and headless (SSH, cron, CI) contexts:

python3 heartbeat_logger.py 8888 10 20          # curses dashboard (default)
python3 heartbeat_logger.py 8888 10 20 --no-ui  # plain terminal, v1 style

This is straightforward to implement — the HTTP server, watchdog, and file logging are already cleanly separated from the UI layer in v2. The --no-ui branch would simply skip curses.wrapper(draw) and print coloured lines directly to stdout instead.

---

Quick Reference

	v1	v2
Terminal UI	Plain scrolling	Curses dashboard + scroll
Alerting	None	Watchdog + macOS notification
Late detection	No	Yes
Stability metric	No	Yes
File logging	Yes	Yes
Args	port	port interval_s tolerance_pct
Default port	8765	8888
Stop cleanly	Ctrl-C	Ctrl-C

Code sample, heartbeat_logger.py

#!/usr/bin/env python3
"""
heartbeat_logger.py
-------------------
Lightweight REST/JSON heartbeat receiver for ManagdWALPlus (or any app).
Logs incoming POST requests to the terminal (pretty-printed) and to a file.

Usage:
    python3 heartbeat_logger.py              # default port 8765
    python3 heartbeat_logger.py 9000         # custom port

Your app should POST JSON to:
    http://<your-mac-ip>:<port>/heartbeat

Any path is accepted, so /ping, /log, / etc. all work too.
"""

import http.server
import json
import sys
import os
from datetime import datetime

# ── Configuration ──────────────────────────────────────────────────────────────

PORT      = int(sys.argv[1]) if len(sys.argv) > 1 else 8765
LOG_FILE  = os.path.join(os.path.dirname(os.path.abspath(__file__)), "heartbeats.log")

# ── Colours for terminal output ────────────────────────────────────────────────

GREEN  = "\033[92m"
CYAN   = "\033[96m"
YELLOW = "\033[93m"
RED    = "\033[91m"
DIM    = "\033[2m"
RESET  = "\033[0m"

# ── Request handler ────────────────────────────────────────────────────────────

class HeartbeatHandler(http.server.BaseHTTPRequestHandler):

    def do_POST(self):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        client_ip = self.client_address[0]

        # Read body
        length = int(self.headers.get("Content-Length", 0))
        raw_body = self.rfile.read(length) if length > 0 else b""

        # Parse JSON
        try:
            payload = json.loads(raw_body.decode("utf-8")) if raw_body else {}
            pretty  = json.dumps(payload, indent=2, ensure_ascii=False)
            parse_ok = True
        except json.JSONDecodeError as e:
            pretty   = raw_body.decode("utf-8", errors="replace")
            parse_ok = False
            json_err = str(e)

        # ── Terminal output ──
        sep = "─" * 60
        print(f"\n{CYAN}{sep}{RESET}")
        print(f"{GREEN}▶ {timestamp}  {YELLOW}{client_ip}{RESET}  →  {self.path}")
        if parse_ok:
            # Colour-tint the pretty JSON slightly
            for line in pretty.splitlines():
                print(f"  {DIM}{line}{RESET}")
        else:
            print(f"  {RED}[JSON parse error: {json_err}]{RESET}")
            print(f"  {DIM}{pretty}{RESET}")
        print(f"{CYAN}{sep}{RESET}")

        # ── File output ──
        log_entry = {
            "timestamp": timestamp,
            "client_ip": client_ip,
            "path":      self.path,
            "payload":   payload if parse_ok else {"_raw": pretty},
        }
        try:
            with open(LOG_FILE, "a", encoding="utf-8") as f:
                f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
        except OSError as e:
            print(f"{RED}  [Could not write to log file: {e}]{RESET}")

        # ── HTTP response ──
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"status":"ok"}')

    def do_GET(self):
        """Simple status page so you can verify the server is up from a browser."""
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.end_headers()
        msg = f"heartbeat_logger running on port {PORT}\nLog file: {LOG_FILE}\n"
        self.wfile.write(msg.encode())

    def log_message(self, format, *args):
        """Suppress the default access log — we do our own pretty printing."""
        pass


# ── Entry point ────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    server = http.server.HTTPServer(("0.0.0.0", PORT), HeartbeatHandler)

    print(f"\n{GREEN}✔  Heartbeat logger started{RESET}")
    print(f"   Listening on  : {CYAN}http://0.0.0.0:{PORT}{RESET}")
    print(f"   Log file      : {CYAN}{LOG_FILE}{RESET}")
    print(f"   POST endpoint : {CYAN}http://<your-mac-ip>:{PORT}/heartbeat{RESET}")
    print(f"   Status check  : open the URL above in a browser")
    print(f"\n{DIM}   Press Ctrl-C to stop.{RESET}\n")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print(f"\n{YELLOW}  Stopped.{RESET}\n")

Code sample, heartbeat_logger-2.py:

#!/usr/bin/env python3
"""
heartbeat_logger.py  v2
-----------------------
Enhanced REST/JSON heartbeat receiver with:
  • Two-pane curses terminal (fixed dashboard + scrolling log)
  • Watchdog thread — alerts if heartbeat is late by more than TOLERANCE_PCT
  • macOS native popup notifications via osascript (no installs needed)
  • File logging (one JSON line per beat)

Usage:
    python3 heartbeat_logger.py                   # port 8888, 10s interval, 20% tolerance
    python3 heartbeat_logger.py 9000              # custom port
    python3 heartbeat_logger.py 9000 30           # port 9000, 30s interval
    python3 heartbeat_logger.py 9000 30 25        # + 25% tolerance

POST JSON to:  http://<your-mac-ip>:<port>/heartbeat

Please keep reference to original source in your use of code, including derivates: Copyleft 2026, Royal Cloud Solutions, please use and modify as long as this reference or similar is kept

"""

import curses
import http.server
import json
import os
import subprocess
import sys
import threading
import time
from collections import deque
from datetime import datetime

# ── Configuration ──────────────────────────────────────────────────────────────

PORT              = int(sys.argv[1]) if len(sys.argv) > 1 else 8888
EXPECTED_INTERVAL = int(sys.argv[2]) if len(sys.argv) > 2 else 10   # seconds
TOLERANCE_PCT     = int(sys.argv[3]) if len(sys.argv) > 3 else 20   # percent
LOG_FILE          = os.path.join(os.path.dirname(os.path.abspath(__file__)), "heartbeats.log")
DEADLINE          = EXPECTED_INTERVAL * (1 + TOLERANCE_PCT / 100)

# ── Shared state ───────────────────────────────────────────────────────────────

state = {
    "total":         0,
    "late":          0,
    "missed":        0,
    "last_beat":     None,   # time.time() float
    "last_beat_str": "—",
    "last_client":   "—",
    "alert":         None,   # current alert string or None
    "_prev_miss_n":  0,      # internal watchdog counter
}
state_lock = threading.Lock()
log_lines  = deque(maxlen=200)
log_lock   = threading.Lock()

# ── Helpers ────────────────────────────────────────────────────────────────────

def add_log(line: str):
    with log_lock:
        log_lines.append(line)


def notify(title: str, message: str):
    """macOS native notification via osascript — silently ignored if unavailable."""
    try:
        script = (f'display notification "{message}" '
                  f'with title "{title}" sound name "Basso"')
        subprocess.run(["osascript", "-e", script],
                       check=False, capture_output=True, timeout=3)
    except Exception:
        pass


def stability_str(total: int, late: int, missed: int) -> str:
    if total == 0:
        return "—"
    good = max(0, total - late - missed)
    return f"{100 * good / (total + missed):.1f} %"


# ── Watchdog thread ────────────────────────────────────────────────────────────

def watchdog():
    while True:
        time.sleep(1)
        with state_lock:
            lb = state["last_beat"]
            if lb is None:
                continue
            elapsed  = time.time() - lb
            miss_n   = int(elapsed // EXPECTED_INTERVAL)
            prev     = state["_prev_miss_n"]

            if elapsed > DEADLINE and miss_n > prev:
                state["_prev_miss_n"] = miss_n
                state["missed"]      += 1
                msg = (f"No heartbeat for {elapsed:.0f}s "
                       f"(expected every {EXPECTED_INTERVAL}s ±{TOLERANCE_PCT}%)")
                state["alert"] = msg
                add_log(f"  ⚠  ALERT: {msg}")
                notify("⚠ Heartbeat Missing", msg)

            elif elapsed <= DEADLINE:
                state["_prev_miss_n"] = 0
                state["alert"]        = None


# ── HTTP handler ───────────────────────────────────────────────────────────────

class HeartbeatHandler(http.server.BaseHTTPRequestHandler):

    def do_POST(self):
        now       = time.time()
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        client_ip = self.client_address[0]

        length = int(self.headers.get("Content-Length", 0))
        raw    = self.rfile.read(length) if length > 0 else b""

        try:
            payload  = json.loads(raw.decode("utf-8")) if raw else {}
            pretty   = json.dumps(payload, indent=2, ensure_ascii=False)
            parse_ok = True
        except json.JSONDecodeError:
            payload  = {}
            pretty   = raw.decode("utf-8", errors="replace")
            parse_ok = False

        with state_lock:
            late = False
            if state["last_beat"] is not None:
                if now - state["last_beat"] > DEADLINE:
                    late = True
                    state["late"] += 1
            state["total"]         += 1
            state["last_beat"]      = now
            state["last_beat_str"]  = timestamp
            state["last_client"]    = client_ip
            state["_prev_miss_n"]   = 0
            state["alert"]          = None

        # File log
        entry = {
            "timestamp": timestamp,
            "client_ip": client_ip,
            "late":      late,
            "payload":   payload if parse_ok else {"_raw": pretty},
        }
        try:
            with open(LOG_FILE, "a", encoding="utf-8") as f:
                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
        except OSError as e:
            add_log(f"  [log write error: {e}]")

        # Scrolling log entry
        tag = "⚠ LATE" if late else "✓"
        add_log(f"  {tag:<7} {timestamp}  {client_ip}  {self.path}")
        if pretty.strip():
            for line in pretty.splitlines()[:6]:
                add_log(f"           {line}")

        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"status":"ok"}')

    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.end_headers()
        self.wfile.write(
            f"heartbeat_logger v2  port={PORT}  interval={EXPECTED_INTERVAL}s\n".encode()
        )

    def log_message(self, *args):
        pass


# ── Curses UI ──────────────────────────────────────────────────────────────────

DASH_ROWS = 11

def draw(stdscr):
    curses.curs_set(0)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(1, curses.COLOR_GREEN,  -1)
    curses.init_pair(2, curses.COLOR_CYAN,   -1)
    curses.init_pair(3, curses.COLOR_YELLOW, -1)
    curses.init_pair(4, curses.COLOR_RED,    -1)
    curses.init_pair(5, curses.COLOR_WHITE,  -1)

    C_GREEN  = curses.color_pair(1)
    C_CYAN   = curses.color_pair(2)
    C_YELLOW = curses.color_pair(3)
    C_RED    = curses.color_pair(4)
    C_NORM   = curses.color_pair(5)

    def safe(row, col, text, attr=0):
        rows, cols = stdscr.getmaxyx()
        if row >= rows - 1 or col >= cols:
            return
        try:
            stdscr.addstr(row, col, text[:cols - col - 1], attr)
        except curses.error:
            pass

    while True:
        rows, cols = stdscr.getmaxyx()
        stdscr.erase()

        with state_lock:
            total    = state["total"]
            late     = state["late"]
            missed   = state["missed"]
            last_str = state["last_beat_str"]
            last_cli = state["last_client"]
            last_beat= state["last_beat"]
            alert    = state["alert"]

        overdue     = False
        elapsed_str = "—"
        if last_beat:
            elapsed     = time.time() - last_beat
            elapsed_str = f"{elapsed:.1f}s ago"
            overdue     = elapsed > DEADLINE

        stab = stability_str(total, late, missed)

        # Header
        title = (f" ♥  Heartbeat Monitor  │  :{PORT}  │  "
                 f"Every {EXPECTED_INTERVAL}s ±{TOLERANCE_PCT}%  │  {os.path.basename(LOG_FILE)} ")
        safe(0, 0, title[:cols].center(cols), C_CYAN | curses.A_BOLD)
        safe(1, 0, "─" * cols, C_CYAN)

        # Stats rows
        def stat(r, label, value, colour):
            safe(r, 2,  label, C_NORM)
            safe(r, 22, str(value), colour | curses.A_BOLD)

        stat(2, "Total received  :", total,  C_GREEN)
        stat(3, "Late arrivals   :", late,   C_YELLOW if late   > 0 else C_GREEN)
        stat(4, "Missed entirely :", missed, C_RED    if missed > 0 else C_GREEN)
        stat(5, "Stability       :", stab,   C_GREEN  if late + missed == 0 else C_YELLOW)

        safe(6, 2,  "Last heartbeat  :", C_NORM)
        safe(6, 22, f"{last_str}  ({elapsed_str})",
             C_RED if overdue else C_NORM)
        safe(7, 2,  "Last client     :", C_NORM)
        safe(7, 22, last_cli, C_NORM)

        safe(8, 0, "─" * cols, C_CYAN)

        if overdue and alert:
            safe(9, 2, f"⚠  {alert}", C_RED | curses.A_BOLD)
        elif total == 0:
            safe(9, 2, "Waiting for first heartbeat…", C_YELLOW)
        else:
            safe(9, 2, "● Status: OK", C_GREEN | curses.A_BOLD)

        safe(10, 0, "─" * cols, C_CYAN)

        # Scrolling log
        log_rows = max(1, rows - DASH_ROWS)
        with log_lock:
            visible = list(log_lines)[-(log_rows):]
        for i, line in enumerate(visible):
            r = DASH_ROWS + i
            if r >= rows - 1:
                break
            attr = C_RED if ("ALERT" in line or "LATE" in line) else C_NORM
            safe(r, 0, line, attr)

        stdscr.refresh()
        time.sleep(0.4)


# ── Main ───────────────────────────────────────────────────────────────────────

def main():
    server   = http.server.HTTPServer(("0.0.0.0", PORT), HeartbeatHandler)
    t_server = threading.Thread(target=server.serve_forever, daemon=True)
    t_server.start()

    t_watch = threading.Thread(target=watchdog, daemon=True)
    t_watch.start()

    add_log("  heartbeat_logger v2 started")
    add_log(f"  Endpoint : http://0.0.0.0:{PORT}/heartbeat")
    add_log(f"  Interval : {EXPECTED_INTERVAL}s  tolerance {TOLERANCE_PCT}%"
            f"  (deadline {DEADLINE:.1f}s)")
    add_log(f"  Log file : {LOG_FILE}")
    add_log("")

    try:
        curses.wrapper(draw)
    except KeyboardInterrupt:
        pass
    finally:
        server.shutdown()
        print("\nheartbeat_logger stopped.\n")


if __name__ == "__main__":
    main()