app_android.py

import gradio as gr
import openai
import os
import subprocess
import threading
import time
import base64
import io
from PIL import Image
import cv2
import numpy as np

"""

Nebius AI Studio OpenAI-compatible API integration with DeepSeek model + Android Integration

"""

# Initialize OpenAI client with Nebius configuration
def get_openai_client():
    api_key = os.getenv("NEBIUSAISTUDIOAPIKEY")
    if not api_key:
        raise ValueError("NEBIUSAISTUDIOAPIKEY environment variable is not set. Please configure your API key in Hugging Face Space secrets.")
    
    return openai.OpenAI(
        api_key=api_key,
        base_url="https://api.studio.nebius.ai/v1/"
    )

# Initialize client (will be created when needed)
client = None

def respond(message, history, system_message, max_tokens, temperature, top_p):
    global client
    
    # Initialize client if not already done
    if client is None:
        try:
            client = get_openai_client()
        except ValueError as e:
            yield f"Configuration Error: {str(e)}"
            return
    
    messages = [{"role": "system", "content": system_message}]

    for val in history:
        if val[0]:
            messages.append({"role": "user", "content": val[0]})
        if val[1]:
            messages.append({"role": "assistant", "content": val[1]})

    messages.append({"role": "user", "content": message})

    try:
        # Use DeepSeek model through Nebius API
        response = client.chat.completions.create(
            model="deepseek-ai/DeepSeek-V3",
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
            top_p=top_p,
            stream=True,
        )

        partial_message = ""
        for chunk in response:
            if chunk.choices[0].delta.content is not None:
                partial_message += chunk.choices[0].delta.content
                yield partial_message
                
    except Exception as e:
        yield f"Error: {str(e)}. Please check your API key and try again."

# Android Device Management
class AndroidDeviceManager:
    def __init__(self):
        self.connected_devices = []
        self.current_device = None
        self.screenshot_thread = None
        self.is_capturing = False
        self.latest_screenshot = None
        
    def get_connected_devices(self):
        """Get list of connected Android devices via ADB"""
        try:
            result = subprocess.run(['adb', 'devices'], capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                lines = result.stdout.strip().split('\n')[1:]  # Skip header
                devices = []
                for line in lines:
                    if line.strip() and '\tdevice' in line:
                        device_id = line.split('\t')[0]
                        devices.append(device_id)
                self.connected_devices = devices
                return devices
            return []
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            print(f"ADB error: {e}")
            return []
    
    def connect_device(self, device_id):
        """Connect to a specific Android device"""
        try:
            if device_id in self.connected_devices:
                self.current_device = device_id
                return True
            return False
        except Exception as e:
            print(f"Device connection error: {e}")
            return False
    
    def capture_screenshot(self):
        """Capture screenshot from connected Android device"""
        if not self.current_device:
            return None
            
        try:
            result = subprocess.run([
                'adb', '-s', self.current_device, 'exec-out', 'screencap', '-p'
            ], capture_output=True, timeout=15)
            
            if result.returncode == 0:
                # Convert bytes to PIL Image
                img_bytes = result.stdout
                img = Image.open(io.BytesIO(img_bytes))
                
                # Resize for display (maintain aspect ratio)
                max_width = 400
                ratio = max_width / img.width
                new_height = int(img.height * ratio)
                img = img.resize((max_width, new_height), Image.Resampling.LANCZOS)
                
                # Convert to base64 for display in Gradio
                buffered = io.BytesIO()
                img.save(buffered, format="PNG")
                img_b64 = base64.b64encode(buffered.getvalue()).decode()
                
                self.latest_screenshot = f"data:image/png;base64,{img_b64}"
                return self.latest_screenshot
            return None
        except Exception as e:
            print(f"Screenshot error: {e}")
            return None
    
    def get_device_info(self):
        """Get detailed information about the connected device"""
        if not self.current_device:
            return "No device connected"
        
        try:
            # Get device model
            model_result = subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'getprop', 'ro.product.model'
            ], capture_output=True, text=True, timeout=10)
            
            # Get Android version
            version_result = subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'getprop', 'ro.build.version.release'
            ], capture_output=True, text=True, timeout=10)
            
            # Get screen resolution
            resolution_result = subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'wm', 'size'
            ], capture_output=True, text=True, timeout=10)
            
            model = model_result.stdout.strip() if model_result.returncode == 0 else "Unknown"
            version = version_result.stdout.strip() if version_result.returncode == 0 else "Unknown"
            resolution = resolution_result.stdout.strip().split(': ')[-1] if resolution_result.returncode == 0 else "Unknown"
            
            return f"Model: {model}\nAndroid: {version}\nResolution: {resolution}"
        except Exception as e:
            return f"Error getting device info: {e}"
    
    def send_tap(self, x, y):
        """Send tap command to device"""
        if not self.current_device:
            return False
        
        try:
            subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'input', 'tap', str(x), str(y)
            ], timeout=5)
            return True
        except Exception as e:
            print(f"Tap error: {e}")
            return False
    
    def send_text(self, text):
        """Send text input to device"""
        if not self.current_device:
            return False
        
        try:
            subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'input', 'text', f'"{text}"'
            ], timeout=10)
            return True
        except Exception as e:
            print(f"Text input error: {e}")
            return False
    
    def press_key(self, keycode):
        """Press a key on the device (back, home, etc.)"""
        if not self.current_device:
            return False
        
        key_mapping = {
            'back': '4',
            'home': '3',
            'menu': '82',
            'power': '26',
            'volume_up': '24',
            'volume_down': '25'
        }
        
        keycode_num = key_mapping.get(keycode.lower(), keycode)
        
        try:
            subprocess.run([
                'adb', '-s', self.current_device, 'shell', 'input', 'keyevent', keycode_num
            ], timeout=5)
            return True
        except Exception as e:
            print(f"Key press error: {e}")
            return False

# Initialize device manager
device_manager = AndroidDeviceManager()

# Custom CSS for better styling
custom_css = """

.gradio-container {

    font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;

}



.chat-container {

    border-radius: 10px;

    box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);

}



.android-panel {

    background: linear-gradient(135deg, #4CAF50 0%, #45a049 100%);

    border-radius: 10px;

    padding: 15px;

    margin: 10px 0;

}



.device-info {

    background: #f8f9fa;

    border-radius: 8px;

    padding: 10px;

    margin: 5px 0;

    border-left: 4px solid #4CAF50;

}



.status-indicator {

    display: inline-block;

    width: 10px;

    height: 10px;

    border-radius: 50%;

    background-color: #28a745;

    margin-right: 5px;

}



h1 {

    background: linear-gradient(90deg, #667eea, #764ba2);

    -webkit-background-clip: text;

    -webkit-text-fill-color: transparent;

    text-align: center;

    font-size: 2.5em;

    margin-bottom: 10px;

}



.device-screen {

    border: 2px solid #ddd;

    border-radius: 20px;

    padding: 10px;

    background: #000;

}

"""

# Pre-defined system prompts for different use cases
system_prompts = {
    "Android App Tester": "You are an expert Android app tester and QA engineer. Help analyze Android apps, identify bugs, suggest improvements, and guide testing procedures. You can see the device screen and help with app navigation.",
    "UX/UI Analyzer": "You are a UX/UI expert specializing in mobile app design. Analyze Android app interfaces, provide design feedback, suggest improvements for user experience, and evaluate accessibility.",
    "Android Developer": "You are an expert Android developer. Help with app development, debugging, code review, and provide solutions for Android-specific challenges. You can analyze app behavior through screen captures.",
    "General Assistant": "You are a helpful and knowledgeable AI assistant. Provide accurate, detailed, and thoughtful responses.",
    "Code Expert": "You are an expert programmer and software engineer. Help with coding questions, debugging, code review, and best practices across multiple programming languages."
}

def create_android_interface():
    with gr.Blocks(css=custom_css, title="DeepSeek AI Studio - Android App Testing Interface") as demo:
        # Header
        gr.HTML("""

        <h1>🚀 DeepSeek AI Studio + Android Integration</h1>

        <div class="device-info">

            <p><span class="status-indicator"></span><strong>Model:</strong> DeepSeek-V3 via Nebius AI Studio</p>

            <p><strong>Features:</strong> Android app testing, screen analysis, device control, and AI-powered insights</p>

        </div>

        """)
        
        with gr.Row():
            # Left Column: Chat Interface
            with gr.Column(scale=2):
                gr.HTML("<h3>💬 AI Assistant</h3>")
                chatbot = gr.Chatbot(
                    height=400,
                    show_label=False,
                    container=True,
                    bubble_full_width=False,
                    show_copy_button=True,
                )
                
                with gr.Row():
                    msg = gr.Textbox(
                        placeholder="Ask about the Android app, request analysis, or get testing guidance...",
                        show_label=False,
                        scale=4,
                        lines=2,
                        max_lines=5
                    )
                    submit_btn = gr.Button("Send 🚀", variant="primary", scale=1)
                
                with gr.Row():
                    clear_btn = gr.Button("🗑️ Clear", variant="secondary")
                    analyze_btn = gr.Button("📸 Analyze Screen", variant="primary")
                    export_btn = gr.Button("📄 Export", variant="secondary")
            
            # Middle Column: Android Device Feed
            with gr.Column(scale=2):
                gr.HTML("<h3>📱 Android Device</h3>")
                
                # Device connection controls
                with gr.Row():
                    refresh_btn = gr.Button("🔄", variant="secondary", scale=1)
                    device_dropdown = gr.Dropdown(
                        choices=[],
                        label="Connected Devices",
                        scale=2,
                        interactive=True
                    )
                    connect_btn = gr.Button("📱 Connect", variant="primary", scale=1)
                
                # Device status
                device_status = gr.HTML("""

                <div style="background: #f8f9fa; padding: 10px; border-radius: 8px; margin: 10px 0;">

                    <p><strong>Status:</strong> No device connected</p>

                    <p><strong>Setup:</strong> Connect Android device via USB and enable USB debugging</p>

                </div>

                """)
                
                # Live device screen
                device_screen = gr.Image(
                    height=500,
                    show_label=False,
                    interactive=False,
                    container=True
                )
                
                # Device controls
                with gr.Row():
                    back_btn = gr.Button("⬅️", variant="secondary", scale=1)
                    home_btn = gr.Button("🏠", variant="secondary", scale=1)
                    recent_btn = gr.Button("📱", variant="secondary", scale=1)
                    screenshot_btn = gr.Button("📸", variant="primary", scale=1)
            
            # Right Column: Controls
            with gr.Column(scale=1):
                gr.HTML("<h3>⚙️ Controls</h3>")
                
                # AI Personality
                with gr.Accordion("🎭 AI Role", open=True):
                    prompt_preset = gr.Dropdown(
                        choices=list(system_prompts.keys()),
                        value="Android App Tester",
                        label="Preset",
                        info="Choose AI specialization"
                    )
                    
                    system_message = gr.Textbox(
                        value=system_prompts["Android App Tester"],
                        label="Custom Prompt",
                        lines=3,
                        max_lines=8
                    )
                
                # Generation Settings
                with gr.Accordion("🎛️ AI Settings", open=False):
                    max_tokens = gr.Slider(1, 4096, 2048, label="Max Tokens")
                    temperature = gr.Slider(0.0, 2.0, 0.7, step=0.01, label="Temperature")
                    top_p = gr.Slider(0.0, 1.0, 0.95, step=0.01, label="Top-p")
                
                # Device Interaction
                with gr.Accordion("📱 Device Control", open=True):
                    with gr.Row():
                        tap_x = gr.Number(label="X", value=200, scale=1)
                        tap_y = gr.Number(label="Y", value=400, scale=1)
                    tap_btn = gr.Button("👆 Tap", variant="primary")
                    
                    device_text = gr.Textbox(
                        placeholder="Text to send...",
                        label="Send Text",
                        lines=2
                    )
                    send_text_btn = gr.Button("📝 Send Text", variant="primary")
                
                # Auto features
                with gr.Accordion("🤖 Auto Features", open=False):
                    auto_refresh = gr.Checkbox(label="Auto-refresh screen", value=False)
                    include_screen = gr.Checkbox(label="Include screen in AI context", value=True)
                    refresh_interval = gr.Slider(0.5, 5.0, 2.0, step=0.1, label="Refresh interval (s)")
        
        # Event handlers
        def update_system_prompt(preset):
            return system_prompts.get(preset, system_prompts["Android App Tester"])
        
        def refresh_devices():
            devices = device_manager.get_connected_devices()
            if devices:
                return gr.Dropdown(choices=devices, value=devices[0] if devices else None)
            return gr.Dropdown(choices=[], value=None)
        
        def connect_device(device_id):
            if device_id and device_manager.connect_device(device_id):
                info = device_manager.get_device_info()
                status_html = f"""

                <div style="background: #d4edda; padding: 10px; border-radius: 8px; margin: 10px 0; border-left: 4px solid #28a745;">

                    <p><strong>Status:</strong> ✅ Connected to {device_id}</p>

                    <p><strong>Info:</strong><br>{info.replace(chr(10), '<br>')}</p>

                </div>

                """
                screenshot = device_manager.capture_screenshot()
                return status_html, screenshot
            else:
                return """

                <div style="background: #f8d7da; padding: 10px; border-radius: 8px; margin: 10px 0; border-left: 4px solid #dc3545;">

                    <p><strong>Status:</strong> ❌ Connection failed</p>

                    <p>Check USB debugging and device authorization</p>

                </div>

                """, None
        
        def enhanced_respond_with_screen(message, history, system_msg, max_tok, temp, top_p_val, include_screen):
            enhanced_message = message
            if include_screen and device_manager.current_device and device_manager.latest_screenshot:
                enhanced_message = f"[ANDROID SCREEN CONTEXT: I can see the current Android device screen. Analyze it with this message.]\n\nUser: {message}"
            
            for response in respond(enhanced_message, history, system_msg, max_tok, temp, top_p_val):
                yield response
        
        def analyze_screen():
            if not device_manager.current_device:
                return "No device connected"
            
            screenshot = device_manager.capture_screenshot()
            if screenshot:
                return "Screen captured. Ask AI to analyze what's displayed."
            return "Screenshot failed"
        
        def device_tap(x, y):
            if device_manager.send_tap(int(x), int(y)):
                time.sleep(0.5)
                return device_manager.capture_screenshot()
            return None
        
        def device_send_text(text):
            if device_manager.send_text(text):
                time.sleep(0.5)
                return device_manager.capture_screenshot()
            return None
        
        def device_key(key):
            if device_manager.press_key(key):
                time.sleep(0.5)
                return device_manager.capture_screenshot()
            return None
        
        def take_screenshot():
            return device_manager.capture_screenshot()
        
        # Connect events
        refresh_btn.click(refresh_devices, outputs=[device_dropdown])
        connect_btn.click(connect_device, inputs=[device_dropdown], outputs=[device_status, device_screen])
        
        analyze_btn.click(analyze_screen, outputs=[device_status])
        screenshot_btn.click(take_screenshot, outputs=[device_screen])
        
        tap_btn.click(device_tap, inputs=[tap_x, tap_y], outputs=[device_screen])
        send_text_btn.click(device_send_text, inputs=[device_text], outputs=[device_screen])
        
        back_btn.click(lambda: device_key("back"), outputs=[device_screen])
        home_btn.click(lambda: device_key("home"), outputs=[device_screen])
        recent_btn.click(lambda: device_key("menu"), outputs=[device_screen])
        
        prompt_preset.change(update_system_prompt, inputs=[prompt_preset], outputs=[system_message])
        
        # Chat functionality
        msg.submit(
            enhanced_respond_with_screen,
            inputs=[msg, chatbot, system_message, max_tokens, temperature, top_p, include_screen],
            outputs=[chatbot]
        ).then(lambda: "", outputs=[msg])
        
        submit_btn.click(
            enhanced_respond_with_screen,
            inputs=[msg, chatbot, system_message, max_tokens, temperature, top_p, include_screen],
            outputs=[chatbot]
        ).then(lambda: "", outputs=[msg])
        
        clear_btn.click(lambda: ([], ""), outputs=[chatbot, msg])
        
        # Auto-refresh devices on load
        demo.load(refresh_devices, outputs=[device_dropdown])
        
        # Footer
        gr.HTML("""

        <div style="text-align: center; margin-top: 20px; padding: 10px; background: #f8f9fa; border-radius: 8px;">

            <p style="margin: 0; color: #6c757d;">

                🤖 <strong>DeepSeek-V3</strong> + 📱 <strong>Android ADB</strong> | Built with ❤️ using <strong>Gradio</strong>

            </p>

        </div>

        """)
    
    return demo

# Create the interface
demo = create_android_interface()

if __name__ == "__main__":
    # Check if API key is available
    api_key = os.getenv("NEBIUSAISTUDIOAPIKEY")
    if not api_key:
        print("Warning: NEBIUSAISTUDIOAPIKEY environment variable not set!")
        print("Please set your Nebius AI Studio API key as an environment variable or Hugging Face Space secret.")
        print("The app will start but API calls will fail until the key is configured.")
    else:
        print("API key found. Ready to connect to Nebius AI Studio.")
    
    # Check ADB availability
    try:
        subprocess.run(['adb', 'version'], capture_output=True, check=True)
        print("ADB found. Android device integration ready.")
    except (subprocess.CalledProcessError, FileNotFoundError):
        print("Warning: ADB not found. Please install Android SDK platform-tools.")
        print("Android device integration will not work without ADB.")
    
    demo.launch()