From 2d0054a8f1081963000aaeaa16e982d9003d9c14 Mon Sep 17 00:00:00 2001 From: Ben Nicholson Date: Wed, 22 Jul 2026 14:31:46 +1000 Subject: [PATCH] Upload files to "watch_remote_top" --- watch_remote_top/build_exe.bat | 30 +++ watch_remote_top/watch_remote_top.py | 300 +++++++++++++++++++++++++ watch_remote_top/watch_remote_top.spec | 45 ++++ 3 files changed, 375 insertions(+) create mode 100644 watch_remote_top/build_exe.bat create mode 100644 watch_remote_top/watch_remote_top.py create mode 100644 watch_remote_top/watch_remote_top.spec diff --git a/watch_remote_top/build_exe.bat b/watch_remote_top/build_exe.bat new file mode 100644 index 0000000..36e879d --- /dev/null +++ b/watch_remote_top/build_exe.bat @@ -0,0 +1,30 @@ +@echo off +REM Builds watch_remote_top.py into a standalone Windows .exe using PyInstaller. +REM Run this ON WINDOWS, in the same folder as watch_remote_top.py. + +echo Installing/updating build dependencies (pyinstaller, paramiko)... +pip install --upgrade pyinstaller paramiko +if errorlevel 1 ( + echo pip install failed. Make sure Python and pip are installed and on PATH. + pause + exit /b 1 +) + +echo. +echo Building watch_remote_top.exe ... +REM --onefile : single .exe, no extra files needed alongside it +REM --console : keep the console window (needed for input()/getpass prompts) +REM --collect-all paramiko : bundles paramiko's crypto backends, which PyInstaller +REM sometimes misses with plain --hidden-import +pyinstaller --onefile --console --name watch_remote_top --collect-all paramiko watch_remote_top.py + +if errorlevel 1 ( + echo Build failed - see errors above. + pause + exit /b 1 +) + +echo. +echo Done. Executable is at: dist\watch_remote_top.exe +echo You can copy that single file anywhere - no Python install needed to run it. +pause diff --git a/watch_remote_top/watch_remote_top.py b/watch_remote_top/watch_remote_top.py new file mode 100644 index 0000000..17f0bfa --- /dev/null +++ b/watch_remote_top/watch_remote_top.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 +""" +watch_remote_top.py + +Prompts for a remote host/IP, SSH username, and password (interactively, +password not echoed), then every 10 minutes: + - SSHes in and runs `top -bn1 -o %MEM` (one-shot, non-interactive, sorted by RAM) + - Saves the full raw output to a local log file + - Parses out CPU usage, RAM usage, and the top RAM-consuming process into a + local CSV summary + +Everything is saved on THIS machine, not the remote host. + +Requires: paramiko + pip install paramiko + +Usage: + python3 watch_remote_top.py + python3 watch_remote_top.py --once # single run, no loop + python3 watch_remote_top.py --interval 5 # poll every 5 minutes instead of 10 + python3 watch_remote_top.py --log-dir ./logs # where to save logs (default: ./TopLogs) +""" + +import argparse +import csv +import getpass +import re +import sys +import time +from datetime import datetime +from pathlib import Path + +try: + import paramiko +except ImportError: + sys.exit( + "The 'paramiko' package is required but not installed.\n" + "Install it with: pip install paramiko" + ) + +# Plain `-bn1` (no -o): BusyBox's top doesn't support -o for sorting, so we +# capture unsorted output and sort by memory ourselves in parse_top_output(). +REMOTE_COMMAND = "top -bn1" + + +def get_connection_details(): + host = input("Remote host/IP: ").strip() + username = input("SSH username: ").strip() + password = getpass.getpass("SSH password: ") + return host, username, password + + +def run_remote_top(host, username, password, timeout=10): + """SSH in, run the top command once, return the raw stdout as a string.""" + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect( + hostname=host, + username=username, + password=password, + timeout=timeout, + banner_timeout=timeout, + auth_timeout=timeout, + ) + stdin, stdout, stderr = client.exec_command(REMOTE_COMMAND, timeout=timeout) + output = stdout.read().decode(errors="replace") + err = stderr.read().decode(errors="replace") + if not output and err: + raise RuntimeError(f"Remote command produced no output. stderr: {err.strip()}") + return output + finally: + client.close() + + +def _to_mib(value, unit): + """Convert a K/M/G-suffixed value (as used by BusyBox's `Mem:` line) to MiB.""" + value = float(value) + unit = (unit or "").upper() + if unit.startswith("K"): + return value / 1024 + if unit.startswith("G"): + return value * 1024 + return value # already M / MiB, or no unit + + +def parse_top_output(raw_output): + """ + Pull CPU used %, mem used/total/free (MiB), mem used %, and the top + RAM-consuming process out of raw `top -bn1` output. Handles both the + standard Linux (procps) format and BusyBox's leaner top format, since + the remote command has no reliable way to sort by memory itself + (BusyBox top doesn't support -o), so the top process is found by + scanning and sorting the process rows here instead. + """ + lines = raw_output.splitlines() + + cpu_used = cpu_idle = None + mem_total = mem_used = mem_free = mem_used_pct = None + top_process = None + + is_procps = any("%Cpu(s)" in line for line in lines) + is_busybox = any(re.match(r"^CPU:", line.strip()) for line in lines) + + if is_procps: + for line in lines: + if "%Cpu(s)" in line: + m = re.search(r"([\d.]+)\s*id", line) + if m: + cpu_idle = float(m.group(1)) + cpu_used = round(100 - cpu_idle, 1) + + if re.search(r"(MiB|KiB|GiB)\s+Mem", line): + m_total = re.search(r"([\d.]+)\s+total", line) + m_free = re.search(r"([\d.]+)\s+free", line) + m_used = re.search(r"([\d.]+)\s+used", line) + if m_total: + mem_total = float(m_total.group(1)) + if m_free: + mem_free = float(m_free.group(1)) + if m_used: + mem_used = float(m_used.group(1)) + if mem_total and mem_used: + mem_used_pct = round((mem_used / mem_total) * 100, 1) + + # PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND + header_idx, mem_col_idx = None, None + for i, line in enumerate(lines): + if re.match(r"^\s*PID\s+USER", line): + header_cols = line.split() + if "%MEM" in header_cols: + header_idx = i + mem_col_idx = header_cols.index("%MEM") + break + + if header_idx is not None and mem_col_idx is not None: + best_pct = -1.0 + for line in lines[header_idx + 1:]: + cols = line.split() + if len(cols) <= mem_col_idx: + continue + try: + pct = float(cols[mem_col_idx]) + except ValueError: + continue + if pct > best_pct: + best_pct = pct + top_process = f"{cols[-1]} ({pct}% MEM)" + + elif is_busybox: + # e.g. "Mem: 51380K used, 178308K free, 0K shrd, 3844K buff, 21520K cached" + for line in lines: + m = re.search( + r"Mem:\s*([\d.]+)\s*([KMG])?\s*used,\s*([\d.]+)\s*([KMG])?\s*free", + line, + re.IGNORECASE, + ) + if m: + mem_used = round(_to_mib(m.group(1), m.group(2)), 1) + mem_free = round(_to_mib(m.group(3), m.group(4)), 1) + mem_total = round(mem_used + mem_free, 1) + if mem_total: + mem_used_pct = round((mem_used / mem_total) * 100, 1) + + # e.g. "CPU: 2.3% usr 1.1% sys 0.0% nic 96.4% idle ..." + m_idle = re.search(r"([\d.]+)%?\s*idle", line, re.IGNORECASE) + if m_idle: + cpu_idle = float(m_idle.group(1)) + cpu_used = round(100 - cpu_idle, 1) + + # e.g. " PID PPID USER STAT VSZ %VSZ %CPU COMMAND" + header_idx, mem_col_idx = None, None + for i, line in enumerate(lines): + cols = line.split() + if cols[:1] == ["PID"] and "USER" in cols: + header_idx = i + if "%VSZ" in cols: + mem_col_idx = cols.index("%VSZ") + elif "%MEM" in cols: + mem_col_idx = cols.index("%MEM") + break + + if header_idx is not None and mem_col_idx is not None: + best_pct = -1.0 + for line in lines[header_idx + 1:]: + cols = line.split() + if len(cols) <= mem_col_idx: + continue + try: + pct = float(cols[mem_col_idx].rstrip("%")) + except ValueError: + continue + if pct > best_pct: + best_pct = pct + top_process = f"{cols[-1]} ({pct}% VSZ)" + + return { + "cpu_used": cpu_used, + "cpu_idle": cpu_idle, + "mem_total": mem_total, + "mem_used": mem_used, + "mem_free": mem_free, + "mem_used_pct": mem_used_pct, + "top_process": top_process, + } + + +def ensure_csv_header(csv_path): + if not csv_path.exists(): + with open(csv_path, "w", newline="", encoding="utf-8") as f: + writer = csv.writer(f) + writer.writerow( + [ + "Timestamp", + "CPU_Used_Pct", + "CPU_Idle_Pct", + "Mem_Total_MiB", + "Mem_Used_MiB", + "Mem_Free_MiB", + "Mem_Used_Pct", + "Top_RAM_Process", + ] + ) + + +def collect_snapshot(host, username, password, full_log_path, summary_csv_path): + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[{timestamp}] Connecting to {host}...") + + try: + raw_output = run_remote_top(host, username, password) + except Exception as exc: + print(f"[{timestamp}] SSH/command failed: {exc}", file=sys.stderr) + with open(summary_csv_path, "a", newline="", encoding="utf-8") as f: + csv.writer(f).writerow([timestamp, "ERROR", "", "", "", "", "", f"SSH failed: {exc}"]) + return + + # Save full raw snapshot + with open(full_log_path, "a", encoding="utf-8") as f: + f.write(f"===== {timestamp} =====\n") + f.write(raw_output) + f.write("\n\n") + + parsed = parse_top_output(raw_output) + + with open(summary_csv_path, "a", newline="", encoding="utf-8") as f: + csv.writer(f).writerow( + [ + timestamp, + parsed["cpu_used"], + parsed["cpu_idle"], + parsed["mem_total"], + parsed["mem_used"], + parsed["mem_free"], + parsed["mem_used_pct"], + parsed["top_process"], + ] + ) + + print( + f"[{timestamp}] CPU used: {parsed['cpu_used']}% | " + f"Mem used: {parsed['mem_used_pct']}% " + f"({parsed['mem_used']}/{parsed['mem_total']} MiB) | " + f"Top RAM proc: {parsed['top_process']}" + ) + + +def main(): + parser = argparse.ArgumentParser(description="Poll a remote Linux host's `top` output over SSH and log it locally.") + parser.add_argument("--once", action="store_true", help="Run a single collection and exit instead of looping.") + parser.add_argument("--interval", type=int, default=10, help="Polling interval in minutes (default: 10).") + parser.add_argument("--log-dir", type=str, default="TopLogs", help="Local folder to save logs in (default: ./TopLogs).") + args = parser.parse_args() + + host, username, password = get_connection_details() + + log_dir = Path(args.log_dir) + log_dir.mkdir(parents=True, exist_ok=True) + full_log_path = log_dir / "top_full_snapshots.log" + summary_csv_path = log_dir / "cpu_ram_summary.csv" + ensure_csv_header(summary_csv_path) + + print(f"\nFull snapshots -> {full_log_path}") + print(f"Summary CSV -> {summary_csv_path}\n") + + if args.once: + collect_snapshot(host, username, password, full_log_path, summary_csv_path) + return + + print(f"Polling {host} every {args.interval} minute(s). Press Ctrl+C to stop.\n") + try: + while True: + collect_snapshot(host, username, password, full_log_path, summary_csv_path) + time.sleep(args.interval * 60) + except KeyboardInterrupt: + print("\nStopped.") + + +if __name__ == "__main__": + main() diff --git a/watch_remote_top/watch_remote_top.spec b/watch_remote_top/watch_remote_top.spec new file mode 100644 index 0000000..41b1e18 --- /dev/null +++ b/watch_remote_top/watch_remote_top.spec @@ -0,0 +1,45 @@ +# -*- mode: python ; coding: utf-8 -*- +from PyInstaller.utils.hooks import collect_all + +datas = [] +binaries = [] +hiddenimports = [] +tmp_ret = collect_all('paramiko') +datas += tmp_ret[0]; binaries += tmp_ret[1]; hiddenimports += tmp_ret[2] + + +a = Analysis( + ['watch_remote_top.py'], + pathex=[], + binaries=binaries, + datas=datas, + hiddenimports=hiddenimports, + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + noarchive=False, + optimize=0, +) +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.datas, + [], + name='watch_remote_top', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=True, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +)