#!/usr/bin/env python3 PK;]Gc __main__.py"""Entry point for the single-file app (isp-connection-check.pyz).""" import sys if len(sys.argv) > 1 and sys.argv[1] in ("--monitor", "--cli"): del sys.argv[1] import pingmon raise SystemExit(pingmon.main()) import pingmon_gui raise SystemExit(pingmon_gui.main()) PK;]߃uu pingmon.py#!/usr/bin/env python3 """Monitor configured targets with ping and print failures.""" from __future__ import annotations import csv import datetime as datetime_module import ipaddress import json import platform import re import signal import socket import subprocess import sys import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path PROGRAM_NAME = "ISP Connection Check" PROGRAM_VERSION = "1.0.0" PROGRAM_LICENSE = "MIT" PROGRAM_URL = "https://github.com/amrheing/ISP-Connection-Check" @dataclass class TargetState: name: str target: str failure_start: float | None = None first_failure_at: float | None = None outages: int = 0 downtime: int = 0 consecutive_failures: int = 0 def get_default_gateway() -> str | None: try: if platform.system() == "Darwin": output = subprocess.run( ["route", "-n", "get", "default"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True ).stdout match = re.search(r"gateway:\s*(\S+)", output) else: output = subprocess.run( ["ip", "route", "show", "default"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True ).stdout match = re.search(r"default via (\S+)", output) except OSError: return None return match.group(1) if match else None def get_system_resolvers() -> list[str]: # The DNS servers actually used to resolve web/host targets, to tell if they depend on our "dns" test target. try: if platform.system() == "Darwin": output = subprocess.run( ["scutil", "--dns"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True ).stdout found = re.findall(r"nameserver\[\d+\]\s*:\s*(\S+)", output) else: output = Path("/etc/resolv.conf").read_text() found = re.findall(r"^\s*nameserver\s+(\S+)", output, re.MULTILINE) except OSError: return [] resolvers: list[str] = [] for resolver in found: if resolver not in resolvers: resolvers.append(resolver) return resolvers def _is_ip_address(value: str) -> bool: try: ipaddress.ip_address(value) return True except ValueError: return False def get_isp_info(timeout: int = 3, retries: int = 2) -> dict | None: # Identifies the current internet provider (ISP/ASN) via the public IP, useful to tell Starlink vs. carrier apart. # Retries with a short delay since a just-switched WAN interface may need a moment before routing works. for attempt in range(retries): try: with urllib.request.urlopen( "http://ip-api.com/json/?fields=isp,org,as,query", timeout=timeout ) as response: data = json.loads(response.read()) except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError): if attempt + 1 < retries: time.sleep(1) continue isp = data.get("isp") or data.get("org") if isp: return {"name": isp, "public_ip": data.get("query")} return None CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10") # RFC 6598 shared address space used by carrier-grade NAT def get_local_ip(family: socket.AddressFamily) -> str | None: probe = ("8.8.8.8", 80) if family == socket.AF_INET else ("2001:4860:4860::8888", 80) try: with socket.socket(family, socket.SOCK_DGRAM) as sock: sock.connect(probe) return sock.getsockname()[0] except OSError: return None def get_public_ip(url: str, timeout: int = 3) -> str | None: try: with urllib.request.urlopen(url, timeout=timeout) as response: value = response.read().decode().strip() except (urllib.error.URLError, TimeoutError, OSError): return None return value or None def classify_ip(public_ip: str | None, local_ip: str | None) -> str: if not public_ip: return "unavailable" try: addr = ipaddress.ip_address(public_ip) except ValueError: return "unknown" if addr.version == 4 and addr in CGNAT_NETWORK: return "CGNAT (ISP-level shared address, not a usable external IP)" if addr.is_private or addr.is_link_local: return "private/link-local (not globally routable)" if local_ip and local_ip == public_ip: return "directly routable (no NAT)" return "NAT'd (usable external IP, translated by a router)" def get_ip_status(family: socket.AddressFamily, url: str) -> dict: public_ip = get_public_ip(url) local_ip = get_local_ip(family) return {"public_ip": public_ip, "status": classify_ip(public_ip, local_ip)} SPEED_TEST_BYTES = 5_000_000 SPEED_TEST_HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; pingmon-speedtest/1.0)"} def run_speed_test(payload_bytes: int = SPEED_TEST_BYTES, timeout: int = 15) -> dict: # Single-connection HTTP test against Cloudflare's speed endpoint - a rough estimate, not an Ookla-grade measurement. # Cloudflare returns 403 for requests without a browser-like User-Agent, hence the header below. result = {"download_mbps": None, "upload_mbps": None} try: request = urllib.request.Request( f"https://speed.cloudflare.com/__down?bytes={payload_bytes}", headers=SPEED_TEST_HEADERS ) start = time.monotonic() with urllib.request.urlopen(request, timeout=timeout) as response: downloaded = len(response.read()) elapsed = time.monotonic() - start if elapsed > 0 and downloaded > 0: result["download_mbps"] = round((downloaded * 8 / 1_000_000) / elapsed, 2) except (urllib.error.URLError, TimeoutError, OSError): pass try: payload = b"0" * payload_bytes request = urllib.request.Request( "https://speed.cloudflare.com/__up", data=payload, method="POST", headers=SPEED_TEST_HEADERS ) start = time.monotonic() with urllib.request.urlopen(request, timeout=timeout) as response: response.read() elapsed = time.monotonic() - start if elapsed > 0: result["upload_mbps"] = round((payload_bytes * 8 / 1_000_000) / elapsed, 2) except (urllib.error.URLError, TimeoutError, OSError): pass return result DEFAULT_CONNECTION_PROFILES = { "starlink": { "match": ["starlink", "spacex", "space exploration technologies"], "interval": 2, "ping_timeout": 1, "failure_threshold": 3, }, "cellular": { "match": ["telekom", "vodafone", "t-mobile", "o2 ", "mobilfunk", "cellular", " lte", "wireless", "mobile"], "interval": 2, "ping_timeout": 1, "failure_threshold": 1, }, "default": { "match": [], "interval": 2, "ping_timeout": 1, "failure_threshold": 1, }, } def select_connection_profile(isp_name: str | None, profiles: dict) -> str: if isp_name: lower = isp_name.lower() for name, profile in profiles.items(): if name != "default" and any(keyword in lower for keyword in profile.get("match", [])): return name return "default" DEFAULT_CONFIG = { "targets": { "dns": "1.1.1.1", "web": "google.com", "gateway": "", }, "connection_profiles": DEFAULT_CONNECTION_PROFILES, "isp_check_interval": 15, "speed_test_on_connection_change": True, "speed_test_interval_minutes": 60, "log_dir": "logs", "log_prefix": "ping_monitor", "logfiles_to_keep": 10, "verbose_downtime": True, } def create_default_config(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(DEFAULT_CONFIG, indent=2) + "\n") def load_config(path: Path) -> tuple[list[TargetState], dict]: try: config = json.loads(path.read_text()) except FileNotFoundError as error: raise ValueError(f"Config file not found: {path}") from error except json.JSONDecodeError as error: raise ValueError(f"Invalid JSON in {path}: {error}") from error targets_config = dict(config.get("targets", {})) if not isinstance(targets_config, dict): raise ValueError("The config must contain a 'targets' object") if not targets_config.get("gateway"): default_gateway = get_default_gateway() if default_gateway: targets_config["gateway"] = default_gateway config["_gateway_source"] = "default (auto-detected)" else: targets_config.pop("gateway", None) config["_gateway_source"] = "unavailable" else: config["_gateway_source"] = "configured" if not targets_config: raise ValueError("The config must contain a non-empty 'targets' object") targets = [TargetState(name, target) for name, target in targets_config.items()] interval = config.get("interval", 2) ping_timeout = config.get("ping_timeout", 1) failure_threshold = config.get("failure_threshold", 1) if not isinstance(interval, int) or interval < 1: raise ValueError("interval must be a positive integer") if not isinstance(ping_timeout, int) or ping_timeout < 1: raise ValueError("ping_timeout must be a positive integer") if not isinstance(failure_threshold, int) or failure_threshold < 1: raise ValueError("failure_threshold must be a positive integer") return targets, config def timestamp(value: float | None = None) -> str: moment = datetime_module.datetime.fromtimestamp(value or time.time()) return moment.strftime("%Y-%m-%d %H:%M:%S") def format_duration(seconds: int) -> str: hours, remainder = divmod(seconds, 3600) minutes, seconds = divmod(remainder, 60) return f"{hours:02d}h {minutes:02d}m {seconds:02d}s" def format_bytes(num_bytes: float) -> str: for unit in ("B", "KB", "MB", "GB"): if num_bytes < 1024 or unit == "GB": return f"{num_bytes:.1f} {unit}" num_bytes /= 1024 return f"{num_bytes:.1f} GB" PING_ROUNDTRIP_BYTES = 128 # rough ICMP request+reply size per target per check ISP_CHECK_RESPONSE_BYTES = 500 # rough ip-api.com JSON response size def estimate_hourly_data_usage( num_targets: int, interval: int, isp_check_interval: int, speed_test_interval_minutes: int, ) -> dict: checks_per_hour = 3600 / interval ping_bytes = checks_per_hour * num_targets * PING_ROUNDTRIP_BYTES isp_checks_per_hour = 3600 / isp_check_interval if isp_check_interval else 0 isp_bytes = isp_checks_per_hour * ISP_CHECK_RESPONSE_BYTES speed_tests_per_hour = 60 / speed_test_interval_minutes if speed_test_interval_minutes else 0 speed_bytes = speed_tests_per_hour * (SPEED_TEST_BYTES * 2) # download + upload return { "ping": ping_bytes, "isp_check": isp_bytes, "speed_test": speed_bytes, "total": ping_bytes + isp_bytes + speed_bytes, } def ping(target: str, timeout: int) -> bool: if platform.system() == "Darwin": command = ["ping", "-c", "1", "-W", str(timeout * 1000), target] else: command = ["ping", "-c", "1", "-W", str(timeout), target] try: result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except OSError: return False return result.returncode == 0 def prune_old_logs(log_dir: Path, log_prefix: str, keep: int) -> None: if keep < 1: return for suffix in (".log", ".csv"): files = sorted(log_dir.glob(f"{log_prefix}_*{suffix}"), key=lambda p: p.stat().st_mtime, reverse=True) for old_file in files[keep:]: old_file.unlink(missing_ok=True) def load_speed_stats(path: Path) -> dict: try: return json.loads(path.read_text()) except (FileNotFoundError, json.JSONDecodeError): return {} def save_speed_stats(path: Path, stats: dict) -> None: path.write_text(json.dumps(stats, indent=2) + "\n") def record_speed_result(stats: dict, provider: str, result: dict) -> dict: # Keeps a running average per provider so failovers can be compared against each uplink's own history. entry = stats.setdefault(provider, {}) if result["download_mbps"] is not None: count = entry.get("count_download", 0) + 1 avg = entry.get("download_avg", 0.0) + (result["download_mbps"] - entry.get("download_avg", 0.0)) / count entry["count_download"] = count entry["download_avg"] = round(avg, 2) entry["last_download"] = result["download_mbps"] if result["upload_mbps"] is not None: count = entry.get("count_upload", 0) + 1 avg = entry.get("upload_avg", 0.0) + (result["upload_mbps"] - entry.get("upload_avg", 0.0)) / count entry["count_upload"] = count entry["upload_avg"] = round(avg, 2) entry["last_upload"] = result["upload_mbps"] return entry def format_speed_stat_entry(entry: dict | None) -> str: if not entry: return "no data yet" down = ( f"down avg={entry['download_avg']} Mbps ({entry['count_download']} runs, last={entry['last_download']})" if "download_avg" in entry else "down: no data" ) up = ( f"up avg={entry['upload_avg']} Mbps ({entry['count_upload']} runs, last={entry['last_upload']})" if "upload_avg" in entry else "up: no data" ) return f"{down}, {up}" def main() -> int: print("=" * 50) print(f" {PROGRAM_NAME} v{PROGRAM_VERSION}") print(f" License: {PROGRAM_LICENSE} - {PROGRAM_URL}") print("=" * 50) script_dir = Path(__file__).resolve().parent if script_dir.is_file(): # running from inside the .pyz single-file app script_dir = script_dir.parent config_path = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else script_dir / "pingmon.json" if not config_path.is_absolute(): config_path = Path.cwd() / config_path if not config_path.exists(): create_default_config(config_path) print(f"No config found, created default config at {config_path}") try: targets, config = load_config(config_path) except ValueError as error: print(f"ERROR: {error}", file=sys.stderr) return 1 log_dir = Path(config.get("log_dir", "logs")) if not log_dir.is_absolute(): log_dir = script_dir / log_dir log_dir.mkdir(parents=True, exist_ok=True) log_prefix = config.get("log_prefix", "ping_monitor") logfiles_to_keep = config.get("logfiles_to_keep", 10) prune_old_logs(log_dir, log_prefix, logfiles_to_keep - 1) # -1: this run's new log/csv still to be created stamp = time.strftime("%Y%m%d_%H%M%S") log_path = log_dir / f"{log_prefix}_{stamp}.log" csv_path = log_dir / f"{log_prefix}_{stamp}.csv" isp_info = get_isp_info() ipv4_status = get_ip_status(socket.AF_INET, "https://api.ipify.org") ipv6_status = get_ip_status(socket.AF_INET6, "https://api6.ipify.org") speed_result = run_speed_test() speed_stats_path = config_path.parent / "pingmon_speed_stats.json" speed_stats = load_speed_stats(speed_stats_path) record_speed_result(speed_stats, isp_info["name"] if isp_info else "unknown", speed_result) save_speed_stats(speed_stats_path, speed_stats) profiles = {**DEFAULT_CONNECTION_PROFILES, **(config.get("connection_profiles") or {})} profile_name = select_connection_profile(isp_info["name"] if isp_info else None, profiles) profile = profiles.get(profile_name, DEFAULT_CONNECTION_PROFILES["default"]) # Explicit top-level settings always win over the auto-selected connection profile. interval = config["interval"] if "interval" in config else profile.get("interval", 2) ping_timeout = config["ping_timeout"] if "ping_timeout" in config else profile.get("ping_timeout", 1) failure_threshold = ( config["failure_threshold"] if "failure_threshold" in config else profile.get("failure_threshold", 1) ) verbose = config.get("verbose_downtime", True) isp_check_interval = config.get("isp_check_interval", 15) speed_test_on_connection_change = config.get("speed_test_on_connection_change", True) speed_test_interval_minutes = config.get("speed_test_interval_minutes", 60) started = time.time() stopping = False def stop(_signum: int, _frame: object) -> None: nonlocal stopping stopping = True signal.signal(signal.SIGINT, stop) signal.signal(signal.SIGTERM, stop) with csv_path.open("w", newline="") as csv_file, log_path.open("w") as log_file: writer = csv.writer(csv_file) writer.writerow(["name", "target", "failure_start", "failure_end", "duration_seconds", "status"]) def report(level: str, message: str) -> None: line = f"[{timestamp()}] [{level}] {message}" print(line, flush=True) log_file.write(line + "\n") log_file.flush() print(f"PING FAILURE MONITOR\nConfig: {config_path}\nTargets: {len(targets)}") report("INFO", "Monitoring started") gateway_source = config.get("_gateway_source", "unavailable") provider_display = f"{isp_info['name']} ({isp_info['public_ip']})" if isp_info else "unknown (no internet or lookup failed)" target_width = max(len(state.name) for state in targets) system_resolvers = get_system_resolvers() dns_target = next((state for state in targets if state.name == "dns"), None) dns_in_resolvers = bool(dns_target) and dns_target.target in system_resolvers hostname_targets = [ state.name for state in targets if state is not dns_target and not _is_ip_address(state.target) ] if not system_resolvers: dns_chain_note = "system resolver(s) unknown" elif not dns_target: dns_chain_note = f"system resolver(s) {system_resolvers}; no 'dns' target configured to compare" elif dns_in_resolvers: dns_chain_note = ( f"system resolver(s) {system_resolvers}; dns test host IS used to resolve " f"{hostname_targets or 'hostname targets'} - their failures may just reflect a dns outage" ) else: dns_chain_note = ( f"system resolver(s) {system_resolvers}; dns test host is NOT used to resolve " f"{hostname_targets or 'hostname targets'} - their failures are independent of the dns target" ) print("-" * 50) print(" Parameter legend") print(" Verbose downtime : log every failed ping, incl. ones still within failure_threshold, and WARN while down") print(" Gateway : source of the gateway target (configured vs. auto-detected)") print(" Provider : ISP/ASN owning the current public IP") print(" Profile : connection profile auto-selected based on the provider") print(" Interval : seconds between ping checks") print(" Ping timeout : seconds to wait for a ping reply") print(" Failure threshold : consecutive failed checks before a target is logged as down") print(" IPv4 / IPv6 : public address and whether it's NAT'd/CGNAT or a directly usable external IP") print(" DNS chain : whether hostname targets resolve via the same server as the 'dns' target") print(" ISP check interval: how often (s) the provider is re-checked, to catch silent multi-WAN failovers") print(" Speed test : single-connection download/upload estimate") print(" Logfiles to keep : max log/CSV file pairs retained in the log dir, oldest pruned on startup") print("-" * 50) print(" General") print(f" Log file : {log_path}") print(f" CSV file : {csv_path}") print(f" Verbose downtime : {verbose}") print(f" Gateway : {gateway_source}") print(f" Logfiles to keep : {logfiles_to_keep}") print(" Targets") for state in targets: print(f" {state.name.ljust(target_width)} -> {state.target}") print(" Connection") print(f" Provider : {provider_display}") print(f" Profile : {profile_name}") print(f" Interval : {interval}s") print(f" Ping timeout : {ping_timeout}s") print(f" Failure threshold : {failure_threshold} consecutive fail(s)") print(f" IPv4 : {ipv4_status['public_ip'] or 'unavailable'} ({ipv4_status['status']})") print(f" IPv6 : {ipv6_status['public_ip'] or 'unavailable'} ({ipv6_status['status']})") print(f" DNS chain : {dns_chain_note}") print(f" ISP check interval: {isp_check_interval}s") print( f" Speed test : " f"down={speed_result['download_mbps'] if speed_result['download_mbps'] is not None else 'unavailable'} Mbps, " f"up={speed_result['upload_mbps'] if speed_result['upload_mbps'] is not None else 'unavailable'} Mbps " f"(on connection change: {speed_test_on_connection_change}, every: " f"{f'{speed_test_interval_minutes} min' if speed_test_interval_minutes else 'disabled'})" ) print(f" Speed avg (provider): {format_speed_stat_entry(speed_stats.get(isp_info['name'] if isp_info else 'unknown'))}") print("-" * 50) print(" Info") if speed_result["download_mbps"] is not None or speed_result["upload_mbps"] is not None: print( f" Ran a baseline speed test at startup: down={speed_result['download_mbps']} Mbps, " f"up={speed_result['upload_mbps']} Mbps. It will re-run " f"{'on every connection change' if speed_test_on_connection_change else 'only on the schedule below'}" f"{' and' if speed_test_on_connection_change and speed_test_interval_minutes else ''}" f"{f' every {speed_test_interval_minutes} min' if speed_test_interval_minutes else ''}." ) else: print(" Baseline speed test at startup failed or is unavailable (no internet or the endpoint was unreachable).") usage = estimate_hourly_data_usage(len(targets), interval, isp_check_interval, speed_test_interval_minutes) print(" Estimated data volume from these settings (excludes extra speed tests triggered by connection changes):") print(f" Pings : ~{format_bytes(usage['ping'])}/hour") print(f" ISP checks : ~{format_bytes(usage['isp_check'])}/hour") print(f" Speed tests: ~{format_bytes(usage['speed_test'])}/hour") print( f" Total : ~{format_bytes(usage['total'])}/hour, " f"~{format_bytes(usage['total'] * 24)}/day, ~{format_bytes(usage['total'] * 24 * 30)}/month" ) print("-" * 50) def refresh_connection() -> None: # Behind a multi-WAN failover router, the active uplink can change without any ping ever failing. nonlocal isp_info, profile_name, profile, interval, ping_timeout, failure_threshold, speed_result new_isp_info = get_isp_info() if new_isp_info is None: # A failed lookup (e.g. during a real outage, no internet yet) is not a confirmed provider # change - keep the current profile instead of flip-flopping to "unknown"/default. return new_provider = new_isp_info["name"] old_provider = isp_info["name"] if isp_info else None if new_provider == old_provider: return new_profile_name = select_connection_profile(new_provider, profiles) new_profile = profiles.get(new_profile_name, DEFAULT_CONNECTION_PROFILES["default"]) report( "INFO", f"Connection changed: {old_provider or 'unknown'} -> {new_provider or 'unknown'} " f"(profile: {profile_name} -> {new_profile_name})", ) report( "INFO", f"Previous provider ({old_provider or 'unknown'}) speed: " f"{format_speed_stat_entry(speed_stats.get(old_provider or 'unknown'))}", ) isp_info = new_isp_info profile_name = new_profile_name profile = new_profile if "interval" not in config: interval = profile.get("interval", 2) if "ping_timeout" not in config: ping_timeout = profile.get("ping_timeout", 1) if "failure_threshold" not in config: failure_threshold = profile.get("failure_threshold", 1) if speed_test_on_connection_change: speed_result = run_speed_test() record_speed_result(speed_stats, new_provider, speed_result) save_speed_stats(speed_stats_path, speed_stats) report( "INFO", f"Speed test: down=" f"{speed_result['download_mbps'] if speed_result['download_mbps'] is not None else 'unavailable'} Mbps, " f"up={speed_result['upload_mbps'] if speed_result['upload_mbps'] is not None else 'unavailable'} Mbps", ) def run_periodic_speed_test() -> None: nonlocal speed_result speed_result = run_speed_test() record_speed_result(speed_stats, isp_info["name"] if isp_info else "unknown", speed_result) save_speed_stats(speed_stats_path, speed_stats) report( "INFO", f"Speed test: down=" f"{speed_result['download_mbps'] if speed_result['download_mbps'] is not None else 'unavailable'} Mbps, " f"up={speed_result['upload_mbps'] if speed_result['upload_mbps'] is not None else 'unavailable'} Mbps", ) total_outages = 0 down_targets: set[str] = set() last_isp_check = time.time() last_speed_test = time.time() while not stopping: if time.time() - last_isp_check >= isp_check_interval: refresh_connection() last_isp_check = time.time() if speed_test_interval_minutes and time.time() - last_speed_test >= speed_test_interval_minutes * 60: run_periodic_speed_test() last_speed_test = time.time() for state in targets: if ping(state.target, ping_timeout): state.consecutive_failures = 0 if state.failure_start is not None: ended = time.time() seconds = int(ended - state.first_failure_at) state.outages += 1 state.downtime += seconds report("INFO", f"{state.name} ({state.target}) recovered after {format_duration(seconds)}") writer.writerow([state.name, state.target, timestamp(state.first_failure_at), timestamp(ended), seconds, "COMPLETE"]) csv_file.flush() state.failure_start = None state.first_failure_at = None down_targets.discard(state.name) # Overlapping outages across targets count as a single combined outage. if not down_targets: print(f"--- outages since start ({timestamp(started)}): {total_outages} ---") refresh_connection() last_isp_check = time.time() print() continue state.consecutive_failures += 1 if state.consecutive_failures == 1: state.first_failure_at = time.time() if state.consecutive_failures < failure_threshold: if verbose: report( "WARN", f"{state.name} ({state.target}) ping failed " f"({state.consecutive_failures}/{failure_threshold}, within failure_threshold)", ) continue if state.failure_start is None: state.failure_start = time.time() if not down_targets: total_outages += 1 # Check right as the outage starts - a brief auto-failover may already have # switched (and switched back) before we'd otherwise notice at recovery. refresh_connection() last_isp_check = time.time() down_targets.add(state.name) report("ERROR", f"{state.name} ({state.target}) FAILURE") elif verbose: seconds = int(time.time() - state.first_failure_at) report("WARN", f"{state.name} ({state.target}) still down for {format_duration(seconds)}") time.sleep(interval) ended = time.time() for state in targets: if state.failure_start is not None: seconds = int(ended - state.first_failure_at) state.outages += 1 state.downtime += seconds writer.writerow([state.name, state.target, timestamp(state.first_failure_at), timestamp(ended), seconds, "INCOMPLETE"]) print("\nPING MONITOR SUMMARY") print(f"Runtime: {format_duration(int(ended - started))}") for state in targets: print(f"{state.name} ({state.target}): {state.outages} outages, {format_duration(state.downtime)} downtime") print(f"Log: {log_path}\nCSV: {csv_path}") return 0 if __name__ == "__main__": raise SystemExit(main()) PK˅;]WRpingmon_gui.py#!/usr/bin/env python3 """Desktop GUI for pingmon.py: live monitor view plus a settings editor for pingmon.json.""" from __future__ import annotations import json import glob import os import queue import re import signal import subprocess import sys import threading import time import tkinter as tk from pathlib import Path from tkinter import filedialog, messagebox, ttk from tkinter.scrolledtext import ScrolledText import pingmon MODULE_DIR = Path(__file__).resolve().parent PACKAGED = MODULE_DIR.is_file() # running from inside the .pyz single-file app # Config and logs live next to the app file, never inside the .pyz. SCRIPT_DIR = MODULE_DIR.parent if PACKAGED else MODULE_DIR MONITOR_COMMAND = [str(MODULE_DIR), "--monitor"] if PACKAGED else [str(MODULE_DIR / "pingmon.py")] DEFAULT_CONFIG_PATH = SCRIPT_DIR / "pingmon.json" SECTION_HEADERS = {"Parameter legend", "General", "Targets", "Connection", "Info"} EVENT_RE = re.compile(r"^\[[^\]]+\] \[(INFO|WARN|ERROR)\] (.*)$") KV_RE = re.compile(r"^ {3}(\S[^:]*?)\s*:\s*(.*)$") DATA_KV_RE = re.compile(r"^ {5}(\S[^:]*?)\s*:\s*(.*)$") TARGET_RE = re.compile(r"^ {3}(\S+)\s*->\s*(.*)$") FAILURE_RE = re.compile(r"^(\S+) \((.*)\) FAILURE$") RECOVERED_RE = re.compile(r"^(\S+) \((.*)\) recovered after (\d+)h (\d+)m (\d+)s$") FLAKY_RE = re.compile(r"^(\S+) \((.*)\) ping failed") CHANGED_RE = re.compile(r"^Connection changed: (.*) -> (.*) \(profile: \S+ -> (\S+)\)$") SPEED_RE = re.compile(r"Speed test\s*:\s*down=(\S+) Mbps, up=(\S+) Mbps") STATUS_COLORS = { "STOPPED": "#7f8c8d", "STARTING": "#3498db", "ONLINE": "#27ae60", "DEGRADED": "#f39c12", "OUTAGE": "#e74c3c", } def format_match(keywords: list[str]) -> str: # JSON-quoted so keywords with meaningful spaces (e.g. "o2 ", " lte") survive editing. return ", ".join(json.dumps(keyword) for keyword in keywords) def parse_match(text: str) -> list[str]: text = text.strip() if not text: return [] try: parsed = json.loads(f"[{text}]") if all(isinstance(item, str) for item in parsed): return parsed except json.JSONDecodeError: pass return [part.strip() for part in text.split(",") if part.strip()] class PingmonGui(tk.Tk): def __init__(self, config_path: Path) -> None: super().__init__() self.title(f"{pingmon.PROGRAM_NAME} v{pingmon.PROGRAM_VERSION}") self.geometry("1150x780") self.minsize(950, 620) self.proc: subprocess.Popen | None = None self.output: queue.Queue[str | None] = queue.Queue() self.started_at: float | None = None self.raw_config: dict = {} self.targets: dict[str, str] = {} self.profiles: dict[str, dict] = {} self.section: str | None = None self.target_rows: dict[str, dict] = {} self.down_targets: set[str] = set() self.total_outages = 0 self.config_path_var = tk.StringVar(value=str(config_path)) notebook = ttk.Notebook(self) notebook.pack(fill="both", expand=True) monitor_tab = ttk.Frame(notebook) settings_tab = ttk.Frame(notebook) notebook.add(monitor_tab, text="Monitor") notebook.add(settings_tab, text="Settings") self.build_monitor_tab(monitor_tab) self.build_settings_tab(settings_tab) self.protocol("WM_DELETE_WINDOW", self.on_close) self.load_config() self.set_status("STOPPED") self.after(150, self.poll_output) self.after(1000, self.tick) # ------------------------------------------------------------------ monitor tab def build_monitor_tab(self, parent: ttk.Frame) -> None: bar = ttk.Frame(parent) bar.pack(fill="x", padx=8, pady=8) self.status_label = tk.Label(bar, text="", fg="white", font=("TkDefaultFont", 14, "bold"), padx=16, pady=6) self.status_label.pack(side="left") self.runtime_label = ttk.Label(bar, text="") self.runtime_label.pack(side="left", padx=12) ttk.Button(bar, text="Open log folder", command=self.open_log_folder).pack(side="right") ttk.Button(bar, text="Clear output", command=self.clear_output).pack(side="right", padx=4) self.stop_button = ttk.Button(bar, text="Stop", command=self.stop_monitor, state="disabled") self.stop_button.pack(side="right", padx=4) self.start_button = ttk.Button(bar, text="Start", command=self.start_monitor) self.start_button.pack(side="right", padx=4) panes = ttk.PanedWindow(parent, orient="vertical") panes.pack(fill="both", expand=True, padx=8, pady=(0, 8)) top = ttk.Frame(panes) panes.add(top, weight=2) top.columnconfigure(0, weight=1) top.columnconfigure(1, weight=1) top.rowconfigure(0, weight=1) info_frame = ttk.LabelFrame(top, text="Connection & settings in use") info_frame.grid(row=0, column=0, sticky="nsew", padx=(0, 4)) self.info_tree = ttk.Treeview(info_frame, columns=("value",), show="tree headings", height=10) self.info_tree.heading("#0", text="Parameter") self.info_tree.heading("value", text="Value") self.info_tree.column("#0", width=170, stretch=False) self.info_tree.column("value", width=380) info_scroll = ttk.Scrollbar(info_frame, orient="vertical", command=self.info_tree.yview) self.info_tree.configure(yscrollcommand=info_scroll.set) self.info_tree.pack(side="left", fill="both", expand=True) info_scroll.pack(side="right", fill="y") self.info_items: dict[str, str] = {} targets_frame = ttk.LabelFrame(top, text="Targets") targets_frame.grid(row=0, column=1, sticky="nsew", padx=(4, 0)) columns = ("target", "status", "outages", "downtime", "last") self.status_tree = ttk.Treeview(targets_frame, columns=columns, show="tree headings", height=10) self.status_tree.heading("#0", text="Name") self.status_tree.column("#0", width=90, stretch=False) for column, title, width in ( ("target", "Target", 130), ("status", "Status", 80), ("outages", "Outages", 65), ("downtime", "Downtime", 95), ("last", "Last event", 140), ): self.status_tree.heading(column, text=title) self.status_tree.column(column, width=width) self.status_tree.tag_configure("UP", foreground="#1e8449") self.status_tree.tag_configure("FLAKY", foreground="#b9770e") self.status_tree.tag_configure("DOWN", foreground="#c0392b") self.status_tree.pack(fill="both", expand=True) output_frame = ttk.LabelFrame(panes, text="Output (same as terminal)") panes.add(output_frame, weight=3) self.output_text = ScrolledText(output_frame, wrap="word", height=14, font=("TkFixedFont", 10), state="disabled") self.output_text.pack(fill="both", expand=True) self.output_text.tag_configure("INFO", foreground="#1f618d") self.output_text.tag_configure("WARN", foreground="#b9770e") self.output_text.tag_configure("ERROR", foreground="#c0392b") self.output_text.tag_configure("OK", foreground="#1e8449") self.output_text.tag_configure("DIM", foreground="#7f8c8d") def set_status(self, status: str) -> None: self.status_label.configure(text=status, bg=STATUS_COLORS[status]) def set_info(self, key: str, value: str) -> None: if key in self.info_items: self.info_tree.item(self.info_items[key], values=(value,)) else: self.info_items[key] = self.info_tree.insert("", "end", text=key, values=(value,)) def update_target_row(self, name: str, **fields: object) -> None: row = self.target_rows.setdefault( name, {"target": "", "status": "-", "outages": 0, "downtime": 0, "last": ""} ) row.update(fields) values = ( row["target"], row["status"], row["outages"], pingmon.format_duration(row["downtime"]), row["last"], ) if self.status_tree.exists(name): self.status_tree.item(name, values=values, tags=(row["status"],)) else: self.status_tree.insert("", "end", iid=name, text=name, values=values, tags=(row["status"],)) def refresh_overall_status(self) -> None: if not self.proc: return if not self.target_rows: self.set_status("STARTING") elif not self.down_targets: self.set_status("ONLINE") elif len(self.down_targets) >= len(self.target_rows): self.set_status("OUTAGE") else: self.set_status("DEGRADED") def append_output(self, line: str, tag: str | None) -> None: self.output_text.configure(state="normal") self.output_text.insert("end", line + "\n", (tag,) if tag else ()) # Keep memory bounded for multi-day runs; the full history is in the log file anyway. if int(self.output_text.index("end-1c").split(".")[0]) > 5000: self.output_text.delete("1.0", "1000.0") self.output_text.configure(state="disabled") self.output_text.see("end") def clear_output(self) -> None: self.output_text.configure(state="normal") self.output_text.delete("1.0", "end") self.output_text.configure(state="disabled") def handle_line(self, line: str) -> None: stripped = line.strip() tag = None if stripped in SECTION_HEADERS and line.startswith(" "): self.section = stripped tag = "DIM" elif set(stripped) == {"-"} or set(stripped) == {"="}: if self.section == "Info": self.section = None tag = "DIM" elif self.section == "Targets" and (match := TARGET_RE.match(line)): self.update_target_row(match.group(1), target=match.group(2), status="UP") self.refresh_overall_status() elif self.section in ("General", "Connection") and (match := KV_RE.match(line)): self.set_info(match.group(1), match.group(2)) elif self.section == "Info" and (match := DATA_KV_RE.match(line)): self.set_info(f"Data volume: {match.group(1)}", match.group(2)) elif self.section == "Parameter legend": tag = "DIM" elif event := EVENT_RE.match(line): tag = event.group(1) tag = self.handle_event(event.group(2), tag) elif stripped.startswith("ERROR"): tag = "ERROR" self.append_output(line, tag) def handle_event(self, message: str, tag: str) -> str: now = time.strftime("%H:%M:%S") if match := FAILURE_RE.match(message): name = match.group(1) if not self.down_targets: self.total_outages += 1 self.set_info("Outages (combined)", str(self.total_outages)) self.down_targets.add(name) self.update_target_row(name, status="DOWN", last=f"{now} failure") elif match := RECOVERED_RE.match(message): name = match.group(1) hours, minutes, seconds = (int(match.group(i)) for i in (3, 4, 5)) row = self.target_rows.get(name, {"outages": 0, "downtime": 0}) self.down_targets.discard(name) self.update_target_row( name, status="UP", outages=row["outages"] + 1, downtime=row["downtime"] + hours * 3600 + minutes * 60 + seconds, last=f"{now} recovered", ) tag = "OK" elif match := FLAKY_RE.match(message): name = match.group(1) if name not in self.down_targets: self.update_target_row(name, status="FLAKY", last=f"{now} ping failed") elif match := CHANGED_RE.match(message): self.set_info("Provider", match.group(2)) self.set_info("Profile", match.group(3)) tag = "WARN" elif match := SPEED_RE.search(message): self.set_info("Speed test", f"down={match.group(1)} Mbps, up={match.group(2)} Mbps ({now})") self.refresh_overall_status() return tag def poll_output(self) -> None: try: while True: line = self.output.get_nowait() if line is None: self.on_process_exit() break self.handle_line(line.rstrip("\n")) except queue.Empty: pass self.after(150, self.poll_output) def tick(self) -> None: if self.proc and self.started_at: self.runtime_label.configure( text=f"Running for {pingmon.format_duration(int(time.time() - self.started_at))}" ) self.after(1000, self.tick) def start_monitor(self) -> None: if self.proc: return config_path = Path(self.config_path_var.get()).expanduser() try: pending = self.collect_config() except ValueError as error: messagebox.showerror("Invalid settings", str(error)) return if pending != self.raw_config or not config_path.exists(): answer = messagebox.askyesnocancel( "Unsaved settings", "Settings differ from the config file. Save them before starting?" ) if answer is None: return if answer and not self.save_config(): return self.info_tree.delete(*self.info_tree.get_children()) self.info_items.clear() self.status_tree.delete(*self.status_tree.get_children()) self.target_rows.clear() self.down_targets.clear() self.total_outages = 0 self.section = None command = [sys.executable, "-u", *MONITOR_COMMAND, str(config_path)] kwargs: dict = {} if os.name == "nt": kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP try: self.proc = subprocess.Popen( command, cwd=SCRIPT_DIR, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, **kwargs, ) except OSError as error: messagebox.showerror("Start failed", str(error)) self.proc = None return threading.Thread(target=self.read_process_output, args=(self.proc,), daemon=True).start() self.started_at = time.time() self.start_button.configure(state="disabled") self.stop_button.configure(state="normal") self.set_status("STARTING") self.append_output( "(Startup runs ISP lookup and a baseline speed test first - this can take ~20s.)", "DIM" ) def read_process_output(self, proc: subprocess.Popen) -> None: assert proc.stdout is not None for line in proc.stdout: self.output.put(line) proc.wait() self.output.put(None) def stop_monitor(self) -> None: if not self.proc or self.proc.poll() is not None: return self.stop_button.configure(state="disabled") self.append_output("Stopping (waits for the current check to finish)...", "DIM") # SIGINT lets pingmon finish the current cycle and print its summary + close the CSV cleanly. if os.name == "nt": self.proc.send_signal(signal.CTRL_BREAK_EVENT) else: self.proc.send_signal(signal.SIGINT) self.after(15000, self.force_kill, self.proc) def force_kill(self, proc: subprocess.Popen) -> None: if proc.poll() is None: proc.kill() def on_process_exit(self) -> None: code = self.proc.returncode if self.proc else None self.proc = None self.started_at = None self.start_button.configure(state="normal") self.stop_button.configure(state="disabled") self.set_status("STOPPED") self.append_output(f"Monitor process exited (code {code}).", "DIM") def open_log_folder(self) -> None: log_dir = Path(self.raw_config.get("log_dir", "logs")).expanduser() if not log_dir.is_absolute(): log_dir = SCRIPT_DIR / log_dir log_dir.mkdir(parents=True, exist_ok=True) if sys.platform == "darwin": subprocess.Popen(["open", str(log_dir)]) elif os.name == "nt": os.startfile(log_dir) # type: ignore[attr-defined] else: subprocess.Popen(["xdg-open", str(log_dir)]) def on_close(self) -> None: if self.proc and self.proc.poll() is None: if not messagebox.askokcancel("Quit", "Monitoring is running. Stop it and quit?"): return self.proc.send_signal(signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGINT) try: self.proc.wait(timeout=5) except subprocess.TimeoutExpired: self.proc.kill() self.destroy() # ----------------------------------------------------------------- settings tab def build_settings_tab(self, parent: ttk.Frame) -> None: top = ttk.Frame(parent) top.pack(fill="x", padx=8, pady=8) ttk.Label(top, text="Config file:").pack(side="left") ttk.Entry(top, textvariable=self.config_path_var).pack(side="left", fill="x", expand=True, padx=4) ttk.Button(top, text="Browse...", command=self.browse_config).pack(side="left") ttk.Button(top, text="Reload", command=self.load_config).pack(side="left", padx=2) ttk.Button(top, text="Load defaults", command=self.load_defaults).pack(side="left", padx=2) ttk.Button(top, text="Save", command=self.save_config).pack(side="left", padx=2) self.settings_message = ttk.Label(parent, text="", foreground="#1e8449") self.settings_message.pack(fill="x", padx=8) body = ttk.Frame(parent) body.pack(fill="both", expand=True, padx=8, pady=(0, 8)) body.columnconfigure(0, weight=1) body.columnconfigure(1, weight=1) body.rowconfigure(1, weight=1) general = ttk.LabelFrame(body, text="General") general.grid(row=0, column=0, sticky="nsew", padx=(0, 4), pady=(0, 6)) self.var_isp_interval = tk.StringVar() self.var_speed_interval = tk.StringVar() self.var_speed_on_change = tk.BooleanVar() self.var_verbose = tk.BooleanVar() self.var_log_dir = tk.StringVar() self.var_log_prefix = tk.StringVar() self.var_logs_keep = tk.StringVar() self.form_row(general, 0, "ISP check interval (s)", self.var_isp_interval, "re-check provider to catch silent failovers") self.form_row(general, 1, "Speed test every (min)", self.var_speed_interval, "0 = disabled") ttk.Checkbutton(general, text="Speed test on connection change", variable=self.var_speed_on_change).grid( row=2, column=0, columnspan=3, sticky="w", padx=6, pady=2 ) ttk.Checkbutton(general, text="Verbose downtime (log every failed ping)", variable=self.var_verbose).grid( row=3, column=0, columnspan=3, sticky="w", padx=6, pady=2 ) self.form_row(general, 4, "Log directory", self.var_log_dir, "relative to the app folder") self.form_row(general, 5, "Log file prefix", self.var_log_prefix, "") self.form_row(general, 6, "Log files to keep", self.var_logs_keep, "oldest pruned on start") overrides = ttk.LabelFrame(body, text="Ping overrides (empty = use connection profile)") overrides.grid(row=0, column=1, sticky="nsew", padx=(4, 0), pady=(0, 6)) self.var_interval = tk.StringVar() self.var_timeout = tk.StringVar() self.var_threshold = tk.StringVar() self.form_row(overrides, 0, "Interval (s)", self.var_interval, "seconds between ping checks") self.form_row(overrides, 1, "Ping timeout (s)", self.var_timeout, "wait for a reply") self.form_row(overrides, 2, "Failure threshold", self.var_threshold, "consecutive fails = down") ttk.Label( overrides, text="Values set here apply to every provider and disable\nthe automatic per-profile values below.", foreground="#7f8c8d", ).grid(row=3, column=0, columnspan=3, sticky="w", padx=6, pady=6) self.build_targets_editor(body) self.build_profiles_editor(body) def form_row(self, parent: ttk.Frame, row: int, label: str, var: tk.StringVar, hint: str) -> None: ttk.Label(parent, text=label).grid(row=row, column=0, sticky="w", padx=6, pady=2) ttk.Entry(parent, textvariable=var, width=18).grid(row=row, column=1, sticky="w", padx=6, pady=2) ttk.Label(parent, text=hint, foreground="#7f8c8d").grid(row=row, column=2, sticky="w", padx=6, pady=2) def build_targets_editor(self, body: ttk.Frame) -> None: frame = ttk.LabelFrame(body, text="Ping targets") frame.grid(row=1, column=0, sticky="nsew", padx=(0, 4)) frame.columnconfigure(1, weight=1) frame.rowconfigure(0, weight=1) self.targets_tree = ttk.Treeview(frame, columns=("host",), show="tree headings", height=6) self.targets_tree.heading("#0", text="Name") self.targets_tree.heading("host", text="Host / IP") self.targets_tree.column("#0", width=120, stretch=False) self.targets_tree.grid(row=0, column=0, columnspan=2, sticky="nsew", padx=6, pady=6) self.targets_tree.bind("<>", self.on_target_select) self.var_target_name = tk.StringVar() self.var_target_host = tk.StringVar() ttk.Label(frame, text="Name").grid(row=1, column=0, sticky="w", padx=6) ttk.Entry(frame, textvariable=self.var_target_name).grid(row=1, column=1, sticky="ew", padx=6, pady=2) ttk.Label(frame, text="Host / IP").grid(row=2, column=0, sticky="w", padx=6) ttk.Entry(frame, textvariable=self.var_target_host).grid(row=2, column=1, sticky="ew", padx=6, pady=2) buttons = ttk.Frame(frame) buttons.grid(row=3, column=0, columnspan=2, sticky="w", padx=6, pady=4) ttk.Button(buttons, text="Add / Update", command=self.upsert_target).pack(side="left") ttk.Button(buttons, text="Remove", command=self.remove_target).pack(side="left", padx=4) ttk.Label( frame, text="Tip: target 'gateway' with empty host = auto-detect your router.\n" "Target 'dns' is compared with your system resolver (DNS chain check).", foreground="#7f8c8d", ).grid(row=4, column=0, columnspan=2, sticky="w", padx=6, pady=(0, 6)) def build_profiles_editor(self, body: ttk.Frame) -> None: frame = ttk.LabelFrame(body, text="Connection profiles (auto-selected by ISP name)") frame.grid(row=1, column=1, sticky="nsew", padx=(4, 0)) frame.columnconfigure(1, weight=1) frame.rowconfigure(0, weight=1) columns = ("match", "interval", "timeout", "threshold") self.profiles_tree = ttk.Treeview(frame, columns=columns, show="tree headings", height=6) self.profiles_tree.heading("#0", text="Profile") self.profiles_tree.column("#0", width=90, stretch=False) for column, title, width in ( ("match", "Match keywords", 200), ("interval", "Interval", 60), ("timeout", "Timeout", 60), ("threshold", "Threshold", 70), ): self.profiles_tree.heading(column, text=title) self.profiles_tree.column(column, width=width, stretch=column == "match") self.profiles_tree.grid(row=0, column=0, columnspan=2, sticky="nsew", padx=6, pady=6) self.profiles_tree.bind("<>", self.on_profile_select) self.var_profile_name = tk.StringVar() self.var_profile_match = tk.StringVar() self.var_profile_interval = tk.StringVar() self.var_profile_timeout = tk.StringVar() self.var_profile_threshold = tk.StringVar() for row, (label, var) in enumerate( ( ("Name", self.var_profile_name), ("Match keywords", self.var_profile_match), ("Interval (s)", self.var_profile_interval), ("Ping timeout (s)", self.var_profile_timeout), ("Failure threshold", self.var_profile_threshold), ), start=1, ): ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", padx=6) ttk.Entry(frame, textvariable=var).grid(row=row, column=1, sticky="ew", padx=6, pady=2) buttons = ttk.Frame(frame) buttons.grid(row=6, column=0, columnspan=2, sticky="w", padx=6, pady=4) ttk.Button(buttons, text="Add / Update", command=self.upsert_profile).pack(side="left") ttk.Button(buttons, text="Remove", command=self.remove_profile).pack(side="left", padx=4) ttk.Label( frame, text='Keywords: comma-separated, quote them to keep spaces, e.g. "o2 ", " lte".\n' "Matched case-insensitively against the ISP name; 'default' is used otherwise.", foreground="#7f8c8d", ).grid(row=7, column=0, columnspan=2, sticky="w", padx=6, pady=(0, 6)) def refresh_targets_tree(self) -> None: self.targets_tree.delete(*self.targets_tree.get_children()) for name, host in self.targets.items(): self.targets_tree.insert("", "end", iid=name, text=name, values=(host or "(auto-detect)",)) def refresh_profiles_tree(self) -> None: self.profiles_tree.delete(*self.profiles_tree.get_children()) for name, profile in self.profiles.items(): self.profiles_tree.insert( "", "end", iid=name, text=name, values=( format_match(profile.get("match", [])), profile.get("interval", ""), profile.get("ping_timeout", ""), profile.get("failure_threshold", ""), ), ) def on_target_select(self, _event: object) -> None: selection = self.targets_tree.selection() if selection: name = selection[0] self.var_target_name.set(name) self.var_target_host.set(self.targets.get(name, "")) def on_profile_select(self, _event: object) -> None: selection = self.profiles_tree.selection() if selection: name = selection[0] profile = self.profiles.get(name, {}) self.var_profile_name.set(name) self.var_profile_match.set(format_match(profile.get("match", []))) self.var_profile_interval.set(str(profile.get("interval", ""))) self.var_profile_timeout.set(str(profile.get("ping_timeout", ""))) self.var_profile_threshold.set(str(profile.get("failure_threshold", ""))) def upsert_target(self) -> None: name = self.var_target_name.get().strip() if not name: messagebox.showerror("Target", "Name is required.") return self.targets[name] = self.var_target_host.get().strip() self.refresh_targets_tree() def remove_target(self) -> None: for name in self.targets_tree.selection(): self.targets.pop(name, None) self.refresh_targets_tree() def upsert_profile(self) -> None: name = self.var_profile_name.get().strip() if not name: messagebox.showerror("Profile", "Name is required.") return try: profile = { "match": parse_match(self.var_profile_match.get()), "interval": self.parse_int(self.var_profile_interval.get(), "Interval", 1), "ping_timeout": self.parse_int(self.var_profile_timeout.get(), "Ping timeout", 1), "failure_threshold": self.parse_int(self.var_profile_threshold.get(), "Failure threshold", 1), } except ValueError as error: messagebox.showerror("Profile", str(error)) return self.profiles[name] = profile self.refresh_profiles_tree() def remove_profile(self) -> None: for name in self.profiles_tree.selection(): if name == "default": messagebox.showinfo("Profile", "The 'default' profile is required and can't be removed.") continue self.profiles.pop(name, None) self.refresh_profiles_tree() @staticmethod def parse_int(text: str, label: str, minimum: int) -> int: try: value = int(str(text).strip()) except ValueError: raise ValueError(f"{label} must be a whole number.") from None if value < minimum: raise ValueError(f"{label} must be at least {minimum}.") return value def browse_config(self) -> None: path = filedialog.askopenfilename( initialdir=str(Path(self.config_path_var.get()).expanduser().parent), filetypes=[("JSON config", "*.json"), ("All files", "*")], ) if path: self.config_path_var.set(path) self.load_config() def load_defaults(self) -> None: self.populate_form(json.loads(json.dumps(pingmon.DEFAULT_CONFIG))) self.settings_message.configure(text="Defaults loaded - click Save to write them to the config file.") def load_config(self) -> None: path = Path(self.config_path_var.get()).expanduser() if not path.exists(): self.raw_config = {} self.populate_form(json.loads(json.dumps(pingmon.DEFAULT_CONFIG))) self.settings_message.configure(text=f"{path} does not exist yet - showing defaults, Save to create it.") return try: config = json.loads(path.read_text()) except (OSError, json.JSONDecodeError) as error: messagebox.showerror("Config", f"Could not read {path}:\n{error}") return self.raw_config = config self.populate_form(config) self.settings_message.configure(text=f"Loaded {path}") def populate_form(self, config: dict) -> None: self.targets = dict(config.get("targets", {})) profiles = config.get("connection_profiles") or pingmon.DEFAULT_CONNECTION_PROFILES self.profiles = json.loads(json.dumps(profiles)) self.profiles.setdefault("default", dict(pingmon.DEFAULT_CONNECTION_PROFILES["default"])) self.var_isp_interval.set(str(config.get("isp_check_interval", 15))) self.var_speed_interval.set(str(config.get("speed_test_interval_minutes", 60))) self.var_speed_on_change.set(bool(config.get("speed_test_on_connection_change", True))) self.var_verbose.set(bool(config.get("verbose_downtime", True))) self.var_log_dir.set(str(config.get("log_dir", "logs"))) self.var_log_prefix.set(str(config.get("log_prefix", "ping_monitor"))) self.var_logs_keep.set(str(config.get("logfiles_to_keep", 10))) self.var_interval.set(str(config["interval"]) if "interval" in config else "") self.var_timeout.set(str(config["ping_timeout"]) if "ping_timeout" in config else "") self.var_threshold.set(str(config["failure_threshold"]) if "failure_threshold" in config else "") self.refresh_targets_tree() self.refresh_profiles_tree() def collect_config(self) -> dict: if not self.targets: raise ValueError("At least one ping target is required.") config = dict(self.raw_config) # keeps keys this GUI doesn't know about config["targets"] = dict(self.targets) config["connection_profiles"] = json.loads(json.dumps(self.profiles)) config["isp_check_interval"] = self.parse_int(self.var_isp_interval.get(), "ISP check interval", 1) config["speed_test_on_connection_change"] = bool(self.var_speed_on_change.get()) config["speed_test_interval_minutes"] = self.parse_int(self.var_speed_interval.get(), "Speed test interval", 0) config["log_dir"] = self.var_log_dir.get().strip() or "logs" config["log_prefix"] = self.var_log_prefix.get().strip() or "ping_monitor" config["logfiles_to_keep"] = self.parse_int(self.var_logs_keep.get(), "Log files to keep", 1) config["verbose_downtime"] = bool(self.var_verbose.get()) for key, var, label in ( ("interval", self.var_interval, "Interval override"), ("ping_timeout", self.var_timeout, "Ping timeout override"), ("failure_threshold", self.var_threshold, "Failure threshold override"), ): if var.get().strip(): config[key] = self.parse_int(var.get(), label, 1) else: config.pop(key, None) return config def save_config(self) -> bool: try: config = self.collect_config() except ValueError as error: messagebox.showerror("Invalid settings", str(error)) return False path = Path(self.config_path_var.get()).expanduser() try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(config, indent=2) + "\n") except OSError as error: messagebox.showerror("Save failed", str(error)) return False self.raw_config = config note = " (restart monitoring to apply)" if self.proc else "" self.settings_message.configure(text=f"Saved {path} at {time.strftime('%H:%M:%S')}{note}") return True RELAUNCH_ENV = "ISP_CHECK_RELAUNCHED" PYTHON_CANDIDATES = ( "/opt/homebrew/bin/python3", "/usr/local/bin/python3", *sorted(glob.glob("/Library/Frameworks/Python.framework/Versions/3.*/bin/python3"), reverse=True), *sorted(glob.glob("/opt/homebrew/bin/python3.*[0-9]"), reverse=True), ) def find_python_with_modern_tk() -> str | None: current = os.path.realpath(sys.executable) for candidate in PYTHON_CANDIDATES: if not os.access(candidate, os.X_OK) or os.path.realpath(candidate) == current: continue try: result = subprocess.run( [candidate, "-c", "import sys, tkinter; sys.exit(0 if tkinter.TkVersion >= 8.6 else 1)"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15, ) except (OSError, subprocess.TimeoutExpired): continue if result.returncode == 0: return candidate return None def check_tk_version() -> bool: # Apple's /usr/bin/python3 bundles Tk 8.5, which renders an empty black window on current macOS. patchlevel = tk.Tcl().eval("info patchlevel") if tk.TkVersion >= 8.6: return True if not os.environ.get(RELAUNCH_ENV): better = find_python_with_modern_tk() if better: print(f"Tk {patchlevel} is too old, restarting with {better}", file=sys.stderr) os.environ[RELAUNCH_ENV] = "1" os.execv(better, [better, *sys.argv]) print( f"ERROR: Tk {patchlevel} is too old for the GUI (needs 8.6+), the window would stay blank.\n" f"Python used: {sys.executable}\n" "Fix on macOS: install Python from https://www.python.org/downloads/ and start with\n" " /usr/local/bin/python3 isp-connection-check.pyz\n" "or with Homebrew: brew install python python-tk && $(brew --prefix)/bin/python3 isp-connection-check.pyz\n" "The command line version works with any Python: python3 isp-connection-check.pyz --cli", file=sys.stderr, ) return False def main() -> int: if not check_tk_version(): return 1 config_path = Path(sys.argv[1]).expanduser() if len(sys.argv) > 1 else DEFAULT_CONFIG_PATH if not config_path.is_absolute(): config_path = Path.cwd() / config_path app = PingmonGui(config_path) app.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main()) PK;]Gc __main__.pyPK;]߃uu Zpingmon.pyPK˅;]WR[wpingmon_gui.pyPK1