Michael Hu
Create domain service interfaces
55e29e2
"""AudioChunk value object for streaming audio data."""
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class AudioChunk:
"""Value object representing a chunk of audio data for streaming."""
data: bytes
format: str
sample_rate: int
chunk_index: int
is_final: bool = False
timestamp: Optional[float] = None
def __post_init__(self):
"""Validate audio chunk after initialization."""
self._validate()
def _validate(self):
"""Validate audio chunk properties."""
if not isinstance(self.data, bytes):
raise TypeError("Audio data must be bytes")
if not self.data:
raise ValueError("Audio data cannot be empty")
if self.format not in ['wav', 'mp3', 'flac', 'ogg', 'raw']:
raise ValueError(f"Unsupported audio format: {self.format}")
if not isinstance(self.sample_rate, int) or self.sample_rate <= 0:
raise ValueError("Sample rate must be a positive integer")
if not isinstance(self.chunk_index, int) or self.chunk_index < 0:
raise ValueError("Chunk index must be a non-negative integer")
if not isinstance(self.is_final, bool):
raise TypeError("is_final must be a boolean")
if self.timestamp is not None:
if not isinstance(self.timestamp, (int, float)) or self.timestamp < 0:
raise ValueError("Timestamp must be a non-negative number")
@property
def size_bytes(self) -> int:
"""Get the size of audio chunk data in bytes."""
return len(self.data)
@property
def duration_estimate(self) -> float:
"""Estimate duration in seconds based on data size and sample rate."""
# Rough estimation assuming 16-bit audio (2 bytes per sample)
bytes_per_second = self.sample_rate * 2
return len(self.data) / bytes_per_second if bytes_per_second > 0 else 0.0