LEDMatrix/plugin-repos/web-ui-info/manager.py

"""
Web UI Info Plugin for LEDMatrix

A simple plugin that displays the web UI URL for easy access.
Shows "visit web ui at http://[deviceID]:5000"

API Version: 1.0.0
"""

import logging
import os
import socket
import subprocess
import time
from pathlib import Path
from typing import Dict, Any, Optional
from PIL import Image, ImageDraw, ImageFont

from src.plugin_system.base_plugin import BasePlugin

logger = logging.getLogger(__name__)


class WebUIInfoPlugin(BasePlugin):
    """
    Web UI Info plugin that displays the web UI URL.
    
    Configuration options:
        display_duration (float): Display duration in seconds (default: 10)
        enabled (bool): Enable/disable plugin (default: true)
    """
    
    def __init__(self, plugin_id: str, config: Dict[str, Any],
                 display_manager, cache_manager, plugin_manager):
        """Initialize the Web UI Info plugin."""
        super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
        
        # Get device hostname
        try:
            self.device_id = socket.gethostname()
        except Exception as e:
            self.logger.warning(f"Could not get hostname: {e}, using 'localhost'")
            self.device_id = "localhost"
        
        # Get device IP address
        self.device_ip = self._get_local_ip()
        
        # IP refresh tracking
        self.last_ip_refresh = time.time()
        self.ip_refresh_interval = 300.0  # Refresh IP every 5 minutes

        # AP mode cache
        self._ap_mode_cached = False
        self._ap_mode_cache_time = 0.0
        self._ap_mode_cache_ttl = 60.0  # Cache AP mode check for 60 seconds

        # Rotation state
        self.current_display_mode = "hostname"  # "hostname" or "ip"
        self.last_rotation_time = time.time()
        self.rotation_interval = 10.0  # Rotate every 10 seconds

        self.web_ui_url = f"http://{self.device_id}:5000"

        # Display cache - avoid re-rendering when nothing changed
        self._cached_display_image = None
        self._display_dirty = True
        self._font_small = self._load_font()

        self.logger.info(f"Web UI Info plugin initialized - Hostname: {self.device_id}, IP: {self.device_ip}")
    
    def _load_font(self):
        """Load and cache the display font."""
        try:
            current_dir = Path(__file__).resolve().parent
            project_root = current_dir.parent.parent
            font_path = project_root / "assets" / "fonts" / "4x6-font.ttf"
            if font_path.exists():
                return ImageFont.truetype(str(font_path), 6)
            font_path = "assets/fonts/4x6-font.ttf"
            if os.path.exists(font_path):
                return ImageFont.truetype(font_path, 6)
            return ImageFont.load_default()
        except (FileNotFoundError, OSError) as e:
            self.logger.debug(f"Could not load custom font: {e}, using default")
            return ImageFont.load_default()

    def _is_ap_mode_active(self) -> bool:
        """
        Check if AP mode is currently active (cached with TTL).

        Returns:
            bool: True if AP mode is active, False otherwise
        """
        current_time = time.time()
        if current_time - self._ap_mode_cache_time < self._ap_mode_cache_ttl:
            return self._ap_mode_cached

        try:
            result = subprocess.run(
                ["systemctl", "is-active", "hostapd"],
                capture_output=True,
                text=True,
                timeout=2
            )
            if result.returncode == 0 and result.stdout.strip() == "active":
                self._ap_mode_cached = True
                self._ap_mode_cache_time = current_time
                return True

            result = subprocess.run(
                ["ip", "addr", "show", "wlan0"],
                capture_output=True,
                text=True,
                timeout=2
            )
            if result.returncode == 0 and "192.168.4.1" in result.stdout:
                self._ap_mode_cached = True
                self._ap_mode_cache_time = current_time
                return True

            self._ap_mode_cached = False
            self._ap_mode_cache_time = current_time
            return False
        except Exception as e:
            self.logger.debug(f"Error checking AP mode status: {e}")
            self._ap_mode_cached = False
            self._ap_mode_cache_time = current_time
            return False
    
    def _get_local_ip(self) -> str:
        """
        Get the local IP address of the device using network interfaces.
        Handles AP mode, no internet connectivity, and network state changes.

        Returns:
            str: Local IP address, or "localhost" if unable to determine
        """
        # First check if AP mode is active
        if self._is_ap_mode_active():
            self.logger.debug("AP mode detected, returning AP IP: 192.168.4.1")
            return "192.168.4.1"

        try:
            # Try socket method first (zero subprocess overhead, fastest)
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                try:
                    s.connect(('8.8.8.8', 80))
                    ip = s.getsockname()[0]
                    if ip and not ip.startswith("127.") and ip != "192.168.4.1":
                        self.logger.debug(f"Found IP via socket method: {ip}")
                        return ip
                finally:
                    s.close()
            except Exception:
                pass

            # Fallback: Try using 'hostname -I'
            result = subprocess.run(
                ["hostname", "-I"],
                capture_output=True,
                text=True,
                timeout=2
            )
            if result.returncode == 0:
                ips = result.stdout.strip().split()
                for ip in ips:
                    ip = ip.strip()
                    if ip and not ip.startswith("127.") and ip != "192.168.4.1":
                        self.logger.debug(f"Found IP via hostname -I: {ip}")
                        return ip

            # Fallback: Use 'ip addr show' to get interface IPs
            result = subprocess.run(
                ["ip", "-4", "addr", "show"],
                capture_output=True,
                text=True,
                timeout=3
            )
            if result.returncode == 0:
                current_interface = None
                for line in result.stdout.split('\n'):
                    line = line.strip()
                    if ':' in line and not line.startswith('inet'):
                        parts = line.split(':')
                        if len(parts) >= 2:
                            current_interface = parts[1].strip().split('@')[0]
                    elif line.startswith('inet '):
                        parts = line.split()
                        if len(parts) >= 2:
                            ip_with_cidr = parts[1]
                            ip = ip_with_cidr.split('/')[0]
                            if not ip.startswith("127.") and ip != "192.168.4.1":
                                if current_interface and (
                                    current_interface.startswith("eth") or
                                    current_interface.startswith("enp")
                                ):
                                    self.logger.debug(f"Found Ethernet IP: {ip} on {current_interface}")
                                    return ip
                                elif current_interface == "wlan0":
                                    self.logger.debug(f"Found WiFi IP: {ip} on {current_interface}")
                                    return ip
            except Exception:
                pass
            
            # Last resort: try hostname resolution (often returns 127.0.0.1)
            try:
                ip = socket.gethostbyname(socket.gethostname())
                if ip and not ip.startswith("127.") and ip != "192.168.4.1":
                    self.logger.debug(f"Found IP via hostname resolution: {ip}")
                    return ip
            except Exception:
                pass
            
            self.logger.warning("Could not determine IP address, using 'localhost'")
            return "localhost"
            
        except Exception as e:
            self.logger.warning(f"Error getting IP address: {e}, using 'localhost'")
            return "localhost"
    
    def update(self) -> None:
        """
        Update method - refreshes IP address periodically to handle network state changes.

        The hostname is determined at initialization and doesn't change,
        but IP address can change when network state changes (WiFi connect/disconnect, AP mode, etc.)
        """
        current_time = time.time()
        if current_time - self.last_ip_refresh >= self.ip_refresh_interval:
            new_ip = self._get_local_ip()
            if new_ip != self.device_ip:
                self.logger.info(f"IP address changed from {self.device_ip} to {new_ip}")
                self.device_ip = new_ip
                self._display_dirty = True
            self.last_ip_refresh = current_time
    
    def display(self, force_clear: bool = False) -> None:
        """
        Display the web UI URL message.
        Rotates between hostname and IP address every 10 seconds.

        Args:
            force_clear: If True, clear display before rendering
        """
        try:
            # Check if we need to rotate between hostname and IP
            current_time = time.time()
            if current_time - self.last_rotation_time >= self.rotation_interval:
                if self.current_display_mode == "hostname":
                    self.current_display_mode = "ip"
                else:
                    self.current_display_mode = "hostname"
                self.last_rotation_time = current_time
                self._display_dirty = True
                self.logger.debug(f"Rotated to display mode: {self.current_display_mode}")

            if force_clear:
                self.display_manager.clear()
                self._display_dirty = True

            # Use cached image if nothing changed
            if not self._display_dirty and self._cached_display_image is not None:
                self.display_manager.image = self._cached_display_image
                self.display_manager.update_display()
                return

            # Get display dimensions
            width = self.display_manager.matrix.width
            height = self.display_manager.matrix.height

            # Create a new image for the display
            img = Image.new('RGB', (width, height), (0, 0, 0))
            draw = ImageDraw.Draw(img)

            # Determine which address to display
            if self.current_display_mode == "ip":
                address = self.device_ip
            else:
                address = self.device_id

            lines = [
                "visit web ui",
                f"at {address}:5000"
            ]

            y_start = 5
            line_height = 8

            for i, line in enumerate(lines):
                bbox = draw.textbbox((0, 0), line, font=self._font_small)
                text_width = bbox[2] - bbox[0]
                x = (width - text_width) // 2
                y = y_start + (i * line_height)
                draw.text((x, y), line, font=self._font_small, fill=(255, 255, 255))

            self._cached_display_image = img
            self._display_dirty = False
            self.display_manager.image = img
            self.display_manager.update_display()

            self.logger.debug(f"Displayed web UI info: {address}:5000 (mode: {self.current_display_mode})")

        except Exception as e:
            self.logger.error(f"Error displaying web UI info: {e}")
            try:
                self.display_manager.clear()
                self.display_manager.update_display()
            except Exception:
                pass
    
    def get_display_duration(self) -> float:
        """Get display duration from config."""
        return self.config.get('display_duration', 10.0)
    
    def validate_config(self) -> bool:
        """Validate plugin configuration."""
        # Call parent validation first
        if not super().validate_config():
            return False
        
        # No additional validation needed - this is a simple plugin
        return True
    
    def get_info(self) -> Dict[str, Any]:
        """Return plugin info for web UI."""
        info = super().get_info()
        info.update({
            'device_id': self.device_id,
            'device_ip': self.device_ip,
            'web_ui_url': self.web_ui_url,
            'current_display_mode': self.current_display_mode
        })
        return info