A few small bugfixes and enhancements to playback and enqueuing (#1670)

music_assistant/common/helpers/util.py

@@ -26,6 +26,33 @@
 multi_space_pattern = re.compile(r"\s{2,}")
 end_junk_pattern = re.compile(r"(.+?)(\s\W+)$")
 
+VERSION_PARTS = (
+    # list of common version strings
+    "version",
+    "live",
+    "edit",
+    "remix",
+    "mix",
+    "acoustic",
+    "instrumental",
+    "karaoke",
+    "remaster",
+    "versie",
+    "unplugged",
+    "disco",
+    "akoestisch",
+    "deluxe",
+)
+IGNORE_TITLE_PARTS = (
+    # strings that may be stripped off a title part
+    # (most important the featuring parts)
+    "feat.",
+    "featuring",
+    "ft.",
+    "with ",
+    "explicit",
+)
+
 
 def filename_from_string(string: str) -> str:
     """Create filename from unsafe string."""
@@ -79,81 +106,30 @@ def create_sort_name(input_str: str) -> str:
 
 
 def parse_title_and_version(title: str, track_version: str | None = None) -> tuple[str, str]:
-    """Try to parse clean track title and version from the title."""
-    version = ""
-    for splitter in [" (", " [", " - ", " (", " [", "-"]:
-        if splitter in title:
-            title_parts = title.split(splitter)
-            for title_part in title_parts:
-                # look for the end splitter
-                for end_splitter in [")", "]"]:
-                    if end_splitter in title_part:
-                        title_part = title_part.split(end_splitter)[0]  # noqa: PLW2901
-                for version_str in [
-                    "version",
-                    "live",
-                    "edit",
-                    "remix",
-                    "mix",
-                    "acoustic",
-                    "instrumental",
-                    "karaoke",
-                    "remaster",
-                    "versie",
-                    "radio",
-                    "unplugged",
-                    "disco",
-                    "akoestisch",
-                    "deluxe",
-                ]:
-                    if version_str in title_part.lower():
-                        version = title_part
-                        title = title.split(splitter + version)[0]
-    title = clean_title(title)
-    if not version and track_version:
-        version = track_version
-    version = get_version_substitute(version).title()
-    if version == title:
-        version = ""
+    """Try to parse version from the title."""
+    version = track_version or ""
+    for regex in (r"\(.*?\)", r"\[.*?\]", r" - .*"):
+        for title_part in re.findall(regex, title):
+            for ignore_str in IGNORE_TITLE_PARTS:
+                if ignore_str in title_part.lower():
+                    title = title.replace(title_part, "").strip()
+                    continue
+            for version_str in VERSION_PARTS:
+                if version_str not in title_part.lower():
+                    continue
+                version = (
+                    title_part.replace("(", "")
+                    .replace(")", "")
+                    .replace("[", "")
+                    .replace("]", "")
+                    .replace("-", "")
+                    .strip()
+                )
+                title = title.replace(title_part, "").strip()
+                return (title, version)
     return title, version
 
 
-def clean_title(title: str) -> str:
-    """Strip unwanted additional text from title."""
-    for splitter in [" (", " [", " - ", " (", " [", "-"]:
-        if splitter in title:
-            title_parts = title.split(splitter)
-            for title_part in title_parts:
-                # look for the end splitter
-                for end_splitter in [")", "]"]:
-                    if end_splitter in title_part:
-                        title_part = title_part.split(end_splitter)[0]  # noqa: PLW2901
-                for ignore_str in ["feat.", "featuring", "ft.", "with ", "explicit"]:
-                    if ignore_str in title_part.lower():
-                        return title.split(splitter + title_part)[0].strip()
-    return title.strip()
-
-
-def get_version_substitute(version_str: str) -> str:
-    """Transform provider version str to universal version type."""
-    version_str = version_str.lower()
-    # substitute edit and edition with version
-    if "edition" in version_str or "edit" in version_str:
-        version_str = version_str.replace(" edition", " version")
-        version_str = version_str.replace(" edit ", " version")
-    if version_str.startswith("the "):
-        version_str = version_str.split("the ")[1]
-    if "radio mix" in version_str:
-        version_str = "radio version"
-    elif "video mix" in version_str:
-        version_str = "video version"
-    elif "spanglish" in version_str or "spanish" in version_str:
-        version_str = "spanish version"
-    elif "remaster" in version_str:
-        version_str = "remaster"
-    return version_str.strip()
-
-
 def strip_ads(line: str) -> str:
     """Strip Ads from line."""
     if ad_pattern.search(line):

music_assistant/server/controllers/player_queues.py

@@ -1075,7 +1075,7 @@ async def preload_next_item(
             msg = f"PlayerQueue {queue_id} is not available"
             raise PlayerUnavailableError(msg)
         if current_item_id_or_index is None:
-            cur_index = queue.index_in_buffer
+            cur_index = queue.index_in_buffer or queue.current_index or 0
         elif isinstance(current_item_id_or_index, str):
             cur_index = self.index_by_id(queue_id, current_item_id_or_index)
         else:

music_assistant/server/helpers/process.py

@@ -202,7 +202,7 @@ async def iter_stderr(self) -> AsyncGenerator[str, None]:
             line = await self.read_stderr()
             if line == b"":
                 break
-            line = line.decode().strip()
+            line = line.decode("utf-8", errors="ignore").strip()
             if not line:
                 continue
             yield line

music_assistant/server/providers/chromecast/__init__.py

@@ -445,6 +445,7 @@ def on_new_media_status(self, castplayer: CastPlayer, status: MediaStatus) -> No
             status.player_state,
         )
         # handle castplayer playing from a group
+        group_player: CastPlayer | None = None
         if castplayer.active_group is not None:
             if not (group_player := self.castplayers.get(castplayer.active_group)):
                 return
@@ -471,7 +472,9 @@ def on_new_media_status(self, castplayer: CastPlayer, status: MediaStatus) -> No
             castplayer.player.elapsed_time = status.current_time
 
         # active source
-        if castplayer.cc.app_id == MASS_APP_ID:
+        if group_player:
+            castplayer.player.active_source = group_player.player.active_source
+        elif castplayer.cc.app_id == MASS_APP_ID:
             castplayer.player.active_source = castplayer.player_id
         else:
             castplayer.player.active_source = castplayer.cc.app_display_name

music_assistant/server/providers/dlna/__init__.py

@@ -28,6 +28,7 @@
     CONF_ENTRY_CROSSFADE_FLOW_MODE_REQUIRED,
     CONF_ENTRY_ENABLE_ICY_METADATA,
     CONF_ENTRY_ENFORCE_MP3,
+    CONF_ENTRY_FLOW_MODE_DEFAULT_ENABLED,
     CONF_ENTRY_HTTP_PROFILE,
     ConfigEntry,
     ConfigValueType,
@@ -71,6 +72,9 @@
     CONF_ENTRY_ENFORCE_MP3,
     CONF_ENTRY_HTTP_PROFILE,
     CONF_ENTRY_ENABLE_ICY_METADATA,
+    # enable flow mode by default because
+    # most dlna players do not support enqueueing
+    CONF_ENTRY_FLOW_MODE_DEFAULT_ENABLED,
     create_sample_rates_config_entry(192000, 24, 96000, 24),
 )
 

music_assistant/server/providers/hass_players/__init__.py

@@ -31,6 +31,7 @@
 )
 from music_assistant.common.models.errors import SetupFailedError
 from music_assistant.common.models.player import DeviceInfo, Player, PlayerMedia
+from music_assistant.constants import CONF_FLOW_MODE
 from music_assistant.server.models.player_provider import PlayerProvider
 from music_assistant.server.providers.hass import DOMAIN as HASS_DOMAIN
 
@@ -410,6 +411,14 @@ async def _setup_player(
             supported_features=tuple(supported_features),
             state=StateMap.get(state["state"], PlayerState.IDLE),
         )
+        # bugfix: correct flow-mode setting for players that do not support media_enque
+        # remove this after MA release 2.5+
+        if MediaPlayerEntityFeature.MEDIA_ENQUEUE not in hass_supported_features:
+            self.mass.config.set_raw_player_config_value(
+                player.player_id,
+                CONF_FLOW_MODE,
+                True,
+            )
         if MediaPlayerEntityFeature.GROUPING in hass_supported_features:
             player.can_sync_with = platform_players
         self._update_player_attributes(player, state["attributes"])

music_assistant/server/providers/slimproto/__init__.py

@@ -280,6 +280,7 @@ async def get_player_config_entries(self, player_id: str) -> tuple[ConfigEntry]:
                 *base_entries,
                 CONF_ENTRY_CROSSFADE,
                 CONF_ENTRY_CROSSFADE_DURATION,
+                CONF_ENTRY_HTTP_PROFILE_FORCED_2,
                 create_sample_rates_config_entry(96000, 24, 48000, 24),
             )
 

tests/test_helpers.py

@@ -18,18 +18,26 @@ def test_version_extract() -> None:
     title, version = util.parse_title_and_version(test_str)
     assert title == "Bam Bam"
     assert version == "Karaoke Version"
+    test_str = "Bam Bam (feat. Ed Sheeran) [Karaoke Version]"
+    title, version = util.parse_title_and_version(test_str)
+    assert title == "Bam Bam"
+    assert version == "Karaoke Version"
     test_str = "SuperSong (2011 Remaster)"
     title, version = util.parse_title_and_version(test_str)
     assert title == "SuperSong"
-    assert version == "Remaster"
+    assert version == "2011 Remaster"
     test_str = "SuperSong (Live at Wembley)"
     title, version = util.parse_title_and_version(test_str)
     assert title == "SuperSong"
-    assert version == "Live At Wembley"
+    assert version == "Live at Wembley"
     test_str = "SuperSong (Instrumental)"
     title, version = util.parse_title_and_version(test_str)
     assert title == "SuperSong"
     assert version == "Instrumental"
+    test_str = "SuperSong (Explicit)"
+    title, version = util.parse_title_and_version(test_str)
+    assert title == "SuperSong"
+    assert version == ""
 
 
 async def test_uri_parsing() -> None: