Add plex connect provider

music_assistant/providers/plex/__init__.py

@@ -936,6 +936,15 @@ async def get_stream_details(self, item_id: str, media_type: MediaType) -> Strea
         media_part: PlexMediaPart = media.parts[0]
         audio_stream: PlexAudioStream = media_part.audioStreams()[0]
 
+        plex_data = {
+            "rating_key": plex_track.ratingKey,
+            "key": media_part.key,
+            "duration": plex_track.duration,
+            "server_url": self._baseurl,
+            "token": self.config.get_value(CONF_AUTH_TOKEN),
+            "machine_identifier": self._plex_server.machineIdentifier,
+        }
+
         stream_details = StreamDetails(
             item_id=plex_track.key,
             provider=self.lookup_key,
@@ -945,7 +954,7 @@ async def get_stream_details(self, item_id: str, media_type: MediaType) -> Strea
             ),
             stream_type=StreamType.HTTP,
             duration=plex_track.duration,
-            data=plex_track,
+            data=plex_data,
             can_seek=True,
             allow_seek=True,
         )
@@ -977,7 +986,7 @@ async def on_streamed(
         def mark_played() -> None:
             item = streamdetails.data
             params = {
-                "key": str(item.ratingKey),
+                "key": str(item["rating_key"]),
                 "identifier": "com.plexapp.plugins.library",
             }
             self._plex_server.query("/:/scrobble", params=params)

music_assistant/providers/plex_connect/__init__.py

@@ -0,0 +1,79 @@
+"""
+Plugin provider for Music Assistant to report Plex playback timeline events.
+
+This provider integrates with Plex to monitor and report playback timeline updates
+back to Music Assistant, enabling enhanced synchronization and playback tracking.
+It does not provide audio sources, but acts as a plugin to relay timeline events.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
+    from music_assistant_models.provider import ProviderManifest
+
+    from music_assistant.mass import MusicAssistant
+    from music_assistant.models import ProviderInstanceType
+
+from music_assistant.models.plugin import PluginProvider
+
+from .timeline import PlexTimelineReporter
+
+
+async def setup(
+    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
+) -> ProviderInstanceType:
+    """Initialize provider(instance) with given configuration."""
+    return PlexConnect(mass, manifest, config)
+
+
+async def get_config_entries(
+    mass: MusicAssistant,
+    instance_id: str | None = None,
+    action: str | None = None,
+    values: dict[str, ConfigValueType] | None = None,
+) -> tuple[ConfigEntry, ...]:
+    """
+    Return Config entries to setup this provider.
+
+    instance_id: id of an existing provider instance (None if new instance setup).
+    action: [optional] action key called from config entries UI.
+    values: the (intermediate) raw values for config entries sent with the action.
+    """
+    # ruff: noqa: ARG001
+    # Config Entries are used to configure the Provider if needed.
+    # See the models of ConfigEntry and ConfigValueType for more information what is supported.
+    # The ConfigEntry is a dataclass that represents a single configuration entry.
+    # The ConfigValueType is an Enum that represents the type of value that
+    # can be stored in a ConfigEntry.
+    # If your provider does not need any configuration, you can return an empty tuple.
+    return ()
+
+
+class PlexConnect(PluginProvider):
+    """Plugin provider to report Plex playback timeline."""
+
+    def __init__(
+        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
+    ) -> None:
+        """Initialize PlexConnect plugin provider."""
+        super().__init__(mass, manifest, config)
+        self._reporter: PlexTimelineReporter | None = None
+
+    @property
+    def supported_features(self) -> set:
+        """No audio source; plugin only reports timeline."""
+        return set()
+
+    async def loaded_in_mass(self) -> None:
+        """Set up the Plex timeline reporter after provider is loaded."""
+        self._reporter = PlexTimelineReporter(self.mass)
+        await self._reporter.setup()
+
+    async def unload(self, is_removed: bool = False) -> None:
+        """Clean up reporter on unload."""
+        if self._reporter:
+            await self._reporter.close()
+        await super().unload(is_removed)

music_assistant/providers/plex_connect/manifest.json

@@ -0,0 +1,10 @@
+{
+  "type": "plugin",
+  "domain": "plex_connect",
+  "name": "Plex Connect",
+  "description": "Add Plex Connetion support to ANY Music Assistant player.",
+  "codeowners": ["chicco-carone"],
+  "requirements": ["plexapi==4.16.1"],
+  "documentation": "https://music-assistant.io/music-providers/plex/",
+  "multi_instance": false
+}