media-player-mixed/media_server/services/android_media.py

"""Android media controller using Termux:API.

This controller is designed to run on Android devices using Termux.
It requires the Termux:API app and termux-api package to be installed.

Installation:
1. Install Termux from F-Droid (not Play Store)
2. Install Termux:API from F-Droid
3. In Termux: pkg install termux-api
4. Grant necessary permissions to Termux:API
"""

import asyncio
import json
import logging
import subprocess
from typing import Optional, Any

from ..models import MediaState, MediaStatus
from .media_controller import MediaController

logger = logging.getLogger(__name__)


def _check_termux_api() -> bool:
    """Check if termux-api is available."""
    try:
        result = subprocess.run(
            ["which", "termux-media-player"],
            capture_output=True,
            timeout=5,
        )
        return result.returncode == 0
    except Exception:
        return False


TERMUX_API_AVAILABLE = _check_termux_api()


class AndroidMediaController(MediaController):
    """Media controller for Android using Termux:API.

    Requires:
    - Termux app
    - Termux:API app
    - termux-api package (pkg install termux-api)
    """

    def __init__(self):
        if not TERMUX_API_AVAILABLE:
            logger.warning(
                "Termux:API not available. Install with: pkg install termux-api"
            )

    def _run_termux_command(
        self, command: list[str], timeout: int = 10
    ) -> Optional[str]:
        """Run a termux-api command and return the output."""
        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
            if result.returncode == 0:
                return result.stdout.strip()
            logger.error(f"Termux command failed: {result.stderr}")
            return None
        except subprocess.TimeoutExpired:
            logger.error(f"Termux command timed out: {command}")
            return None
        except Exception as e:
            logger.error(f"Termux command error: {e}")
            return None

    def _send_media_key(self, key: str) -> bool:
        """Send a media key event.

        Args:
            key: One of: play, pause, play-pause, stop, next, previous
        """
        # termux-media-player command
        result = self._run_termux_command(["termux-media-player", key])
        return result is not None

    def _get_media_info(self) -> dict[str, Any]:
        """Get current media playback info using termux-media-player."""
        result = self._run_termux_command(["termux-media-player", "info"])
        if result:
            try:
                return json.loads(result)
            except json.JSONDecodeError:
                pass
        return {}

    def _get_volume(self) -> tuple[int, bool]:
        """Get current volume using termux-volume."""
        result = self._run_termux_command(["termux-volume"])
        if result:
            try:
                volumes = json.loads(result)
                # Find music stream
                for stream in volumes:
                    if stream.get("stream") == "music":
                        volume = stream.get("volume", 0)
                        max_volume = stream.get("max_volume", 15)
                        # Convert to 0-100 scale
                        percent = int((volume / max_volume) * 100) if max_volume > 0 else 0
                        return percent, False
            except (json.JSONDecodeError, KeyError):
                pass
        return 100, False

    def _set_volume_internal(self, volume: int) -> bool:
        """Set volume using termux-volume."""
        # termux-volume expects stream name and volume level
        # Convert 0-100 to device scale (usually 0-15)
        result = self._run_termux_command(["termux-volume"])
        if result:
            try:
                volumes = json.loads(result)
                for stream in volumes:
                    if stream.get("stream") == "music":
                        max_volume = stream.get("max_volume", 15)
                        device_volume = int((volume / 100) * max_volume)
                        self._run_termux_command(
                            ["termux-volume", "music", str(device_volume)]
                        )
                        return True
            except (json.JSONDecodeError, KeyError):
                pass
        return False

    async def get_status(self) -> MediaStatus:
        """Get current media playback status."""
        status = MediaStatus()

        # Get volume
        volume, muted = self._get_volume()
        status.volume = volume
        status.muted = muted

        # Get media info
        info = self._get_media_info()
        if not info:
            status.state = MediaState.IDLE
            return status

        # Parse playback status
        playback_status = info.get("status", "").lower()
        if playback_status == "playing":
            status.state = MediaState.PLAYING
        elif playback_status == "paused":
            status.state = MediaState.PAUSED
        elif playback_status == "stopped":
            status.state = MediaState.STOPPED
        else:
            status.state = MediaState.IDLE

        # Parse track info
        status.title = info.get("title") or info.get("Track") or None
        status.artist = info.get("artist") or info.get("Artist") or None
        status.album = info.get("album") or info.get("Album") or None

        # Duration and position (in milliseconds from some sources)
        duration = info.get("duration", 0)
        if duration > 1000:  # Likely milliseconds
            duration = duration / 1000
        status.duration = duration if duration > 0 else None

        position = info.get("position", info.get("current_position", 0))
        if position > 1000:  # Likely milliseconds
            position = position / 1000
        status.position = position if position > 0 else None

        status.source = "Android"

        return status

    async def play(self) -> bool:
        """Resume playback."""
        return self._send_media_key("play")

    async def pause(self) -> bool:
        """Pause playback."""
        return self._send_media_key("pause")

    async def stop(self) -> bool:
        """Stop playback."""
        return self._send_media_key("stop")

    async def next_track(self) -> bool:
        """Skip to next track."""
        return self._send_media_key("next")

    async def previous_track(self) -> bool:
        """Go to previous track."""
        return self._send_media_key("previous")

    async def set_volume(self, volume: int) -> bool:
        """Set system volume."""
        return self._set_volume_internal(volume)

    async def toggle_mute(self) -> bool:
        """Toggle mute state.

        Note: Android doesn't have a simple mute toggle via termux-api,
        so we set volume to 0 or restore previous volume.
        """
        volume, _ = self._get_volume()
        if volume > 0:
            # Store current volume and mute
            self._previous_volume = volume
            self._set_volume_internal(0)
            return True
        else:
            # Restore previous volume
            prev = getattr(self, "_previous_volume", 50)
            self._set_volume_internal(prev)
            return False

    async def seek(self, position: float) -> bool:
        """Seek to position in seconds.

        Note: Seek functionality may be limited depending on the media player.
        """
        # termux-media-player doesn't support seek directly
        # This is a limitation of the API
        logger.warning("Seek not fully supported on Android via Termux:API")
        return False