Spaces:
Sleeping
Sleeping
| """AudioChunk value object for streaming audio data.""" | |
| from dataclasses import dataclass | |
| from typing import Optional | |
| 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") | |
| def size_bytes(self) -> int: | |
| """Get the size of audio chunk data in bytes.""" | |
| return len(self.data) | |
| 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 |