sample_size = int(sample_rate * (bit_depth / 8) * channels) # 1 second
buffer_size = sample_size * (queue.settings.crossfade_duration or 2)
# force small buffer for radio to prevent too much lag at start
- if queue_track.media_type == MediaType.RADIO:
+ if queue_track.media_type != MediaType.TRACK:
use_crossfade = False
buffer_size = sample_size
PCM_S32LE = "s32le" # PCM signed 32-bit little-endian
PCM_F32LE = "f32le" # PCM 32-bit floating-point little-endian
PCM_F64LE = "f64le" # PCM 64-bit floating-point little-endian
+ UNKNOWN = "?"
@classmethod
- def try_parse(
- cls: "ContentType", string: str, fallback: str = "mp3"
- ) -> "ContentType":
+ def try_parse(cls: "ContentType", string: str) -> "ContentType":
"""Try to parse ContentType from (url)string."""
tempstr = string.lower()
if "." in tempstr:
try:
return cls(tempstr)
except ValueError:
- return cls(fallback)
+ return cls.UNKNOWN
def is_pcm(self):
"""Return if contentype is PCM."""
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
from music_assistant.helpers.audio import get_stream_details
-from music_assistant.models.enums import (
- EventType,
- MediaType,
- QueueOption,
- RepeatMode,
-)
+from music_assistant.models.enums import EventType, MediaType, QueueOption, RepeatMode
from music_assistant.models.errors import (
MediaNotFoundError,
MusicAssistantError,
# invalid MA uri or item not found error
if uri.startswith("http"):
# a plain url was provided
- queue_items.append(QueueItem(uri))
+ queue_items.append(QueueItem.from_url(uri))
continue
raise MediaNotFoundError(f"Invalid uri: {uri}") from err
d.pop("duration")
return d
+ @classmethod
+ def from_url(cls, url: str) -> QueueItem:
+ """Create QueueItem from plain url."""
+ return cls(
+ uri=url,
+ name=url,
+ duration=None,
+ media_type=MediaType.UNKNOWN,
+ image=None,
+ media_item=None,
+ )
+
@classmethod
def from_media_item(cls, media_item: Track | Radio):
"""Construct QueueItem from track/radio item."""