ffmpeg_proxy.py

Go to the documentation of this file.

"""HTTP view that converts audio from a URL to a preferred format."""
 
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
from http import HTTPStatus
import logging
import secrets
from typing import Final
 
from aiohttp import web
from aiohttp.abc import AbstractStreamWriter, BaseRequest
 
from homeassistant.components.ffmpeg import FFmpegManager
from homeassistant.components.http import HomeAssistantView
from homeassistant.core import HomeAssistant
 
from .const import DATA_FFMPEG_PROXY
 
_LOGGER = logging.getLogger(__name__)
 
_MAX_CONVERSIONS_PER_DEVICE: Final[int] = 2
 
 
def async_create_proxy_url(
    hass: HomeAssistant,
    device_id: str,
    media_url: str,
    media_format: str,
    rate: int | None = None,
    channels: int | None = None,
    width: int | None = None,
) -> str:
    """Create a use proxy URL that automatically converts the media."""
    data: FFmpegProxyData = hass.data[DATA_FFMPEG_PROXY]
    return data.async_create_proxy_url(
        device_id, media_url, media_format, rate, channels, width
    )
 
 
@dataclass
class FFmpegConversionInfo:
    """Information for ffmpeg conversion."""
 
    convert_id: str
    """Unique id for media conversion."""
 
    media_url: str
    """Source URL of media to convert."""
 
    media_format: str
    """Target format for media (mp3, flac, etc.)"""
 
    rate: int | None
    """Target sample rate (None to keep source rate)."""
 
    channels: int | None
    """Target number of channels (None to keep source channels)."""
 
    width: int | None
    """Target sample width in bytes (None to keep source width)."""
 
    proc: asyncio.subprocess.Process | None = None
    """Subprocess doing ffmpeg conversion."""
 
    is_finished: bool = False
    """True if conversion has finished."""
 
 
@dataclass
class FFmpegProxyData:
    """Data for ffmpeg proxy conversion."""
 
    # device_id -> [info]
    conversions: dict[str, list[FFmpegConversionInfo]] = field(
        default_factory=lambda: defaultdict(list)
    )
 
    def async_create_proxy_url(
        self,
        device_id: str,
        media_url: str,
        media_format: str,
        rate: int | None,
        channels: int | None,
        width: int | None,
    ) -> str:
        """Create a one-time use proxy URL that automatically converts the media."""
 
        # Remove completed conversions
        device_conversions = [
            info for info in self.conversions[device_id] if not info.is_finished
        ]
 
        while len(device_conversions) >= _MAX_CONVERSIONS_PER_DEVICE:
            # Stop oldest conversion before adding a new one
            convert_info = device_conversions[0]
            if (convert_info.proc is not None) and (
                convert_info.proc.returncode is None
            ):
                _LOGGER.debug(
                    "Stopping existing ffmpeg process for device: %s", device_id
                )
                convert_info.proc.kill()
 
            device_conversions = device_conversions[1:]
 
        convert_id = secrets.token_urlsafe(16)
        device_conversions.append(
            FFmpegConversionInfo(
                convert_id, media_url, media_format, rate, channels, width
            )
        )
        _LOGGER.debug("Media URL allowed by proxy: %s", media_url)
 
        self.conversions[device_id] = device_conversions
 
        return f"/api/esphome/ffmpeg_proxy/{device_id}/{convert_id}.{media_format}"
 
 
class FFmpegConvertResponse(web.StreamResponse):
    """HTTP streaming response that uses ffmpeg to convert audio from a URL."""
 
    def __init__(
        self,
        manager: FFmpegManager,
        convert_info: FFmpegConversionInfo,
        device_id: str,
        proxy_data: FFmpegProxyData,
        chunk_size: int = 2048,
    ) -> None:
        """Initialize response.
 
        Parameters
        ----------
        manager: FFmpegManager
            ffmpeg manager
        convert_info: FFmpegConversionInfo
            Information necessary to do the conversion
        device_id: str
            ESPHome device id
        proxy_data: FFmpegProxyData
            Data object to store ffmpeg process
        chunk_size: int
            Number of bytes to read from ffmpeg process at a time
 
        """
        super().__init__(status=200)
        self.hasshass = manager.hass
        self.managermanager = manager
        self.convert_infoconvert_info = convert_info
        self.device_iddevice_id = device_id
        self.proxy_dataproxy_data = proxy_data
        self.chunk_sizechunk_size = chunk_size
 
    async def transcode(
        self, request: BaseRequest, writer: AbstractStreamWriter
    ) -> None:
        """Stream url through ffmpeg conversion and out to HTTP client."""
        command_args = [
            "-i",
            self.convert_infoconvert_info.media_url,
            "-f",
            self.convert_infoconvert_info.media_format,
        ]
 
        if self.convert_infoconvert_info.rate is not None:
            # Sample rate
            command_args.extend(["-ar", str(self.convert_infoconvert_info.rate)])
 
        if self.convert_infoconvert_info.channels is not None:
            # Number of channels
            command_args.extend(["-ac", str(self.convert_infoconvert_info.channels)])
 
        if self.convert_infoconvert_info.width == 2:
            # 16-bit samples
            command_args.extend(["-sample_fmt", "s16"])
 
        # Remove metadata and cover art
        command_args.extend(["-map_metadata", "-1", "-vn"])
 
        # disable progress stats on stderr
        command_args.append("-nostats")
 
        # Output to stdout
        command_args.append("pipe:")
 
        _LOGGER.debug("%s %s", self.managermanager.binary, " ".join(command_args))
        proc = await asyncio.create_subprocess_exec(
            self.managermanager.binary,
            *command_args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            close_fds=False,  # use posix_spawn in CPython < 3.13
        )
 
        # Only one conversion process per device is allowed
        self.convert_infoconvert_info.proc = proc
 
        # Create background task which will be cancelled when home assistant shuts down
        write_task = self.hasshass.async_create_background_task(
            self._write_ffmpeg_data_write_ffmpeg_data(request, writer, proc), "ESPHome media proxy"
        )
        await write_task
 
    async def _write_ffmpeg_data(
        self,
        request: BaseRequest,
        writer: AbstractStreamWriter,
        proc: asyncio.subprocess.Process,
    ) -> None:
        assert proc.stdout is not None
        assert proc.stderr is not None
 
        stderr_task = self.hasshass.async_create_background_task(
            self._dump_ffmpeg_stderr_dump_ffmpeg_stderr(proc), "ESPHome media proxy dump stderr"
        )
 
        try:
            # Pull audio chunks from ffmpeg and pass them to the HTTP client
            while (
                self.hasshass.is_running
                and (request.transport is not None)
                and (not request.transport.is_closing())
                and (chunk := await proc.stdout.read(self.chunk_sizechunk_size))
            ):
                await self.write(chunk)
        except asyncio.CancelledError:
            _LOGGER.debug("ffmpeg transcoding cancelled")
            # Abort the transport, we don't wait for ESPHome to drain the write buffer;
            # it may need a very long time or never finish if the player is paused.
            if request.transport:
                request.transport.abort()
            raise  # don't log error
        except:
            _LOGGER.exception("Unexpected error during ffmpeg conversion")
            raise
        finally:
            # Allow conversion info to be removed
            self.convert_infoconvert_info.is_finished = True
 
            # stop dumping ffmpeg stderr task
            stderr_task.cancel()
 
            # Terminate hangs, so kill is used
            if proc.returncode is None:
                proc.kill()
 
            # Close connection by writing EOF unless already closing
            if request.transport and not request.transport.is_closing():
                await writer.write_eof()
 
    async def _dump_ffmpeg_stderr(
        self,
        proc: asyncio.subprocess.Process,
    ) -> None:
        assert proc.stdout is not None
        assert proc.stderr is not None
 
        while self.hasshass.is_running and (chunk := await proc.stderr.readline()):
            _LOGGER.debug("ffmpeg[%s] output: %s", proc.pid, chunk.decode().rstrip())
 
 
class FFmpegProxyView(HomeAssistantView):
    """FFmpeg web view to convert audio and stream back to client."""
 
    requires_auth = False
    url = "/api/esphome/ffmpeg_proxy/{device_id}/{filename}"
    name = "api:esphome:ffmpeg_proxy"
 
    def __init__(self, manager: FFmpegManager, proxy_data: FFmpegProxyData) -> None:
        """Initialize an ffmpeg view."""
        self.managermanager = manager
        self.proxy_dataproxy_data = proxy_data
 
    async def get(
        self, request: web.Request, device_id: str, filename: str
    ) -> web.StreamResponse:
        """Start a get request."""
        device_conversions = self.proxy_dataproxy_data.conversions[device_id]
        if not device_conversions:
            return web.Response(
                body="No proxy URL for device", status=HTTPStatus.NOT_FOUND
            )
 
        # {id}.mp3 -> id, mp3
        convert_id, media_format = filename.rsplit(".")
 
        # Look up conversion info
        convert_info: FFmpegConversionInfo | None = None
        for maybe_convert_info in device_conversions:
            if (maybe_convert_info.convert_id == convert_id) and (
                maybe_convert_info.media_format == media_format
            ):
                convert_info = maybe_convert_info
                break
 
        if convert_info is None:
            return web.Response(body="Invalid proxy URL", status=HTTPStatus.BAD_REQUEST)
 
        # Stop previous process if the URL is being reused.
        # We could continue from where the previous connection left off, but
        # there would be no media header.
        if (convert_info.proc is not None) and (convert_info.proc.returncode is None):
            convert_info.proc.kill()
            convert_info.proc = None
 
        # Stream converted audio back to client
        resp = FFmpegConvertResponse(
            self.managermanager, convert_info, device_id, self.proxy_dataproxy_data
        )
        writer = await resp.prepare(request)
        assert writer is not None
        await resp.transcode(request, writer)
        return resp