utils.py

"""
VARIABOT Multi-Library Integration System
Advanced integration framework for seamless bot cooperation and Android/Termux optimization

This module provides comprehensive integration capabilities including:
- Multi-library orchestration
- Cross-platform compatibility (Android 10+, Termux)
- Seamless integration with existing bot formats
- Resource-aware operations for mobile devices
- Advanced AI model management
"""

import os
import sys
import asyncio
import logging
import json
import platform
import subprocess
from pathlib import Path
from typing import Dict, List, Any, Optional, Union, Callable
from dataclasses import dataclass, field
from enum import Enum
import concurrent.futures
import threading
import queue
import time
from contextlib import contextmanager

# Core imports with fallback handling for restricted environments
try:
    import streamlit as st
except ImportError:
    st = None

try:
    from gradio_client import Client
except ImportError:
    Client = None

try:
    import torch
    import transformers
except ImportError:
    torch = None
    transformers = None

try:
    import numpy as np
    import pandas as pd
except ImportError:
    np = None
    pd = None

try:
    from flask import Flask, request, jsonify
    from fastapi import FastAPI
except ImportError:
    Flask = None
    FastAPI = None

# Mobile and platform-specific imports
try:
    from plyer import notification, battery, gps

    MOBILE_FEATURES = True
except ImportError:
    MOBILE_FEATURES = False

try:
    from pyjnius import autoclass, cast

    ANDROID_FEATURES = True
except ImportError:
    ANDROID_FEATURES = False

# Configuration and logging
from loguru import logger


class PlatformType(Enum):
    LINUX_DESKTOP = "linux_desktop"
    ANDROID_TERMUX = "android_termux"
    ANDROID_NATIVE = "android_native"
    KALI_LINUX = "kali_linux"
    UNKNOWN = "unknown"


class ResourceProfile(Enum):
    HIGH_PERFORMANCE = "high_performance"
    BALANCED = "balanced"
    POWER_SAVING = "power_saving"
    MINIMAL = "minimal"


@dataclass
class SystemCapabilities:
    """System capability assessment for optimal configuration."""

    cpu_cores: int = 0
    memory_gb: float = 0.0
    storage_gb: float = 0.0
    gpu_available: bool = False
    network_available: bool = False
    battery_powered: bool = False
    platform: PlatformType = PlatformType.UNKNOWN
    android_version: Optional[int] = None
    termux_available: bool = False


@dataclass
class IntegrationConfig:
    """Configuration for multi-library integration."""

    enabled_libraries: List[str] = field(default_factory=list)
    resource_profile: ResourceProfile = ResourceProfile.BALANCED
    android_optimized: bool = False
    termux_mode: bool = False
    max_concurrent_models: int = 2
    memory_limit_mb: int = 1024
    auto_scaling: bool = True
    fallback_models: List[str] = field(default_factory=list)


class PlatformDetector:
    """Advanced platform detection and capability assessment."""

    @staticmethod
    def detect_platform() -> PlatformType:
        """Detect the current platform with Android-specific checks."""
        system = platform.system().lower()

        # Check for Android
        if "android" in system or os.path.exists("/system/build.prop"):
            return PlatformType.ANDROID_NATIVE

        # Check for Termux
        if os.environ.get("PREFIX", "").endswith("com.termux"):
            return PlatformType.ANDROID_TERMUX

        # Check for Kali Linux
        if os.path.exists("/etc/kali_version"):
            return PlatformType.KALI_LINUX

        # Default to Linux desktop
        if system == "linux":
            return PlatformType.LINUX_DESKTOP

        return PlatformType.UNKNOWN

    @staticmethod
    def get_android_version() -> Optional[int]:
        """Get Android version if running on Android."""
        try:
            if os.path.exists("/system/build.prop"):
                with open("/system/build.prop", "r") as f:
                    for line in f:
                        if "ro.build.version.sdk=" in line:
                            return int(line.split("=")[1].strip())
        except:
            pass

        # Try via Termux API
        try:
            result = subprocess.run(["termux-info"], capture_output=True, text=True)
            if result.returncode == 0:
                info = json.loads(result.stdout)
                return info.get("android_version")
        except:
            pass

        return None

    @staticmethod
    def assess_capabilities() -> SystemCapabilities:
        """Comprehensive system capability assessment."""
        capabilities = SystemCapabilities()

        # Platform detection
        capabilities.platform = PlatformDetector.detect_platform()
        capabilities.android_version = PlatformDetector.get_android_version()
        capabilities.termux_available = os.path.exists("/data/data/com.termux")

        # System resources
        try:
            import psutil

            capabilities.cpu_cores = psutil.cpu_count()
            capabilities.memory_gb = psutil.virtual_memory().total / (1024**3)
            capabilities.storage_gb = psutil.disk_usage("/").total / (1024**3)
            capabilities.network_available = len(psutil.net_if_addrs()) > 1
        except ImportError:
            # Fallback methods for restricted environments
            capabilities.cpu_cores = os.cpu_count() or 1
            capabilities.memory_gb = 1.0  # Conservative estimate

        # GPU detection
        capabilities.gpu_available = torch is not None and torch.cuda.is_available()

        # Battery detection (mobile devices)
        if MOBILE_FEATURES:
            try:
                battery_info = battery.status
                capabilities.battery_powered = battery_info is not None
            except:
                capabilities.battery_powered = capabilities.platform in [
                    PlatformType.ANDROID_TERMUX,
                    PlatformType.ANDROID_NATIVE,
                ]

        return capabilities


class LibraryManager:
    """Manages dynamic library loading and fallback mechanisms."""

    def __init__(self):
        self.loaded_libraries = {}
        self.failed_libraries = set()
        self.capabilities = PlatformDetector.assess_capabilities()

    def load_library(self, library_name: str, required: bool = False) -> bool:
        """Dynamically load library with error handling."""
        if library_name in self.loaded_libraries:
            return True

        if library_name in self.failed_libraries:
            return False

        try:
            if library_name == "streamlit":
                import streamlit as st

                self.loaded_libraries["streamlit"] = st
            elif library_name == "gradio":
                from gradio_client import Client

                self.loaded_libraries["gradio"] = Client
            elif library_name == "torch":
                import torch

                self.loaded_libraries["torch"] = torch
            elif library_name == "transformers":
                import transformers

                self.loaded_libraries["transformers"] = transformers
            elif library_name == "kivy":
                import kivy

                self.loaded_libraries["kivy"] = kivy
            # Add more libraries as needed

            logger.info(f"Successfully loaded {library_name}")
            return True

        except ImportError as e:
            logger.warning(f"Failed to load {library_name}: {e}")
            self.failed_libraries.add(library_name)

            if required:
                raise ImportError(f"Required library {library_name} not available")

            return False

    def get_library(self, library_name: str):
        """Get loaded library or None if not available."""
        return self.loaded_libraries.get(library_name)

    def is_available(self, library_name: str) -> bool:
        """Check if library is available."""
        return library_name in self.loaded_libraries


class AndroidOptimizer:
    """Android and Termux specific optimizations."""

    def __init__(self, capabilities: SystemCapabilities):
        self.capabilities = capabilities
        self.android_context = None

        if ANDROID_FEATURES:
            self._setup_android_context()

    def _setup_android_context(self):
        """Setup Android context for native features."""
        try:
            PythonActivity = autoclass("org.kivy.android.PythonActivity")
            self.android_context = PythonActivity.mActivity
        except:
            logger.warning("Could not setup Android context")

    def optimize_for_mobile(self) -> Dict[str, Any]:
        """Apply mobile-specific optimizations."""
        optimizations = {
            "memory_limit": min(1024, int(self.capabilities.memory_gb * 512)),
            "cpu_threads": min(2, self.capabilities.cpu_cores),
            "cache_size": 100,  # MB
            "aggressive_gc": True,
            "low_power_mode": self.capabilities.battery_powered,
        }

        # Android version specific optimizations
        if self.capabilities.android_version:
            if self.capabilities.android_version >= 13:
                optimizations["use_scoped_storage"] = True
                optimizations["runtime_permissions"] = True
            elif self.capabilities.android_version >= 10:
                optimizations["background_restrictions"] = True

        return optimizations

    def setup_termux_environment(self):
        """Setup Termux-specific environment optimizations."""
        if not self.capabilities.termux_available:
            return

        # Setup Termux paths
        termux_paths = {
            "PREFIX": os.environ.get("PREFIX", "/data/data/com.termux/files/usr"),
            "HOME": os.environ.get("HOME", "/data/data/com.termux/files/home"),
            "TMPDIR": os.environ.get("TMPDIR", "/data/data/com.termux/files/usr/tmp"),
        }

        # Update environment
        for key, path in termux_paths.items():
            os.environ[key] = path

        # Setup Python path for Termux
        if "PYTHONPATH" not in os.environ:
            python_path = f"{termux_paths['PREFIX']}/lib/python3.11/site-packages"
            os.environ["PYTHONPATH"] = python_path

        logger.info("Termux environment configured")


class ModelIntegrationManager:
    """Manages integration between different AI models and bot formats."""

    def __init__(self, config: IntegrationConfig):
        self.config = config
        self.library_manager = LibraryManager()
        self.capabilities = PlatformDetector.assess_capabilities()
        self.android_optimizer = AndroidOptimizer(self.capabilities)
        self.active_models = {}
        self.model_queue = queue.Queue()

        # Apply mobile optimizations
        if self.capabilities.platform in [
            PlatformType.ANDROID_TERMUX,
            PlatformType.ANDROID_NATIVE,
        ]:
            self.mobile_optimizations = self.android_optimizer.optimize_for_mobile()
            self.config.android_optimized = True
            self.android_optimizer.setup_termux_environment()

    def register_existing_bot_integration(self, bot_files: List[str]):
        """Register and integrate with existing bot formats."""
        integrated_bots = {}

        for bot_file in bot_files:
            if not os.path.exists(bot_file):
                continue

            bot_name = Path(bot_file).stem

            # Analyze bot file for integration points
            with open(bot_file, "r") as f:
                content = f.read()

            integration_info = {
                "file_path": bot_file,
                "has_streamlit": "streamlit" in content,
                "has_gradio": "gradio" in content,
                "model_type": self._detect_model_type(content),
                "chat_interface": "chat_message" in content
                or "st.chat_input" in content,
                "history_logging": "writehistory" in content,
                "compatible": True,
            }

            integrated_bots[bot_name] = integration_info
            logger.info(
                f"Registered bot: {bot_name} - {integration_info['model_type']}"
            )

        return integrated_bots

    def _detect_model_type(self, content: str) -> str:
        """Detect the AI model type from bot content."""
        if "qwen" in content.lower():
            return "qwen"
        elif "phi" in content.lower():
            return "phi3"
        elif "openelm" in content.lower():
            return "openelm"
        elif "codet5" in content.lower():
            return "codet5"
        elif "tinyllama" in content.lower():
            return "tinyllama"
        else:
            return "unknown"

    @contextmanager
    def resource_managed_execution(self):
        """Context manager for resource-aware execution."""
        start_memory = self._get_memory_usage()

        try:
            yield
        finally:
            end_memory = self._get_memory_usage()
            memory_used = end_memory - start_memory

            if memory_used > self.config.memory_limit_mb:
                logger.warning(
                    f"Memory usage {memory_used}MB exceeded limit {self.config.memory_limit_mb}MB"
                )
                self._trigger_garbage_collection()

    def _get_memory_usage(self) -> float:
        """Get current memory usage in MB."""
        try:
            import psutil

            process = psutil.Process()
            return process.memory_info().rss / (1024 * 1024)
        except:
            return 0.0

    def _trigger_garbage_collection(self):
        """Force garbage collection for memory cleanup."""
        import gc

        gc.collect()

        if torch and torch.cuda.is_available():
            torch.cuda.empty_cache()

    async def async_model_execution(self, model_call: Callable, *args, **kwargs):
        """Execute model calls asynchronously with resource management."""
        with self.resource_managed_execution():
            loop = asyncio.get_event_loop()

            # Use thread pool for CPU-bound model operations
            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
                result = await loop.run_in_executor(
                    executor, model_call, *args, **kwargs
                )

            return result

    def create_unified_interface(self) -> Dict[str, Any]:
        """Create unified interface for all integrated bots."""
        interface_config = {
            "platform": self.capabilities.platform.value,
            "android_version": self.capabilities.android_version,
            "optimized_for_mobile": self.config.android_optimized,
            "available_libraries": list(self.library_manager.loaded_libraries.keys()),
            "resource_profile": self.config.resource_profile.value,
            "max_memory_mb": self.config.memory_limit_mb,
        }

        if self.config.android_optimized:
            interface_config.update(self.mobile_optimizations)

        return interface_config


# Global integration manager instance
integration_manager = None


def initialize_integration(
    config: Optional[IntegrationConfig] = None,
) -> ModelIntegrationManager:
    """Initialize the global integration manager."""
    global integration_manager

    if config is None:
        # Create default configuration based on platform
        capabilities = PlatformDetector.assess_capabilities()

        config = IntegrationConfig(
            enabled_libraries=["streamlit", "gradio", "torch"],
            resource_profile=ResourceProfile.BALANCED,
            android_optimized=capabilities.platform
            in [PlatformType.ANDROID_TERMUX, PlatformType.ANDROID_NATIVE],
            termux_mode=capabilities.termux_available,
            max_concurrent_models=2 if capabilities.memory_gb < 4 else 4,
            memory_limit_mb=(
                int(capabilities.memory_gb * 512)
                if capabilities.memory_gb > 0
                else 1024
            ),
            auto_scaling=True,
            fallback_models=["codet5-small", "tinyllama"],
        )

    integration_manager = ModelIntegrationManager(config)

    # Load essential libraries
    essential_libs = ["streamlit", "gradio"]
    for lib in essential_libs:
        integration_manager.library_manager.load_library(lib, required=False)

    logger.info(f"Integration manager initialized for {capabilities.platform.value}")
    return integration_manager


def get_integration_manager() -> Optional[ModelIntegrationManager]:
    """Get the global integration manager instance."""
    return integration_manager


# Utility functions for existing bot integration
def integrate_with_existing_bots() -> Dict[str, Any]:
    """Integrate with all existing bot files in the repository."""
    if not integration_manager:
        initialize_integration()

    # Find all existing bot files
    bot_files = [
        "st-Qwen1.5-110B-Chat.py",
        "st-Phi3Mini-128k-Chat.py",
        "st-Openelm-3B.py",
        "st-Qwen1.5-MoE-A2.7B-Chat.py",
        "st-codet5-small.py",
        "st-tinyllama-chat.py",
        "Qwen110BChat.py",
    ]

    return integration_manager.register_existing_bot_integration(bot_files)


def get_mobile_optimized_config() -> Dict[str, Any]:
    """Get configuration optimized for mobile/Android deployment."""
    if not integration_manager:
        initialize_integration()

    return integration_manager.create_unified_interface()


# Example usage and testing
if __name__ == "__main__":
    # Initialize integration system
    manager = initialize_integration()

    # Test platform detection
    capabilities = PlatformDetector.assess_capabilities()
    print(f"Platform: {capabilities.platform.value}")
    print(f"Android Version: {capabilities.android_version}")
    print(f"Memory: {capabilities.memory_gb:.1f}GB")
    print(f"CPU Cores: {capabilities.cpu_cores}")

    # Test bot integration
    integrated_bots = integrate_with_existing_bots()
    print(f"Integrated bots: {list(integrated_bots.keys())}")

    # Test mobile configuration
    mobile_config = get_mobile_optimized_config()
    print(f"Mobile optimized: {mobile_config['optimized_for_mobile']}")