RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/rmi_dashboard.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- Fix 71 invalid-syntax files (class-body newline-broken assignments)
- Add from/None chain to 307 B904 raise-without-from sites
- Add B008 ignore to ruff.toml (already in pyproject.toml)
- Noqa F401 on __init__.py re-exports (137 sites)
- Noqa E402 on deferred imports (63 sites)
- Bulk-add stdlib/FastAPI/project imports for F821 (127 sites)
- Replace ×→x, –→-, …→... in docstrings (4093 chars)
- Manual refactor of 5 SIM103/SIM116 patterns

Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py)
Co-authored-by: opencode <opencode@rugmunch.io>
2026-07-06 15:43:20 +02:00

353 lines
14 KiB
Python
Raw Permalink Blame History

"""
RMI TUI Dashboard - Terminal User Interface
===========================================
Interactive CLI dashboard for RMI platform.
Features:
- System status display
- Menu-driven navigation
- Quick commands for common actions
- Live backend status monitoring
"""
import os
import subprocess
import time
from typing import Any
import httpx
from app.core.logging import get_logger
logger = get_logger(__name__)
# Terminal colors
class Colors:
RED = "\033[0;31m"
GREEN = "\033[0;32m"
YELLOW = "\033[1;33m"
BLUE = "\033[0;34m"
PURPLE = "\033[0;35m"
CYAN = "\033[0;36m"
WHITE = "\033[0;37m"
BOLD = "\033[1m"
RESET = "\033[0m"
# ─── UI UTILITIES ──────────────────────────────────────────────────
def clear_screen():
"""Clear terminal screen."""
os.system("clear" if os.name == "posix" else "cls")
def print_header(title: str):
"""Print centered header."""
width = 60
logger.info(f"\n{Colors.BLUE}{'=' * width}{Colors.RESET}")
print(
f"{Colors.BLUE}║{Colors.RESET}{Colors.BOLD}{title.center(width - 2)}{Colors.RESET}{Colors.BLUE}║{Colors.RESET}"
)
logger.info(f"{Colors.BLUE}{'=' * width}{Colors.RESET}\n")
def print_box(title: str, content: str, color=Colors.WHITE):
"""Print content in a box."""
lines = content.split("\n")
max_len = max(len(l) for l in lines) if lines else 0 # noqa: E741
width = max_len + 4
logger.info(f"\n{color}┌{'─' * width}┐{Colors.RESET}")
logger.info(f"{color}│{Colors.RESET} {Colors.BOLD}{title.center(width - 2)}{Colors.RESET} {color}│{Colors.RESET}")
logger.info(f"{color}├{'─' * width}┤{Colors.RESET}")
for line in lines:
logger.info(f"{color}│{Colors.RESET} {line.ljust(width - 2)} {color}│{Colors.RESET}")
logger.info(f"{color}└{'─' * width}┘{Colors.RESET}\n")
# ─── STATUS MONITOR ────────────────────────────────────────────────
def get_backend_status(url: str = "http://127.0.0.1:8010") -> dict[str, Any]:
"""Get backend health status."""
try:
response = httpx.get(f"{url}/health", timeout=2)
if response.status_code == 200:
return response.json()
except Exception:
pass
return {"error": "Backend not reachable"}
def get_redis_status() -> dict[str, Any]:
"""Get Redis connection status."""
try:
import redis
r = redis.Redis(host="localhost", port=6379, db=0, socket_timeout=2)
if r.ping():
info = r.info()
return {
"connected": True,
"version": info.get("redis_version", "unknown"),
"memory_used": info.get("used_memory_human", "unknown"),
"connections": info.get("connected_clients", 0),
}
except Exception:
pass
return {"error": "Redis not reachable"}
# ─── DASHBOARD UI ──────────────────────────────────────────────────
class Dashboard:
"""Interactive TUI Dashboard."""
def __init__(self):
self.running = True
self.current_view = "main"
def display_main(self):
"""Display main menu."""
clear_screen()
# Header
logger.info(f"{Colors.CYAN}{Colors.BOLD}")
logger.info("╔════════════════════════════════════════════════════════════╗")
logger.info("║ RMI INTELLIGENCE PLATFORM v2.0.0 ║")
logger.info("╚════════════════════════════════════════════════════════════╝")
logger.info(f"{Colors.RESET}")
# Status
backend = get_backend_status()
redis = get_redis_status()
backend_status = (
f"{Colors.GREEN}● Online{Colors.RESET}"
if backend.get("status") == "ok"
else f"{Colors.RED}● Offline{Colors.RESET}"
)
redis_status = (
f"{Colors.GREEN}● Connected{Colors.RESET}"
if redis.get("connected")
else f"{Colors.RED}● Disconnected{Colors.RESET}"
)
logger.info(f"{Colors.YELLOW}System Status:{Colors.RESET}")
logger.info(f" Backend: {backend_status} | Redis: {redis_status}")
if backend.get("status") == "ok":
logger.info(f" Service: {backend.get('service', 'unknown')}")
logger.info(f" Version: {backend.get('version', 'unknown')}")
logger.info(f" Timestamp: {backend.get('timestamp', 'N/A')}")
logger.info(f"\n{Colors.YELLOW}Available Ports:{Colors.RESET}")
logger.info(" 8010 - Main Backend API")
logger.info(" 6379 - Redis Server")
logger.info(" 22 - SSH (if enabled)")
# Features
logger.info(f"\n{Colors.YELLOW}Built-in Features:{Colors.RESET}")
features = [
("[1] Server Control", "Start/stop/restart backend server"),
("[2] Auth System", "User authentication & OAuth management"),
("[3] Security Intel", "Threat intel & contract scanning"),
("[4] x402 Micropayments", "Micropayment enforcement & tools"),
("[5] Chat Interface", "Direct chat with RMI assistant"),
("[6] Dashboard TUI", "Interactive terminal dashboard"),
("[7] Network Status", "Check all service connectivity"),
("[0] Quit", "Exit to shell"),
]
for key, (name, desc) in enumerate(features):
logger.info(f" {Colors.GREEN}{key}{Colors.RESET}. {Colors.CYAN}{name:<20}{Colors.RESET} - {desc}")
logger.info(f"\n{Colors.PURPLE}Use 'help' for more information{Colors.RESET}\n")
def display_server_control(self):
"""Display server control menu."""
print_header("Server Control")
pid = self._get_backend_pid()
status = "Running" if pid else "Stopped"
logger.info(f" Backend Status: {Colors.GREEN if pid else Colors.RED}{status}{Colors.RESET}")
if pid:
logger.info(f" PID: {pid}")
logger.info("\n Actions:")
logger.info(f" {Colors.GREEN}1{Colors.RESET}. Start Backend")
logger.info(f" {Colors.GREEN}2{Colors.RESET}. Stop Backend")
logger.info(f" {Colors.GREEN}3{Colors.RESET}. Restart Backend")
logger.info(f" {Colors.GREEN}0{Colors.RESET}. Back")
choice = input("\n Select: ").strip()
if choice == "1":
self._start_backend()
elif choice == "2":
self._stop_backend()
elif choice == "3":
self._restart_backend()
def _get_backend_pid(self) -> int | None:
"""Get backend process ID."""
try:
result = subprocess.run(["pgrep", "-f", "uvicorn main:app"], capture_output=True, text=True, timeout=5)
if result.stdout.strip():
return int(result.stdout.strip())
except Exception:
pass
return None
def _start_backend(self):
"""Start backend server."""
logger.info("\n{Colors.YELLOW}Starting backend...{Colors.RESET}")
try:
subprocess.Popen(
["python3", "-m", "uvicorn", "main:app", "--host", "127.0.0.1", "--port", "8010"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
time.sleep(3)
logger.info(f"{Colors.GREEN}Backend started!{Colors.RESET}")
except Exception as e:
logger.warning(f"{Colors.RED}Error starting backend: {e}{Colors.RESET}")
def _stop_backend(self):
"""Stop backend server."""
pid = self._get_backend_pid()
if pid:
try:
os.kill(pid, 15)
logger.info(f"{Colors.GREEN}Backend stopped (PID: {pid}){Colors.RESET}")
except Exception as e:
logger.warning(f"{Colors.RED}Error stopping backend: {e}{Colors.RESET}")
def _restart_backend(self):
"""Restart backend server."""
self._stop_backend()
time.sleep(1)
self._start_backend()
def display_auth_system(self):
"""Display auth system info."""
print_header("Authentication System")
logger.info(f" {Colors.CYAN}OAuth Providers:{Colors.RESET}")
logger.info(" • Email + Password")
logger.info(" • Google")
logger.info(" • Telegram")
logger.info(" • GitHub")
logger.info(" • Wallet Signature (EIP-4361)")
logger.info(f"\n {Colors.CYAN}Features:{Colors.RESET}")
logger.info(" • Session management (JWT)")
logger.info(" • Rate limiting per user")
logger.info(" • x402 micropayment enforcement")
logger.info(f"\n {Colors.CYAN}Endpoints:{Colors.RESET}")
logger.info(" POST /api/v1/auth/register - Register new user")
logger.info(" POST /api/v1/auth/login - Login user")
logger.info(" GET /api/v1/auth/status - Get current session")
input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")
def display_security_intel(self):
"""Display security intel info."""
print_header("Security Intelligence")
logger.info(f" {Colors.CYAN}Modules:{Colors.RESET}")
logger.info(" • Slither - Static analysis")
logger.info(" • Mythril - Symbolic execution")
logger.info(" • OFAC Sanctions - Blocklist")
logger.info(" • Threat Intel - Risk scoring")
logger.info(f"\n {Colors.CYAN}Contract Scanning:{Colors.RESET}")
logger.info(" • Reentrancy detection")
logger.info(" • Integer overflow/underflow")
logger.info(" • Unchecked calls")
logger.info(" • DAO patterns")
input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")
def display_x402_menu(self):
"""Display x402 micropayments info."""
print_header("x402 Micropayments")
logger.info(f" {Colors.CYAN}Features:{Colors.RESET}")
logger.info(" • Micropayment enforcement")
logger.info(" • Rate-based pricing")
logger.info(" • Forensic analysis")
logger.info(" • Redis-backed tracking")
logger.info(f"\n {Colors.CYAN}Endpoints:{Colors.RESET}")
logger.info(" GET /api/v1/x402/tools - Available tools catalog")
logger.info(" GET /api/v1/x402/status - Payment status")
logger.info(" POST /api/v1/x402/enforce - Enforce micropayment")
input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")
def display_chat(self):
"""Launch chat interface."""
print_header("Chat Interface")
logger.info("\n{Colors.YELLOW}Direct chat with RMI assistant{Colors.RESET}")
logger.info("\n{Colors.RED}Note:{Colors.RESET} This launches the system shell")
input(f"\n{Colors.PURPLE}Press Enter to return...{Colors.RESET}")
def display_network_status(self):
"""Display network connectivity status."""
print_header("Network Status")
tests = [
("Local Backend", "http://127.0.0.1:8010", "/health"),
("Redis", "localhost", None),
("Ethereum RPC", "https://ethereum-rpc.publicnode.com", None),
("Base RPC", "https://base-rpc.publicnode.com", None),
]
for name, url, path in tests:
try:
full_url = f"{url}{path}" if path else url
response = httpx.get(full_url, timeout=3)
status = f"{Colors.GREEN}✓ OK{Colors.RESET} ({response.status_code})"
except Exception:
status = f"{Colors.RED}✗ Failed{Colors.RESET}"
logger.info(f" {name:<20} {status}")
logger.info()
input(f"{Colors.PURPLE}Press Enter to return...{Colors.RESET}")
def run(self):
"""Main dashboard loop."""
while self.running:
self.display_main()
choice = input("Select an option: ").strip().lower()
if choice == "0":
self.running = False
elif choice == "1":
self.display_server_control()
elif choice == "2":
self.display_auth_system()
elif choice == "3":
self.display_security_intel()
elif choice == "4":
self.display_x402_menu()
elif choice == "5":
self.display_chat()
elif choice == "6":
logger.info("{TUI already running}")
elif choice == "7":
self.display_network_status()
elif choice in ["help", "?"]:
logger.info(f"{Colors.YELLOW}Commands:{Colors.RESET}")
logger.info(" help - Show this help")
logger.info(" clear - Clear screen")
logger.info(" quit - Exit dashboard")
elif choice == "clear":
clear_screen()
elif choice in ["quit", "exit"]:
self.running = False
else:
logger.info(f"{Colors.RED}Invalid option{Colors.RESET}")
# ─── SINGLETON LAUNCHER ────────────────────────────────────────────
def launch_dashboard():
"""Launch the interactive dashboard."""
try:
dashboard = Dashboard()
dashboard.run()
except KeyboardInterrupt:
pass
except Exception as e:
logger.warning(f"{Colors.RED}Dashboard error: {e}{Colors.RESET}")
# ─── CLI ENTRY POINT ───────────────────────────────────────────────
if __name__ == "__main__":
launch_dashboard()
Reference in a new issue View git blame Copy permalink