@@ -0,0 +1,884 @@
#!/usr/bin/env python3
# =============================================================
# pi-stat
# Fast alternative to 'pi-node status'
# - Disk info directly via ZFS (no 'du', no long scan)
# - Peer data directly via stellar-core (no Horizon detour)
# =============================================================
import contextlib
import io
import ipaddress
import json
import os
import re
import select
import shutil
import socket
import subprocess
import sys
import time
from datetime import datetime, timezone
from urllib.request import urlopen, Request
from urllib.error import URLError
try:
import tty
import termios
HAS_TTY = True
except ImportError:
HAS_TTY = False
try:
import docker
except ImportError:
print("ERROR: 'docker' package missing. Install with: pip install docker")
sys.exit(1)
# --- Configuration ---
HAS_ZFS = shutil.which("zpool") is not None
CANDIDATE_CONTAINERS = ["mainnet", "testnet2", "testnet"]
HORIZON_URL = "http://localhost:31401"
CFG_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "stellar-core.cfg")
# --- Colors / Icons ---
GREEN = "\033[0;32m"
YELLOW = "\033[1;33m"
RED = "\033[0;31m"
BOLD = "\033[1m"
RESET = "\033[0m"
def ok(msg): print(f" {GREEN}✅{RESET} {msg}")
def warn(msg): print(f" {YELLOW}⚠️{RESET} {msg}")
def err(msg): print(f" {RED}❌{RESET} {msg}")
def hdr(title):
print(f"\n{BOLD}{title}{RESET}")
print("=" * len(title))
# --- Helper functions ---
def fmt_up(s):
s = int(s)
if s < 60: return f"{s}s"
if s < 3600: return f"{s//60}m{s%60:02d}s"
h = s // 3600; m = (s % 3600) // 60
return f"{h}h{m:02d}m"
def fmt_version(ver_str):
"""Extract vMAJOR.MINOR from a stellar-core version string."""
m = re.search(r'(\d+\.\d+)', ver_str or "")
return f"v{m.group(1)}" if m else "—"
def _f1(v):
s = f"{v:.1f}"
return s[:-2] if s.endswith('.0') else s
def fmt_bytes(b):
if b >= 1024**3: return f"{b/1024**3:.2f}GiB"
if b >= 1024**2: return f"{_f1(b/1024**2)}MiB"
return f"{b/1024:.0f}KiB"
def fmt_rate(bps):
if bps >= 1e9: return f"{bps/1e9:.2f}Gbps"
if bps >= 1e6: return f"{_f1(bps/1e6)}Mbps"
if bps >= 1e3: return f"{bps/1e3:.0f}kbps"
return f"{bps:.0f}bps"
def fmt_ratio_parts(rx, tx):
"""Return (left, right) tuple for aligned-colon display, or None if no data."""
if rx == 0 and tx == 0:
return None
if tx == 0:
return ("∞", "1")
if rx == 0:
return ("1", "∞")
r = rx / tx
def _fv(v): return _f1(v) if v < 10 else f"{v:.0f}"
if r >= 1:
return (_fv(r), "1")
else:
return ("1", _fv(1/r))
def fmt_ratio_cell(parts, l_w, r_w):
"""Render a ratio cell with the colon at a fixed column position."""
if parts is None:
return f"{'—':^{l_w + 1 + r_w}}"
l, r = parts
return f"{l:>{l_w}}:{r:<{r_w}}"
_geo_cache = {}
def geo_batch(ips):
"""Batch-fetch geolocation for multiple IPs via ip-api.com. Updates _geo_cache."""
to_fetch = [ip for ip in ips if ip not in _geo_cache and ip and ip != "—"]
if not to_fetch:
return
try:
payload = json.dumps([{"query": ip} for ip in to_fetch[:100]]).encode()
req = Request(
"http://ip-api.com/batch?fields=status,query,countryCode,city",
data=payload,
headers={"Content-Type": "application/json"},
)
with urlopen(req, timeout=5) as resp:
for item in json.loads(resp.read()):
ip = item.get("query")
if item.get("status") == "success":
_geo_cache[ip] = (item.get("countryCode", ""), item.get("city", ""))
else:
_geo_cache[ip] = None
except Exception:
for ip in to_fetch:
_geo_cache[ip] = None
def geo_label(ip):
"""Format IP with geolocation info."""
geo = _geo_cache.get(ip)
if geo:
return f"{ip} {', '.join(p for p in geo if p)}"
return ip
def geo_str(ip):
"""Return geo string (e.g. 'DE, Berlin') for an IP, or empty string if unavailable."""
geo = _geo_cache.get(ip)
if geo:
return ", ".join(p for p in geo if p)
return ""
def load_validators(cfg_path):
"""Parse [[VALIDATORS]] sections, returning {ip_or_name: (index, public_key)}."""
if not os.path.exists(cfg_path):
print(f"ERROR: stellar-core.cfg not found at {cfg_path}")
sys.exit(1)
result = {}
idx = 0
current_ip = current_key = current_name = None
try:
with open(cfg_path) as f:
for line in f:
if re.match(r'^\s*\[\[VALIDATORS\]\]', line):
if current_key:
idx += 1
if current_ip: result[current_ip] = (idx, current_key)
if current_name: result[current_name] = (idx, current_key)
result[current_key] = (idx, current_key)
current_ip = current_key = current_name = None
m = re.match(r'^\s*ADDRESS\s*=\s*"([^":]+)', line)
if m: current_ip = m.group(1)
m = re.match(r'^\s*PUBLIC_KEY\s*=\s*"([^"]+)"', line)
if m: current_key = m.group(1)
m = re.match(r'^\s*NAME\s*=\s*"([^"]+)"', line)
if m: current_name = m.group(1)
if current_key:
idx += 1
if current_ip: result[current_ip] = (idx, current_key)
if current_name: result[current_name] = (idx, current_key)
result[current_key] = (idx, current_key)
except OSError:
pass
return result
def zfs_get(cmd):
r = subprocess.run(cmd, capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else None
def detect_container(client):
"""Return (container, name) for the first running candidate, or first found."""
found = None
for name in CANDIDATE_CONTAINERS:
try:
c = client.containers.get(name)
if c.status == "running":
return c, name
if found is None:
found = (c, name)
except docker.errors.NotFound:
pass
return found if found else (None, None)
def exec_json(container, command):
"""Executes a stellar-core http-command and returns the parsed JSON."""
try:
_, (stdout, _stderr) = container.exec_run(command, demux=True)
if not stdout:
return None
raw = stdout.decode()
idx = raw.find('{')
if idx >= 0:
obj, _ = json.JSONDecoder().raw_decode(raw, idx)
return obj
except Exception:
pass
return None
def run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, extended_traffic=False):
"""Collect data and render one complete status screen.
Returns (outbound_numbered, container) where outbound_numbered is a list of
dicts [{id, ip, label, location}] for the interactive drop workflow.
"""
# --- Pre-load data ---
container, container_name = detect_container(client)
if container_name is None:
container_name = CANDIDATE_CONTAINERS[0]
core_info = None
if container:
raw = exec_json(container, "stellar-core http-command info")
if raw:
core_info = raw.get("info")
hz_data = None
try:
with urlopen(HORIZON_URL + "/", timeout=5) as resp:
hz_data = json.loads(resp.read())
except Exception:
pass
# =============================================================
# 1. CONTAINER STATUS
# =============================================================
if not peers_extended:
hdr("🐳 Container Status")
if container is None:
err(f"Container '{container_name}' not found")
else:
if container.status == "running":
ok(f"Container: {container_name} (Running)")
else:
err(f"Container: {container_name} ({container.status})")
try:
s = container.stats(stream=False)
cpu_delta = (s["cpu_stats"]["cpu_usage"]["total_usage"]
- s["precpu_stats"]["cpu_usage"]["total_usage"])
sys_delta = (s["cpu_stats"]["system_cpu_usage"]
- s["precpu_stats"]["system_cpu_usage"])
num_cpus = (s["cpu_stats"].get("online_cpus")
or len(s["cpu_stats"]["cpu_usage"].get("percpu_usage", [1])))
cpu_pct = (cpu_delta / sys_delta) * num_cpus * 100.0 if sys_delta > 0 else 0.0
mem_use = s["memory_stats"]["usage"]
mem_lim = s["memory_stats"]["limit"]
mem_str = f"{fmt_bytes(mem_use)} / {fmt_bytes(mem_lim)}"
print(f" CPU: {cpu_pct:.2f}% | RAM: {mem_str}")
except Exception:
print(" Stats not available")
if hz_data:
print(f" Horizon: {hz_data.get('horizon_version', '?')}")
print(f" Core: {hz_data.get('core_version', '?')}")
# =============================================================
# 2. DISK
# =============================================================
if not HAS_ZFS:
hdr("💾 Disk")
try:
path = (os.path.splitdrive(os.path.abspath(__file__))[0] + "\\"
if sys.platform == "win32" else "/")
usage = shutil.disk_usage(path)
cap_pct = f"{usage.used / usage.total * 100:.0f}%"
ok(f"Path: {path}")
print(f" Size: {fmt_bytes(usage.total)} | Used: {fmt_bytes(usage.used)} ({cap_pct}) | Free: {fmt_bytes(usage.free)}")
except Exception as e:
err(f"Disk info not available: {e}")
else:
hdr("💾 Disk (ZFS: z01pool)")
zfs_line = zfs_get(["zpool", "list", "-H", "-o",
"health,size,alloc,free,cap,frag", "z01pool"])
if not zfs_line:
err("Pool z01pool not found (or missing permissions)")
else:
health, size, alloc, free, cap, frag = zfs_line.split("\t")
compress = zfs_get(["zfs", "get", "-H", "-o", "value", "compression", "z01pool"]) or "?"
ratio = zfs_get(["zfs", "get", "-H", "-o", "value", "compressratio", "z01pool"]) or "?"
mount = zfs_get(["zfs", "get", "-H", "-o", "value", "mountpoint", "z01pool"]) or "?"
if health == "ONLINE":
ok(f"Health: {health}")
else:
err(f"Health: {health}")
print(f" Size: {size} | Used: {alloc} ({cap}) | Free: {free}")
print(f" Fragmentation: {frag}")
print(f" Compression: {compress} (ratio: {ratio})")
print(f" Mountpoint: {mount}")
# =============================================================
# 3. PROTOCOL STATUS (stellar-core)
# =============================================================
hdr("⭐ Protocol Status")
if core_info is None:
err("stellar-core not reachable")
else:
q = core_info["quorum"]["qset"]
state = core_info["state"]
block = core_info["ledger"]["num"]
proto = core_info["protocol_version"]
ledger_proto = core_info["ledger"].get("version", "—")
age = core_info["ledger"]["age"]
started = core_info["startedOn"]
if state == "Synced!":
ok(f"State: {state}")
else:
warn(f"State: {state}")
print(f" Block: {block} (Ledger age: {age}s)")
print(f" Protocol: {proto} (Ledger: {ledger_proto})")
phase = q.get('phase', '—')
agree = q.get('agree', '—')
missing = q.get('missing', '—')
lag_ms = q.get('lag_ms', '—')
print(f" Quorum Phase: {phase} (agree: {agree}, missing: {missing}, lag: {lag_ms}ms)")
print(f" Node started: {started}")
# =============================================================
# 4. API STATUS (Horizon)
# =============================================================
hdr("🌅 API Status (Horizon)")
if hz_data is None:
err(f"Horizon API not reachable ({HORIZON_URL})")
else:
core_blk = hz_data.get("core_latest_ledger", 0)
hist_blk = hz_data.get("history_latest_ledger", 0)
ings_blk = hz_data.get("ingest_latest_ledger", 0)
lag = core_blk - ings_blk
closed_at = hz_data.get("history_latest_ledger_closed_at", "?")
if lag <= 2:
ok(f"Status: Synced (Ingest lag: {lag} ledger)")
else:
warn(f"Status: Lagging (Ingest {lag} ledger behind core)")
core_diff = (block - core_blk) if (core_info is not None and core_blk) else None
core_diff_str = f" (Δ {core_diff:+d} vs Core)" if core_diff is not None else ""
print(f" Core Latest Ledger: {core_blk}{core_diff_str}")
print(f" Ingest Latest Ledger: {ings_blk} (last closed: {closed_at})")
print(f" History Oldest: {hist_blk}")
# =============================================================
# 5. NODE METRICS (stellar-core)
# =============================================================
hdr("📊 Node Metrics")
metrics_raw = None
if container:
metrics_raw = exec_json(container, "stellar-core http-command metrics")
if metrics_raw is None:
err("Metrics not available")
else:
m = metrics_raw.get("metrics", {})
lag = m.get("scp.timing.first-to-self-externalize-lag", {})
def _ms(val): return f"{val:.1f}ms" if val is not None else "—"
print(f" SCP externalize lag: min {_ms(lag.get('min'))} mean {_ms(lag.get('mean'))} max {_ms(lag.get('max'))}")
print(f" Percentiles (ms): p50 {_ms(lag.get('median'))} p75 {_ms(lag.get('75%'))} p95 {_ms(lag.get('95%'))} p98 {_ms(lag.get('98%'))} p99 {_ms(lag.get('99%'))} p99.9 {_ms(lag.get('99.9%'))}")
# =============================================================
# 6. PEER CONNECTIONS (stellar-core)
# =============================================================
hdr("🌐 Peer Connections")
peers_raw = None
if container:
peers_raw = exec_json(container, "stellar-core http-command peers?compact=false")
outbound_numbered = [] # [{id, ip, label, location}] in display order, for interactive drop
if peers_raw is None:
err("Peer data not available (stellar-core not reachable)")
else:
ap = peers_raw.get("authenticated_peers", {})
inbound = ap.get("inbound", []) or []
outbound = ap.get("outbound", []) or []
pp = peers_raw.get("pending_peers", {})
p_in = pp.get("inbound", []) or []
p_out = pp.get("outbound", []) or []
in_str = f"Inbound: {len(inbound)}" + (f" (+{len(p_in)} pending)" if p_in else "")
out_str = f"Outbound: {len(outbound)}" + (f" (+{len(p_out)} pending)" if p_out else "")
(warn if p_in else ok)(in_str)
(warn if p_out else ok)(out_str)
if peers_extended:
# --- Fetch full node IDs for interactive drop ---
fullkey_map = {} # address -> full node ID
if interactive and container:
fk_raw = exec_json(
container,
["stellar-core", "http-command", "peers?fullkeys=true"]
)
if fk_raw:
fk_ap = fk_raw.get("authenticated_peers", {})
for peers_list in (fk_ap.get("outbound") or [], fk_ap.get("inbound") or []):
for p in peers_list:
addr = p.get("address", "").rsplit(":", 1)[0]
nid = p.get("id", "")
if addr and nid:
fullkey_map[addr] = nid
# --- Peers DB ---
db_counts = None
try:
exit_code, db_out = container.exec_run([
"psql", "-U", "stellar", "-d", "core",
"-t", "-A", "-F", "|",
"-c", "SELECT type, COUNT(*) FROM peers GROUP BY type;"
])
if exit_code == 0 and db_out:
db_counts = {0: 0, 1: 0, 2: 0}
for line in db_out.decode().strip().splitlines():
parts = line.split("|")
if len(parts) == 2:
try:
t = int(parts[0])
if t in db_counts:
db_counts[t] = int(parts[1])
except ValueError:
pass
except Exception:
pass
print()
if db_counts is not None:
print(f" Peers DB: Inbound: {db_counts[0]} | Outbound: {db_counts[1]} | Preferred: {db_counts[2]}")
else:
print(f" Peers DB: not queryable (sqlite3 or path missing?)")
validators = load_validators(CFG_PATH)
def scp_direction(rx, tx):
if rx == 0 and tx == 0:
return "◆"
if tx == 0 or rx / max(tx, 1) > 1.5:
return "▼" # Receiving more from peer
if rx == 0 or tx / max(rx, 1) > 1.5:
return "▲" # Sending more to peer
return "◆"
# Node uptime
node_up_str = "—"
if core_info:
try:
dt = datetime.fromisoformat(core_info["startedOn"].replace("Z", "+00:00"))
node_elapsed = int((datetime.now(timezone.utc) - dt).total_seconds())
node_up_str = fmt_up(node_elapsed)
except Exception:
pass
# Node ID
node_id_str = core_info.get("node_id", "—") if core_info else "—"
if node_id_str == "—" and container:
try:
_, out = container.exec_run(
["grep", "-oP", r"(?<=nodeID=)G[A-Z0-9]{55}", "/tmp/stellar-core.log"]
)
if out:
node_id_str = out.decode().splitlines()[0].strip() or "—"
except Exception:
pass
# Node IP
node_ip_str = "—"
try:
node_ip_str = socket.gethostbyname(socket.gethostname())
except Exception:
pass
# Quorum phase for Flow column
node_flow_str = "—"
if core_info:
try:
phase = core_info["quorum"]["qset"]["phase"]
node_flow_str = "◆" if phase == "EXTERNALIZE" else phase[:4]
except Exception:
pass
# --- Geolocation batch lookup ---
if geo_enabled:
_all_ips = [p.get("address", "").rsplit(":", 1)[0] for p in inbound + outbound]
if node_ip_str != "—":
_all_ips.append(node_ip_str)
geo_batch([ip for ip in _all_ips if ip])
def _is_private(ip):
try:
return ipaddress.ip_address(ip).is_private
except Exception:
return False
# pid_w: use actual displayed length (validators are truncated to 5)
def _disp_len(p):
ra = p.get("address", "?").rsplit(":", 1)[0]
pid = p.get("id", "")
return 5 if (validators.get(ra) or validators.get(pid)) else len(pid)
pid_w = max((_disp_len(p) for p in inbound + outbound), default=5)
col_w = pid_w + 5 # pid + " (Vn)" or pid + spaces
# Node version: prefer core_info["build"], fall back to Horizon
node_ver_str = "—"
if core_info:
node_ver_str = fmt_version(core_info.get("build", ""))
if node_ver_str == "—" and hz_data:
node_ver_str = fmt_version(hz_data.get("core_version", ""))
# ver_w: dynamic width based on actual version strings
_all_ver_strs = [fmt_version(p.get("ver", "")) for p in inbound + outbound]
ver_w = max((len(s) for s in _all_ver_strs + [node_ver_str]), default=7)
ver_w = max(ver_w, len("Version"))
# addr_w: dynamic width based on actual IP strings (only when show_ip)
if show_ip:
_all_addr_strs = [
p.get("address", "").rsplit(":", 1)[0]
for p in inbound + outbound
]
if node_ip_str != "—":
_all_addr_strs.append(node_ip_str)
ip_col_hdr = "IP"
addr_w = max((len(s) for s in _all_addr_strs), default=15)
addr_w = max(addr_w, len(ip_col_hdr))
if geo_enabled:
_all_geo_strs = [
geo_str(p.get("address", "").rsplit(":", 1)[0])
for p in inbound + outbound
]
if node_ip_str != "—":
_all_geo_strs.append(geo_str(node_ip_str))
geo_col_hdr = "Location"
geo_w = max((len(s) for s in _all_geo_strs + [""]), default=8)
geo_w = max(geo_w, len(geo_col_hdr))
if show_ip:
node_addr_disp = node_ip_str if node_ip_str != "—" else "—"
if geo_enabled:
node_geo_disp = geo_str(node_ip_str) if node_ip_str != "—" else "—"
# In interactive mode outbound rows get a number prefix
num_col = interactive
if num_col:
hdr_num_prefix = " # "
node_row_prefix = " "
else:
hdr_num_prefix = ""
node_row_prefix = ""
_ip_hdr_part = f" {ip_col_hdr:<{addr_w}}" if show_ip else ""
_ip_node_part = f" {node_addr_disp:<{addr_w}}" if show_ip else ""
_geo_hdr_part = f" {geo_col_hdr:<{geo_w}}" if geo_enabled else ""
_geo_node_part = f" {node_geo_disp:<{geo_w}}" if geo_enabled else ""
# Summed rates across all peers for the node summary row
total_rx_rate = sum(
(p.get("byte_read", 0) or 0) * 8 / (p.get("elapsed", 0) or 1)
for p in inbound + outbound if (p.get("elapsed", 0) or 0) > 0
)
total_tx_rate = sum(
(p.get("byte_write", 0) or 0) * 8 / (p.get("elapsed", 0) or 1)
for p in inbound + outbound if (p.get("elapsed", 0) or 0) > 0
)
total_rx_str = fmt_rate(total_rx_rate) if total_rx_rate > 0 else "—"
total_tx_str = fmt_rate(total_tx_rate) if total_tx_rate > 0 else "—"
if extended_traffic:
_traffic_hdr = f" {'RX':>9} {'TX':>9} {'RX/s':>8} {'TX/s':>8}"
_traffic_node = f" {'—':>9} {'—':>9} {total_rx_str:>8} {total_tx_str:>8}"
else:
_all_ratio_parts = [
fmt_ratio_parts(p.get("byte_read", 0) or 0, p.get("byte_write", 0) or 0)
for p in inbound + outbound
]
_valid_parts = [p for p in _all_ratio_parts if p is not None]
ratio_l_w = max((len(p[0]) for p in _valid_parts), default=1)
ratio_r_w = max((len(p[1]) for p in _valid_parts), default=1)
ratio_total = ratio_l_w + 1 + ratio_r_w
ratio_total = max(ratio_total, len("RX/TX"))
_traffic_hdr = f" {'RX/TX':>{ratio_total}}"
_traffic_node = f" {fmt_ratio_cell(None, ratio_l_w, ratio_r_w):>{ratio_total}}"
HDR = f" {hdr_num_prefix}{'NodeID':<{col_w}} {'Version':<{ver_w}}{_ip_hdr_part}{_geo_hdr_part} {'Latency':>8} {'Uptime':<8}{_traffic_hdr} Flow"
NODE_ROW = f" {node_row_prefix}{node_id_str:<{col_w}} {node_ver_str:<{ver_w}}{_ip_node_part}{_geo_node_part} {'—':>8} {node_up_str:<8}{_traffic_node} {node_flow_str}"
SEP = " " + "─" * (len(HDR) - 2)
def show(peers, is_out):
for p in sorted(peers, key=lambda p: p.get("elapsed", 0), reverse=True):
raw_addr = p.get("address", "?").rsplit(":", 1)[0]
pid = p.get("id", "?")
display_addr = raw_addr
addr_disp = display_addr if show_ip else ""
geo_disp = geo_str(display_addr) if geo_enabled else ""
v_match = validators.get(display_addr) or validators.get(raw_addr) or validators.get(pid)
if v_match:
node_col = f"{v_match[1][:5]} (V{v_match[0]})"
else:
node_col = pid
ver = fmt_version(p.get("ver", ""))
lat = p.get("latency", "?")
lat_s = "n/a" if lat == 86400000 else f"{lat}ms"
up = fmt_up(p.get("elapsed", 0))
rx = p.get("byte_read", 0) or 0
tx = p.get("byte_write", 0) or 0
elapsed = p.get("elapsed", 0) or 0
flow = scp_direction(rx, tx)
rx_rate_s = fmt_rate(rx * 8 / elapsed) if elapsed > 0 else "—"
tx_rate_s = fmt_rate(tx * 8 / elapsed) if elapsed > 0 else "—"
if num_col and is_out:
n = len(outbound_numbered) + 1
full_id = (fullkey_map.get(display_addr)
or fullkey_map.get(raw_addr)
or pid)
location = geo_label(display_addr) if geo_enabled else display_addr
outbound_numbered.append({"id": full_id, "ip": display_addr, "label": node_col, "location": location})
row_prefix = f"{n:>2} "
elif num_col:
row_prefix = " "
else:
row_prefix = ""
_ip_row_part = f" {addr_disp:<{addr_w}}" if show_ip else ""
_geo_row_part = f" {geo_disp:<{geo_w}}" if geo_enabled else ""
if extended_traffic:
_traffic_row = f" {fmt_bytes(rx):>9} {fmt_bytes(tx):>9} {rx_rate_s:>8} {tx_rate_s:>8}"
else:
_parts = fmt_ratio_parts(rx, tx)
_traffic_row = f" {fmt_ratio_cell(_parts, ratio_l_w, ratio_r_w):>{ratio_total}}"
print(f" {row_prefix}{node_col:<{col_w}} {ver:<{ver_w}}{_ip_row_part}{_geo_row_part} {lat_s:>8} {up:<8}{_traffic_row} {flow}")
print()
print(HDR)
print(NODE_ROW)
print(SEP)
if inbound:
print(" Inbound")
show(inbound, False)
if outbound:
if inbound:
print()
print(" Outbound")
show(outbound, True)
print()
return outbound_numbered, container
def do_drop_workflow(outbound_peers, container):
"""Interactive drop peer workflow."""
if not outbound_peers:
print(" Keine outbound peers verfügbar.")
time.sleep(2)
return
print(f"\n Peer-Nummer (1–
old_settings = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin.fileno())
try:
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print(ch)
if ch == '\x1b' or not ch.isdigit():
return
try:
n = int(ch)
except ValueError:
return
if n < 1 or n > len(outbound_peers):
print(f" Ungültige Nummer (1-{len(outbound_peers)})")
time.sleep(2)
return
peer = outbound_peers[n - 1]
full_id = peer["id"]
ip = peer["ip"]
label = peer["label"]
location = peer.get("location", ip)
print(f" Drop peer {n} ({label[:8]}... / {location})? [ENTER] ja [ESC] abbrechen: ", end="", flush=True)
old_settings = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin.fileno())
try:
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print()
if ch in ('\r', '\n'):
result = exec_json(
container,
["stellar-core", "http-command", f"droppeer?node={full_id}"]
)
if result is None or result.get("status", "").upper() == "OK":
print(f" Peer {full_id[:8]}... ({ip}) getrennt.")
else:
print(f" Unerwartete Antwort: {result.get('status', result)}")
time.sleep(2)
def drop_all_workflow(outbound_peers, container):
"""Drop all outbound peers with a single confirmation."""
if not outbound_peers:
print(" Keine outbound peers verfügbar.")
time.sleep(2)
return
peer_list = " ".join(f"{p['label'][:8]}…/{p.get('location', p['ip'])}" for p in outbound_peers)
print(f"\n Drop all ({len(outbound_peers)}): {peer_list}")
print(f" [ENTER] ja [ESC] abbrechen: ", end="", flush=True)
old_settings = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin.fileno())
try:
ch = os.read(sys.stdin.fileno(), 1).decode(errors="replace")
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print()
if ch not in ('\r', '\n'):
return
ok_count = 0
for peer in outbound_peers:
result = exec_json(
container,
["stellar-core", "http-command", f"droppeer?node={peer['id']}"]
)
if result is None or result.get("status", "").upper() == "OK":
print(f" Getrennt: {peer['label'][:8]}... / {peer.get('location', peer['ip'])}")
ok_count += 1
else:
print(f" Fehlgeschlagen ({peer.get('location', peer['ip'])}): {result.get('status', result)}")
print(f"\n {ok_count}/{len(outbound_peers)} Verbindungen getrennt.")
time.sleep(3)
def interactive_loop(client, refresh_interval, peers_extended, geo_enabled, show_ip, extended_traffic=False):
if not HAS_TTY:
print("ERROR: Interaktiver Modus benötigt 'tty'/'termios'")
sys.exit(1)
outbound_peers = []
container = None
traffic_extended = extended_traffic # local, toggled with 'x'
old_settings = termios.tcgetattr(sys.stdin)
try:
while True:
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
outbound_peers, container = run_cycle(
client, peers_extended, geo_enabled, show_ip, interactive=True,
extended_traffic=traffic_extended
)
sys.stdout.write("\033[2J\033[H" + buf.getvalue())
sys.stdout.flush()
deadline = time.monotonic() + refresh_interval
tty.setcbreak(sys.stdin.fileno())
try:
last_secs = -1
while True:
remaining = max(0.0, deadline - time.monotonic())
secs = int(remaining) + (1 if remaining % 1 > 0 else 0)
if secs != last_secs:
x_label = "RX/TX" if traffic_extended else "detail"
text = f" [d] Drop peer [a] Drop all [x] {x_label} [q] Beenden Refresh in {secs}s "
sys.stdout.write("\r" + text)
sys.stdout.flush()
last_secs = secs
if remaining <= 0:
break
rlist, _, _ = select.select([sys.stdin], [], [], min(remaining, 0.25))
if rlist:
ch = sys.stdin.read(1)
if ch in ('q', 'Q', '\x03', '\x04'):
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print()
return
elif ch in ('d', 'D') and outbound_peers and container:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print()
do_drop_workflow(outbound_peers, container)
break # trigger immediate refresh
elif ch in ('a', 'A') and outbound_peers and container:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
print()
drop_all_workflow(outbound_peers, container)
break # trigger immediate refresh
elif ch in ('x', 'X'):
traffic_extended = not traffic_extended
break # trigger immediate refresh
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
except KeyboardInterrupt:
pass
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings)
sys.stdout.write("\r\n")
sys.stdout.flush()
USAGE = """\
pi-stat — Stellar Core node status monitor
Usage:
./pi-stat.py [OPTIONS]
Options:
-p, --peers Extended peer table
-g, --geo Geolocation per peer (implies -p)
-s, --show-ip Show IP column (hidden by default)
-x, --extended Full RX/TX/RX-rate/TX-rate columns (default: ratio only)
-i [SEC], --interactive [SEC]
Auto-refresh mode (default interval: 10 s)
-?, --help Show this help and exit
Interactive keys:
d Drop one outbound peer
a Drop all outbound peers
q Quit
"""
def main():
# --- CLI options ---
args = sys.argv[1:]
if "-?" in args or "--help" in args:
print(USAGE, end="")
sys.exit(0)
peers_extended = "-p" in args or "--peers" in args
geo_enabled = "-g" in args or "--geo" in args
show_ip = "-s" in args or "--show-ip" in args
interactive = "-i" in args or "--interactive" in args
extended_traffic = "-x" in args or "--extended" in args
refresh_interval = 10
if interactive:
for flag in ("-i", "--interactive"):
if flag in args:
idx = args.index(flag)
if idx + 1 < len(args):
try:
refresh_interval = max(1, int(args[idx + 1]))
except ValueError:
pass
if geo_enabled:
peers_extended = True
# --- Docker client ---
try:
client = docker.from_env()
except Exception as e:
print(f"ERROR: Docker not reachable: {e}")
sys.exit(1)
if interactive:
interactive_loop(client, refresh_interval, peers_extended, geo_enabled, show_ip, extended_traffic)
else:
run_cycle(client, peers_extended, geo_enabled, show_ip, interactive=False, extended_traffic=extended_traffic)
if __name__ == "__main__":
main()
