ableton-mcp-ai/AbletonMCP_AI/MCP_Server/health_check.py

"""
health_check.py - Verificación de salud del sistema
T107-T110: Health checks
"""
import sys
import os
import socket
import json
import logging
from pathlib import Path
from typing import Dict, Any, List

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("HealthCheck")


class AbletonMCPHealthCheck:
    """Verifica la salud del sistema AbletonMCP-AI."""

    def __init__(self):
        self.checks: List[Dict[str, Any]] = []
        self.all_passed = True

    def check_ableton_connection(self) -> bool:
        """Verifica conexión a Ableton Live."""
        try:
            # Intentar conectar al socket de Ableton
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex(('127.0.0.1', 9877))
            sock.close()

            if result == 0:
                self._add_check("Ableton Connection", True, "Connected on port 9877")
                return True
            else:
                self._add_check("Ableton Connection", False, f"Port 9877 not available (code {result})")
                return False
        except Exception as e:
            self._add_check("Ableton Connection", False, str(e))
            return False

    def check_mcp_server(self) -> bool:
        """Verifica que el servidor MCP responde."""
        try:
            # Intentar importar el módulo
            from full_integration import AbletonMCPFullPipeline
            pipeline = AbletonMCPFullPipeline()

            self._add_check("MCP Server", True, "Module imports successfully")
            return True
        except Exception as e:
            self._add_check("MCP Server", False, f"Import error: {e}")
            return False

    def check_sample_library(self) -> bool:
        """Verifica librería de samples."""
        lib_paths = [
            Path("librerias/reggaeton"),  # Primary: reggaeton library
            Path.home() / "embeddings" / "reggaeton",
            Path("librerias/all_tracks"),  # Fallback: flat structure
            Path.home() / "embeddings" / "all_tracks",
        ]

        for path in lib_paths:
            if path.exists():
                wav_files = list(path.rglob("*.wav"))
                if len(wav_files) > 0:
                    self._add_check("Sample Library", True, f"{len(wav_files)} samples at {path}")
                    return True

        self._add_check("Sample Library", False, "No sample library found")
        return False

    def check_dependencies(self) -> bool:
        """Verifica dependencias de Python."""
        required = [
            'numpy',
            'sklearn',
            'sentence_transformers',
        ]

        missing = []
        for dep in required:
            try:
                __import__(dep)
            except ImportError:
                missing.append(dep)

        if missing:
            self._add_check("Dependencies", False, f"Missing: {', '.join(missing)}")
            return False

        self._add_check("Dependencies", True, "All required packages available")
        return True

    def check_vector_index(self) -> bool:
        """Verifica índice de vectores."""
        index_paths = [
            Path("librerias/reggaeton/.sample_embeddings.json"),  # Primary
            Path.home() / "embeddings" / "reggaeton" / ".sample_embeddings.json",
            Path("librerias/all_tracks/.sample_embeddings.json"),  # Fallback
            Path.home() / "embeddings" / "all_tracks" / ".sample_embeddings.json",
        ]

        for path in index_paths:
            if path.exists():
                self._add_check("Vector Index", True, f"Index at {path}")
                return True

        self._add_check("Vector Index", False, "No index found - will be built on first run")
        return False

    def check_persistence_files(self) -> bool:
        """Verifica archivos de persistencia."""
        data_dir = Path.home() / ".abletonmcp_ai"

        files_to_check = [
            "sample_history.json",
            "sample_fatigue.json",
            "collection_coverage.json",
        ]

        all_ok = True
        for file in files_to_check:
            path = data_dir / file
            if path.exists():
                self._add_check(f"Persistence: {file}", True, "File exists")
            else:
                self._add_check(f"Persistence: {file}", False, "Will be created")
                all_ok = False

        return all_ok

    def check_tests(self) -> bool:
        """Verifica que los tests pasan."""
        try:
            import subprocess
            result = subprocess.run(
                [sys.executable, "-m", "unittest", "tests.test_human_feel", "-v"],
                capture_output=True,
                timeout=30,
                cwd=Path(__file__).parent
            )

            if result.returncode == 0:
                self._add_check("Unit Tests", True, "All tests passing")
                return True
            else:
                self._add_check("Unit Tests", False, "Some tests failed")
                return False
        except Exception as e:
            self._add_check("Unit Tests", False, f"Error running tests: {e}")
            return False

    def _add_check(self, name: str, passed: bool, message: str):
        """Agrega un check al reporte."""
        self.checks.append({
            'name': name,
            'passed': passed,
            'message': message
        })
        if not passed:
            self.all_passed = False

    def run_all_checks(self) -> Dict[str, Any]:
        """Ejecuta todos los checks."""
        logger.info("Running health checks...")
        logger.info("=" * 50)

        self.check_ableton_connection()
        self.check_mcp_server()
        self.check_sample_library()
        self.check_dependencies()
        self.check_vector_index()
        self.check_persistence_files()
        self.check_tests()

        # Summary
        passed = sum(1 for c in self.checks if c['passed'])
        total = len(self.checks)

        logger.info("=" * 50)
        logger.info(f"RESULT: {passed}/{total} checks passed")

        return {
            'all_passed': self.all_passed,
            'passed': passed,
            'total': total,
            'checks': self.checks
        }


def main():
    """Ejecuta health check desde línea de comandos."""
    checker = AbletonMCPHealthCheck()
    result = checker.run_all_checks()

    # Guardar resultado
    output_path = Path("health_check_result.json")
    with open(output_path, 'w') as f:
        json.dump(result, f, indent=2)

    # Exit code
    sys.exit(0 if result['all_passed'] else 1)


if __name__ == '__main__':
    main()