feat: implement llm.Audio

Author: teamdandelion

python/mirascope/llm/content/audio.py

@@ -1,7 +1,11 @@
 """The `Audio` content class."""
 
+import base64
 from dataclasses import dataclass
-from typing import Literal
+from pathlib import Path
+from typing import Literal, get_args
+
+import httpx
 
 AudioMimeType = Literal[
     "audio/wav",
@@ -12,6 +16,35 @@
     "audio/flac",
 ]
 
+MIME_TYPES = get_args(AudioMimeType)
+
+# Maximum audio size in bytes (25MB)
+MAX_AUDIO_SIZE = 25 * 1024 * 1024
+
+
+def infer_audio_type(audio_data: bytes) -> AudioMimeType:
+    """Get the MIME type of an audio file from its raw bytes.
+
+    Raises:
+        ValueError: If the audio type cannot be determined or data is too small
+    """
+    if len(audio_data) < 12:
+        raise ValueError("Audio data too small to determine type (minimum 12 bytes)")
+
+    if audio_data.startswith(b"RIFF") and audio_data[8:12] == b"WAVE":
+        return "audio/wav"
+    elif audio_data.startswith(b"ID3") or audio_data.startswith(b"\xff\xfb"):
+        return "audio/mp3"
+    elif audio_data.startswith(b"FORM") and audio_data[8:12] == b"AIFF":
+        return "audio/aiff"
+    elif audio_data.startswith(b"\xff\xf1") or audio_data.startswith(b"\xff\xf9"):
+        return "audio/aac"
+    elif audio_data.startswith(b"OggS"):
+        return "audio/ogg"
+    elif audio_data.startswith(b"fLaC"):
+        return "audio/flac"
+    raise ValueError("Unsupported audio type")
+
 
 @dataclass(kw_only=True)
 class Base64AudioSource:
@@ -38,29 +71,133 @@ class Audio:
     source: Base64AudioSource
 
     @classmethod
-    def from_url(
-        cls,
-        url: str,
-    ) -> "Audio":
-        """Create an `Audio` from a URL."""
-        raise NotImplementedError
+    def download(cls, url: str, *, max_size: int = MAX_AUDIO_SIZE) -> "Audio":
+        """Download and encode an audio file from a URL.
+
+        Args:
+            url: The URL of the audio file to download
+            max_size: Maximum allowed audio size in bytes (default: 25MB)
+
+        Returns:
+            An `Audio` with a `Base64AudioSource`
+
+        Raises:
+            ValueError: If the downloaded audio exceeds max_size
+        """
+        response = httpx.get(url, follow_redirects=True)
+        response.raise_for_status()
+
+        content_length = len(response.content)
+        if content_length > max_size:
+            raise ValueError(
+                f"Audio size ({content_length} bytes) exceeds maximum allowed size ({max_size} bytes)"
+            )
+
+        mime_type = infer_audio_type(response.content)
+        data = base64.b64encode(response.content).decode("utf-8")
+
+        return cls(
+            source=Base64AudioSource(
+                type="base64_audio_source",
+                data=data,
+                mime_type=mime_type,
+            )
+        )
 
     @classmethod
-    def from_file(
-        cls,
-        file_path: str,
-        *,
-        mime_type: AudioMimeType | None,
+    async def download_async(
+        cls, url: str, *, max_size: int = MAX_AUDIO_SIZE
     ) -> "Audio":
-        """Create an `Audio` from a file path."""
-        raise NotImplementedError
+        """Asynchronously download and encode an audio file from a URL.
+
+        Args:
+            url: The URL of the audio file to download
+            max_size: Maximum allowed audio size in bytes (default: 25MB)
+
+        Returns:
+            An `Audio` with a `Base64AudioSource`
+
+        Raises:
+            ValueError: If the downloaded audio exceeds max_size
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, follow_redirects=True)
+            response.raise_for_status()
+
+            content_length = len(response.content)
+            if content_length > max_size:
+                raise ValueError(
+                    f"Audio size ({content_length} bytes) exceeds maximum allowed size ({max_size} bytes)"
+                )
+
+            mime_type = infer_audio_type(response.content)
+            data = base64.b64encode(response.content).decode("utf-8")
+
+            return cls(
+                source=Base64AudioSource(
+                    type="base64_audio_source",
+                    data=data,
+                    mime_type=mime_type,
+                )
+            )
 
     @classmethod
-    def from_bytes(
-        cls,
-        data: bytes,
-        *,
-        mime_type: AudioMimeType | None,
-    ) -> "Audio":
-        """Create an `Audio` from raw bytes."""
-        raise NotImplementedError
+    def from_file(cls, file_path: str, *, max_size: int = MAX_AUDIO_SIZE) -> "Audio":
+        """Create an `Audio` from a file path.
+
+        Args:
+            file_path: Path to the audio file
+            max_size: Maximum allowed audio size in bytes (default: 25MB)
+
+        Raises:
+            FileNotFoundError: If the file does not exist
+            ValueError: If the file size exceeds max_size
+        """
+        path = Path(file_path)
+        file_size = path.stat().st_size
+        if file_size > max_size:
+            raise ValueError(
+                f"Audio file size ({file_size} bytes) exceeds maximum allowed size ({max_size} bytes)"
+            )
+
+        with open(path, "rb") as f:
+            audio_bytes = f.read()
+
+        mime_type = infer_audio_type(audio_bytes)
+        data = base64.b64encode(audio_bytes).decode("utf-8")
+
+        return cls(
+            source=Base64AudioSource(
+                type="base64_audio_source",
+                data=data,
+                mime_type=mime_type,
+            )
+        )
+
+    @classmethod
+    def from_bytes(cls, data: bytes, *, max_size: int = MAX_AUDIO_SIZE) -> "Audio":
+        """Create an `Audio` from raw bytes.
+
+        Args:
+            data: Raw audio bytes
+            max_size: Maximum allowed audio size in bytes (default: 25MB)
+
+        Raises:
+            ValueError: If the data size exceeds max_size
+        """
+        data_size = len(data)
+        if data_size > max_size:
+            raise ValueError(
+                f"Audio data size ({data_size} bytes) exceeds maximum allowed size ({max_size} bytes)"
+            )
+
+        mime_type = infer_audio_type(data)
+        encoded_data = base64.b64encode(data).decode("utf-8")
+
+        return cls(
+            source=Base64AudioSource(
+                type="base64_audio_source",
+                data=encoded_data,
+                mime_type=mime_type,
+            )
+        )