alexei.dolgolyov
e26df64e4b
Split monorepo into separate units for future independent repositories: - media-server/: Standalone FastAPI server with own README, requirements, config example, and CLAUDE.md - haos-integration/: HACS-ready Home Assistant integration with hacs.json, own README, and CLAUDE.md Both components now have their own .gitignore files and can be easily extracted into separate repositories. Also adds custom icon support for scripts configuration. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
233 lines
7.6 KiB
Python
233 lines
7.6 KiB
Python
"""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
|