diagnostics.py

#!/usr/bin/env python3
"""End-to-end diagnostics for the unified dashboard.

Cross-platform: Windows + Linux/WSL.
"""

import ast
import os
import platform
import socket
import subprocess
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent
FILES = [
    "config.py",
    "app.py",
    "pages/home.py",
    "pages/terminal.py",
    "pages/signals.py",
    "pages/signal_scanner.py",
    "pages/trading_terminal.py",
    "pages/system_health.py",
    "pages/core_status.py",
    "pages/settings.py",
    "pages/omniroute.py",
    "pages/logs.py",
    "pages/information.py",
    "pages/analytics.py",
    "jarvis/jarvis_brain.py",
    "jarvis/workspace_watcher.py",
    "status_checks.py",
    "widget_lib.py",
    "jarvis_theme.py",
    "floating_journal.py",
    "eni_assistant.py",
    "command_palette.py",
    "audio_feedback.py",
]
LOGS = {
    "ai": os.path.join(tempfile.gettempdir(), "ai-command-center-error.log"),
    "deepcharts": os.path.join(tempfile.gettempdir(), "deepchartspro.log"),
}
WANTED_PORTS = [8501, 8502]
IS_WINDOWS = platform.system() == "Windows"


def section(title: str) -> None:
    print(f"\n=== {title} ===")


def check_syntax() -> int:
    section("Syntax")
    failures = 0
    for rel in FILES:
        path = ROOT / rel
        try:
            with open(path, "r", encoding="utf-8") as f:
                ast.parse(f.read())
            print(f"[ok] {rel}")
        except Exception as exc:
            print(f"[fail] {rel}: {exc}")
            failures += 1
    return failures


def check_pages_import() -> int:
    section("Page imports")
    failures = 0
    # Only check modules that are registered in app.py PAGE_RENDERERS
    pages = [
        ("pages/home", "render_home"),
        ("pages/terminal", "render_terminal"),
        ("pages/signals", "render_signals"),
        ("pages/signal_scanner", "render_signal_scanner"),
        ("pages/trading_terminal", "render_trading_terminal"),
        ("pages/system_health", "render_system_health"),
        ("pages/core_status", "render_core_status"),
        ("pages/settings", "render_settings"),
        ("pages/omniroute", "render_omniroute"),
        ("pages/logs", "render_logs"),
        ("pages/information", "render_information"),
        ("pages/analytics", "render_analytics"),
        ("jarvis/jarvis_brain", "render_jarvis_console"),
    ]
    for rel, func_name in pages:
        full = ROOT / f"{rel}.py"
        try:
            code = full.read_text(encoding="utf-8")
            if f"def {func_name}" not in code:
                raise RuntimeError(f"missing {func_name}()")
            print(f"[ok] {rel}.py -> {func_name}()")
        except Exception as exc:
            print(f"[fail] {rel}: {exc}")
            failures += 1
    return failures


def check_status_checks() -> int:
    section("Status checks module")
    failures = 0
    path = ROOT / "status_checks.py"
    try:
        code = path.read_text(encoding="utf-8")
        if "STATUS_FUNCS" not in code or "ServiceStatus" not in code:
            raise RuntimeError("missing STATUS_FUNCS or ServiceStatus")
        print("[ok] status_checks.py contains STATUS_FUNCS")
    except Exception as exc:
        print(f"[fail] status_checks.py: {exc}")
        failures += 1
    return failures


def check_ports() -> tuple[int, list[str]]:
    section("Live ports")
    found = []
    for port in WANTED_PORTS:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(2)
            if s.connect_ex(("127.0.0.1", port)) == 0:
                print(f"[ok] {port} is accepting connections")
                found.append(str(port))
            else:
                print(f"[fail] {port} is not accepting connections")
    return len(WANTED_PORTS) - len(found), found


def _windows_process_check(cmd_fragment: str) -> bool:
    """Check if a python process with the given command-line fragment is running on Windows."""
    try:
        ps_cmd = (
            f"Get-CimInstance Win32_Process -Filter \"Name='python.exe'\" "
            f"| Where-Object {{ $_.CommandLine -like '*{cmd_fragment}*' }} "
            f"| Select-Object -ExpandProperty CommandLine"
        )
        res = subprocess.run(
            ["powershell", "-NoProfile", "-Command", ps_cmd],
            capture_output=True,
            text=True,
            timeout=10,
        )
        output = (res.stdout or "").strip()
        return bool(output) and cmd_fragment.lower() in output.lower()
    except Exception:
        return False


def _unix_process_check(cmd_fragment: str) -> bool:
    """Check process on Linux/WSL via ps."""
    try:
        res = subprocess.run(
            ["bash", "-lc", f"ps -ef | grep -v grep | grep -F '{cmd_fragment}'"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        return bool(res.stdout.strip())
    except Exception:
        return False


def check_processes() -> int:
    section("Processes")
    wanted = [
        ("ai-command-center/app.py", "AI Command Center (8502)"),
        ("deepchartspro/dashboard.py", "DeepChartsPro (8501)"),
    ]
    check_fn = _windows_process_check if IS_WINDOWS else _unix_process_check
    hits = 0
    for fragment, label in wanted:
        found = check_fn(fragment)
        if found:
            print(f"[ok] {label} process found")
            hits += 1
        else:
            print(f"[fail] {label} process not found")
    return len(wanted) - hits


def check_logs() -> int:
    section("Logs")
    failures = 0
    for name, path in LOGS.items():
        if Path(path).exists():
            tail = Path(path).read_text(encoding="utf-8", errors="ignore").splitlines()[-20:]
            print(f"[ok] {name} log exists, tail:")
            for line in tail[-5:]:
                print(f"    {line}")
        else:
            print(f"[ok] {name} log not yet created")
    return failures


def main() -> int:
    failures = 0
    failures += check_syntax()
    failures += check_pages_import()
    failures += check_status_checks()
    failures += check_processes()
    port_failures, ports = check_ports()
    failures += port_failures
    failures += check_logs()

    section("Summary")
    if failures == 0 and ports:
        print(f"PASS: dashboards healthy on ports {', '.join(ports)}")
        return 0

    print(f"FAIL: {failures} issue(s) found")
    return 1


if __name__ == "__main__":
    sys.exit(main())