"""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