321 lines
10 KiB
Python
321 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Author: Hendrik Schutter, mail@hendrikschutter.com
|
|
Date of creation: 2022/05/23
|
|
Date of last modification: 2026/08/21
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from rpi_lcd import LCD
|
|
import netifaces as ni
|
|
import subprocess
|
|
import time
|
|
import shutil
|
|
import sys
|
|
import socket
|
|
|
|
lcd = LCD(address=0x3F, bus=2, width=16, rows=2, backlight=True)
|
|
|
|
LCD_WIDTH = 16
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: truncate a string to fit the LCD width
|
|
# ---------------------------------------------------------------------------
|
|
def lcd_fit(text: str, width: int = LCD_WIDTH) -> str:
|
|
"""Truncate text to at most `width` characters."""
|
|
return text[:width]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Network helpers
|
|
# ---------------------------------------------------------------------------
|
|
def get_ip_addr(interface_name: str) -> dict:
|
|
"""Return IPv4 and IPv6 address for the given interface.
|
|
|
|
Returns a dict with keys 'ipv4' and 'ipv6'. Values are the address string
|
|
or None when not available.
|
|
"""
|
|
result = {"ipv4": None, "ipv6": None}
|
|
try:
|
|
addrs = ni.ifaddresses(interface_name)
|
|
if ni.AF_INET in addrs:
|
|
result["ipv4"] = addrs[ni.AF_INET][0].get("addr")
|
|
if ni.AF_INET6 in addrs:
|
|
result["ipv6"] = addrs[ni.AF_INET6][0].get("addr")
|
|
except (ValueError, KeyError):
|
|
pass
|
|
return result
|
|
|
|
|
|
def get_mac_addr(interface_name: str) -> str | None:
|
|
"""Return the MAC address for the given interface, or None on error."""
|
|
try:
|
|
addrs = ni.ifaddresses(interface_name)
|
|
if ni.AF_LINK in addrs:
|
|
return addrs[ni.AF_LINK][0].get("addr")
|
|
except (ValueError, KeyError):
|
|
pass
|
|
return None
|
|
|
|
|
|
def is_interface_up(interface_name: str) -> bool:
|
|
"""Return True when the interface exists and has an IPv4 address."""
|
|
try:
|
|
addrs = ni.ifaddresses(interface_name)
|
|
return ni.AF_INET in addrs
|
|
except (ValueError, KeyError):
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WireGuard helper
|
|
# ---------------------------------------------------------------------------
|
|
def get_wireguard_state(config_name: str) -> bool:
|
|
"""Return True when the WireGuard interface reports a latest handshake."""
|
|
try:
|
|
subprocess.check_output(
|
|
f"wg show {config_name} latest-handshakes",
|
|
shell=True,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
return True
|
|
except subprocess.CalledProcessError:
|
|
return False
|
|
|
|
|
|
def get_wireguard_endpoint(config_name: str) -> str | None:
|
|
"""Return the endpoint hostname/IP of the first WireGuard peer, or None."""
|
|
try:
|
|
output = (
|
|
subprocess.check_output(
|
|
f"wg show {config_name} endpoints",
|
|
shell=True,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
.decode()
|
|
.strip()
|
|
)
|
|
# Output format: "<pubkey>\t<endpoint_ip>:<port>"
|
|
if output:
|
|
parts = output.split()
|
|
if len(parts) >= 2:
|
|
endpoint = parts[1] # "<ip>:<port>" or "[ipv6]:port"
|
|
# Try to resolve hostname; fall back to raw IP
|
|
host = endpoint.rsplit(":", 1)[0].strip("[]")
|
|
try:
|
|
hostname = socket.gethostbyaddr(host)[0]
|
|
return hostname
|
|
except socket.herror:
|
|
return host
|
|
except subprocess.CalledProcessError:
|
|
pass
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# System info helpers
|
|
# ---------------------------------------------------------------------------
|
|
def get_kernel_version() -> str:
|
|
"""Return the running kernel version string."""
|
|
try:
|
|
return (
|
|
subprocess.check_output("uname -r", shell=True, stderr=subprocess.STDOUT)
|
|
.decode()
|
|
.strip()
|
|
)
|
|
except subprocess.CalledProcessError:
|
|
return "unknown"
|
|
|
|
|
|
def get_os_version() -> str:
|
|
"""Return a short OS description from /etc/os-release."""
|
|
try:
|
|
with open("/etc/os-release", "r") as f:
|
|
for line in f:
|
|
if line.startswith("PRETTY_NAME="):
|
|
return line.split("=", 1)[1].strip().strip('"')
|
|
except OSError:
|
|
pass
|
|
return "unknown OS"
|
|
|
|
|
|
def get_disk_usage(disk_path: str) -> str:
|
|
"""Return free disk space as a percentage string, or an error string."""
|
|
try:
|
|
total, used, free = shutil.disk_usage(disk_path)
|
|
if total == 0:
|
|
return "N/A"
|
|
pct = (free // (2**30)) / (total // (2**30)) * 100.0
|
|
return f"{pct:.2f}%"
|
|
except OSError:
|
|
return "disk error"
|
|
|
|
|
|
def get_uptime() -> str:
|
|
"""Return system uptime as a human-readable string."""
|
|
try:
|
|
with open("/proc/uptime", "r") as f:
|
|
uptime_seconds = float(f.readline().split()[0])
|
|
if uptime_seconds >= 60.0 * 60.0 * 24.0:
|
|
return f"{uptime_seconds / (60.0 * 60.0 * 24.0):.2f} days"
|
|
return f"{uptime_seconds / 3600.0:.2f} hours"
|
|
except (OSError, ValueError):
|
|
return "unknown"
|
|
|
|
|
|
def get_system_time() -> str:
|
|
"""Return the current local time as HH:MM."""
|
|
return datetime.now().strftime("%H:%M")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LCD helpers
|
|
# ---------------------------------------------------------------------------
|
|
def show_two_line(
|
|
top: str, bottom: str, duration: float, align_bottom: str = "right"
|
|
) -> None:
|
|
"""Display two lines on the LCD and wait for `duration` seconds."""
|
|
lcd.clear()
|
|
lcd.text(lcd_fit(top), 1)
|
|
lcd.text(lcd_fit(bottom), 2, align=align_bottom)
|
|
time.sleep(duration)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Startup sequence
|
|
# ---------------------------------------------------------------------------
|
|
def show_startup_screens(interfaces: list[str], wireguard_interface: str) -> None:
|
|
"""Display system info screens at startup (2 s each)."""
|
|
|
|
# Welcome banner
|
|
show_two_line(" Offsite Backup", " fckaf.de/FHA", 2.0, align_bottom="left")
|
|
|
|
# Kernel version
|
|
kernel = get_kernel_version()
|
|
show_two_line("Kernel:", kernel, 2.0)
|
|
|
|
# OS version
|
|
os_ver = get_os_version()
|
|
show_two_line("OS:", os_ver, 2.0)
|
|
|
|
# MAC address for each interface
|
|
for iface in interfaces:
|
|
mac = get_mac_addr(iface)
|
|
if mac:
|
|
show_two_line(f"MAC {iface}:", mac, 2.0)
|
|
else:
|
|
show_two_line(f"MAC {iface}:", "not available", 2.0)
|
|
|
|
# WireGuard endpoint
|
|
endpoint = get_wireguard_endpoint(wireguard_interface)
|
|
if endpoint:
|
|
show_two_line("WG Endpoint:", endpoint, 2.0)
|
|
else:
|
|
show_two_line("WG Endpoint:", "not available", 2.0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interface display helpers
|
|
# ---------------------------------------------------------------------------
|
|
def show_interface_info(iface: str, ip_version: str, duration: float) -> None:
|
|
"""Show IPv4 or IPv6 address for an interface, or a 'not connected' message."""
|
|
label = f"{ip_version} {iface}:"
|
|
addrs = get_ip_addr(iface)
|
|
|
|
if ip_version == "IPv4":
|
|
addr = addrs["ipv4"]
|
|
else:
|
|
raw = addrs["ipv6"]
|
|
# Strip well-known link-local prefix suffix used in display
|
|
addr = raw[14:-1] if raw and len(raw) > 15 else raw
|
|
|
|
if addr:
|
|
show_two_line(label, addr, duration)
|
|
else:
|
|
show_two_line(label, f"{iface} not connected", duration, align_bottom="left")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main loop
|
|
# ---------------------------------------------------------------------------
|
|
def main() -> None:
|
|
if len(sys.argv) < 4:
|
|
print("Usage: pyLCDinfo.py <iface1> [iface2 ...] <wireguard_iface> <disk_path>")
|
|
print(
|
|
" At least one local interface, one WireGuard interface, and a disk path are required."
|
|
)
|
|
exit(-1)
|
|
|
|
# Last argument: disk path
|
|
disk_path = sys.argv[-1]
|
|
# Second-to-last argument: WireGuard interface
|
|
wireguard_interface = sys.argv[-2]
|
|
# Everything before that: local interfaces (at least one)
|
|
local_interfaces = sys.argv[1:-2]
|
|
|
|
if not local_interfaces:
|
|
print("Error: at least one local interface must be provided.")
|
|
exit(-1)
|
|
|
|
# --- Startup screens ---
|
|
show_startup_screens(local_interfaces, wireguard_interface)
|
|
|
|
state_ok = False
|
|
|
|
while True:
|
|
if not state_ok:
|
|
lcd.backlight(turn_on=True)
|
|
|
|
# Show IPv4 and IPv6 for each local interface
|
|
for iface in local_interfaces:
|
|
show_interface_info(iface, "IPv4", 2.4)
|
|
show_interface_info(iface, "IPv6", 2.4)
|
|
|
|
# WireGuard waiting animation
|
|
lcd.clear()
|
|
lcd.text("Waiting for VPN", 1)
|
|
for loops in range(4):
|
|
for fill in range(14):
|
|
if loops % 2 == 0:
|
|
lcd.text("<>".rjust(fill), 2, align="left")
|
|
else:
|
|
lcd.text("<>".ljust(fill), 2, align="right")
|
|
time.sleep(0.1)
|
|
|
|
if get_wireguard_state(wireguard_interface):
|
|
state_ok = True
|
|
lcd.clear()
|
|
lcd.text(" VPN connected ", 1)
|
|
t = get_system_time()
|
|
lcd.text(t.rjust(LCD_WIDTH - len(t)), 2, align="left")
|
|
time.sleep(10.0)
|
|
|
|
else:
|
|
lcd.backlight(turn_on=False)
|
|
|
|
# Show IPv4 and IPv6 for each local interface
|
|
for iface in local_interfaces:
|
|
show_interface_info(iface, "IPv4", 5.0)
|
|
show_interface_info(iface, "IPv6", 5.0)
|
|
|
|
# Disk usage
|
|
show_two_line("Free Disk:", get_disk_usage(disk_path), 5.0)
|
|
|
|
# Uptime
|
|
show_two_line("System Uptime:", get_uptime(), 5.0)
|
|
|
|
# VPN status + time — long idle sleep to reduce CPU usage
|
|
lcd.clear()
|
|
lcd.text(" VPN Connected ", 1)
|
|
t = get_system_time()
|
|
lcd.text(t.rjust(LCD_WIDTH - len(t)), 2, align="left")
|
|
time.sleep(40.0)
|
|
|
|
# Re-check WireGuard connectivity
|
|
state_ok = get_wireguard_state(wireguard_interface)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|