SaraTune/Smtc.py

"""Windows: SaraTune in the System Media Transport Controls (the media
overlay by the volume flyout, media keys, the lock screen).

The controls normally belong to a window, but the window here is the Flet
client's, a separate process. The usual way out for apps like that: make a
WinRT MediaPlayer, switch off its own command handling, and drive its
SystemMediaTransportControls by hand. That needs no window handle.

WinRT raises its events on its own threads; they're passed to Flet's loop
with call_soon_threadsafe. Written against the winrt 3.2 type stubs and only
tested as far as importing goes (no real Windows to run it on)."""

import asyncio
import datetime

from winrt.windows.media import (
    MediaPlaybackAutoRepeatMode, MediaPlaybackStatus, MediaPlaybackType, SystemMediaTransportControlsButton,
    SystemMediaTransportControlsTimelineProperties,
)
from winrt.windows.media.playback import MediaPlayer
from winrt.windows.storage import StorageFile
from winrt.windows.storage.streams import RandomAccessStreamReference

from MediaControls import SeekStep

RepeatModes = {  # SaraTune <-> Windows
    "Off": MediaPlaybackAutoRepeatMode.NONE,
    "All": MediaPlaybackAutoRepeatMode.LIST,
    "One": MediaPlaybackAutoRepeatMode.TRACK,
}
TimelineEverySeconds = 5  # Windows doesn't move the position along by itself


def _Time(Milliseconds: int) -> datetime.timedelta:
    return datetime.timedelta(milliseconds=max(0, int(Milliseconds)))


class SmtcBackend:
    def __init__(self, Remote):
        self.Remote = Remote
        self.Loop: asyncio.AbstractEventLoop | None = None
        self.MediaPlayer: MediaPlayer | None = None
        self.Controls = None
        self._Tokens: list[tuple[str, object]] = []
        self._Ticker: asyncio.Task | None = None

    async def Start(self) -> None:
        self.Loop = asyncio.get_running_loop()
        self.MediaPlayer = MediaPlayer()
        self.MediaPlayer.command_manager.is_enabled = False  # SaraTune answers the buttons, not this player
        Controls = self.Controls = self.MediaPlayer.system_media_transport_controls
        Controls.is_enabled = True
        for Name in ("play", "pause", "stop", "next", "previous", "fast_forward", "rewind"):
            setattr(Controls, f"is_{Name}_enabled", True)
        self._Tokens = [
            ("button_pressed", Controls.add_button_pressed(
                lambda Sender, Arguments: self._OnLoop(self._Button, Arguments.button))),
            ("playback_position_change_requested", Controls.add_playback_position_change_requested(
                lambda Sender, Arguments: self._OnLoop(self._SeekTo, Arguments.requested_playback_position))),
            ("shuffle_enabled_change_requested", Controls.add_shuffle_enabled_change_requested(
                lambda Sender, Arguments: self._OnLoop(self.Remote.SetShuffle, Arguments.requested_shuffle_enabled))),
            ("auto_repeat_mode_change_requested", Controls.add_auto_repeat_mode_change_requested(
                lambda Sender, Arguments: self._OnLoop(self._Repeat, Arguments.requested_auto_repeat_mode))),
        ]
        self._Ticker = self.Loop.create_task(self._Tick())

    async def Stop(self) -> None:
        if self._Ticker:
            self._Ticker.cancel()
        if self.Controls is not None:
            for Name, Token in self._Tokens:
                getattr(self.Controls, f"remove_{Name}")(Token)
            self.Controls.display_updater.clear_all()
            self.Controls.is_enabled = False
        if self.MediaPlayer is not None:
            self.MediaPlayer.close()
        self.Controls = self.MediaPlayer = None

    # --- From Windows -------------------------------------------------------

    def _OnLoop(self, Function, *Arguments) -> None:
        """WinRT thread -> Flet's loop."""
        if self.Loop is not None:
            self.Loop.call_soon_threadsafe(Function, *Arguments)

    def _Button(self, Button) -> None:
        R = self.Remote
        Actions = {
            SystemMediaTransportControlsButton.PLAY: (R.Play,),
            SystemMediaTransportControlsButton.PAUSE: (R.Pause,),
            SystemMediaTransportControlsButton.STOP: (R.StopPlaying,),
            SystemMediaTransportControlsButton.NEXT: (R.Next,),
            SystemMediaTransportControlsButton.PREVIOUS: (R.Previous,),
            SystemMediaTransportControlsButton.FAST_FORWARD: (R.SeekBy, SeekStep),
            SystemMediaTransportControlsButton.REWIND: (R.SeekBy, -SeekStep),
        }
        Action = Actions.get(Button)
        if Action:
            R.Run(*Action)

    def _SeekTo(self, Position: datetime.timedelta) -> None:
        self.Remote.Run(self.Remote.SetPosition, int(Position.total_seconds() * 1000))

    def _Repeat(self, Mode) -> None:
        Ours = next((Name for Name, Value in RepeatModes.items() if Value == Mode), "Off")
        self.Remote.SetRepeat(Ours)

    # --- To Windows ---------------------------------------------------------

    def StateChanged(self) -> None:
        if self.Controls is None:
            return
        R = self.Remote
        C = self.Controls
        if not R.Info:
            C.playback_status = MediaPlaybackStatus.CLOSED
        else:
            C.playback_status = MediaPlaybackStatus.PLAYING if R.ThePlayer.Playing else MediaPlaybackStatus.PAUSED
        C.shuffle_enabled = R.Shuffle
        C.auto_repeat_mode = RepeatModes[R.Repeat]
        self._Timeline()

    def TrackChanged(self) -> None:
        if self.Controls is None or self.Loop is None:
            return
        self.Loop.create_task(self._Display())
        self.StateChanged()

    async def _Display(self) -> None:
        Updater = self.Controls.display_updater
        Info = self.Remote.Info
        if Info is None:
            Updater.clear_all()
            Updater.update()
            return
        Updater.clear_all()  # the previous track's cover too
        Updater.type = MediaPlaybackType.MUSIC
        Music = Updater.music_properties
        Music.title = Info.DisplayTitle
        Music.artist = Info.Artist or Info.AlbumArtist or ""
        Music.album_title = Info.Album or ""
        Music.album_artist = Info.AlbumArtist or ""
        Music.track_number = Info.TrackNumber or 0
        Art = self.Remote.ArtPath
        if Art:
            try:
                File = await StorageFile.get_file_from_path_async(str(Art))
                Updater.thumbnail = RandomAccessStreamReference.create_from_file(File)
            except Exception:
                pass  # no cover rather than no controls
        Updater.update()

    def _Timeline(self) -> None:
        R = self.Remote
        Timeline = SystemMediaTransportControlsTimelineProperties()
        Length = R.ThePlayer.Duration or (R.Info.Duration if R.Info else 0)
        Timeline.start_time = _Time(0)
        Timeline.min_seek_time = _Time(0)
        Timeline.end_time = _Time(Length)
        Timeline.max_seek_time = _Time(Length)
        Timeline.position = _Time(R.ThePlayer.Position)
        self.Controls.update_timeline_properties(Timeline)

    def Seeked(self) -> None:
        if self.Controls is not None:
            self._Timeline()

    def VolumeChanged(self) -> None:
        pass  # the overlay has no volume of its own for an app

    async def _Tick(self) -> None:
        while True:
            await asyncio.sleep(TimelineEverySeconds)
            if self.Controls is not None and self.Remote.ThePlayer.Playing:
                try:
                    self._Timeline()
                except Exception:
                    pass