import contextlib
from collections.abc import AsyncGenerator, Callable
from functools import wraps
-from typing import Any, Final, ParamSpec
-
-from music_assistant_models.streamdetails import AudioFormat
+from typing import TYPE_CHECKING, Any, Final, ParamSpec, cast
from music_assistant.helpers.util import close_async_generator
+if TYPE_CHECKING:
+ from music_assistant_models.media_items.audio_format import AudioFormat
+
# Type variables for the buffered decorator
_P = ParamSpec("_P")
_ACTIVE_PRODUCER_TASKS: set[asyncio.Task[Any]] = set()
-async def buffered_audio(
+async def buffered_audio( # noqa: PLR0915
generator: AsyncGenerator[bytes, None],
pcm_format: AudioFormat,
buffer_size: int = DEFAULT_BUFFER_SIZE,
await condition.wait()
if cancelled:
- break
+ break # type: ignore[unreachable]
# Append to shared buffer
data_buffer.extend(chunk)
@wraps(func)
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> AsyncGenerator[bytes, None]:
# Extract pcm_format from function arguments
- pcm_format = kwargs.get(pcm_format_arg)
+ pcm_format = cast("AudioFormat | None", kwargs.get(pcm_format_arg))
if pcm_format is None:
msg = f"Audio buffer decorator requires '{pcm_format_arg}' argument"
raise ValueError(msg)
from typing import TYPE_CHECKING
-from music_assistant_models.config_entries import ProviderConfig
+from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
from music_assistant_models.enums import ConfigEntryType, ProviderFeature
from music_assistant_models.provider import ProviderManifest
from .provider import AirPlayProvider
if TYPE_CHECKING:
- from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
+ from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
from music_assistant_models.provider import ProviderManifest
from music_assistant.models import ProviderInstanceType