SaraTune/MediaControls.py

"""SaraTune as a media player the system knows about: the desktop's media
controls, media keys and lock screen show what's playing and control it.

Remote is the platform-neutral part: it follows whichever mode played last
and turns system commands into calls on that mode. The backends only talk to
Remote: Mpris.py on Linux (D-Bus), Smtc.py on Windows (System Media
Transport Controls). Neither is required: if one can't start, SaraTune just
plays without it."""

import asyncio
import sys
from pathlib import Path
from typing import Awaitable, Callable, Optional

import flet as ft

from Covers import CoverPath
from Metadata import Track
from Player import Player

ArtSize = 512  # pixels; the cover thumbnail handed to the system
SeekStep = 10_000  # ms, for Windows' fast-forward / rewind buttons


class Remote:
    def __init__(self, Page: ft.Page, Players: dict[str, Player], Modes: dict, Start: str,
                 SetVolume: Callable[[float], None], Quit: Callable[[], Awaitable[None]]):
        self.Page = Page
        self.Players = Players
        self.Modes = Modes
        self.ActiveName = Start  # the mode the controls show
        self.SetVolumeEverywhere = SetVolume
        self._Quit = Quit
        self.TrackNumber = 0  # changes with every new track; MPRIS track IDs are made from it
        self.ArtPath: Optional[Path] = None
        self._Shown: Optional[tuple] = None  # (mode, track) the backend last heard about
        self.Backend = None
        for Name, ThePlayer in Players.items():
            ThePlayer.Watchers.append(lambda Kind, Name=Name: self._PlayerEvent(Name, Kind))

    async def Start(self) -> None:
        try:
            if sys.platform.startswith("linux"):
                from Mpris import MprisBackend
                self.Backend = MprisBackend(self)
            elif sys.platform == "win32":
                from Smtc import SmtcBackend
                self.Backend = SmtcBackend(self)
            if self.Backend is not None:
                await self.Backend.Start()
        except Exception as Error:  # no session bus, no WinRT, ...: play without system controls
            print(f"SaraTune: system media controls unavailable: {Error!r}", file=sys.stderr)
            self.Backend = None
            return
        await self._TrackChanged()

    async def Stop(self) -> None:
        if self.Backend is not None:
            try:
                await self.Backend.Stop()
            except Exception:
                pass

    # --- What's playing -----------------------------------------------------

    @property
    def Mode(self):
        return self.Modes[self.ActiveName]

    @property
    def ThePlayer(self) -> Player:
        return self.Players[self.ActiveName]

    @property
    def Info(self) -> Optional[Track]:
        return self.Mode.NowPlaying if self.ThePlayer.Source else None

    @property
    def Shuffle(self) -> bool:
        return bool(getattr(self.Mode, "Queue", None) and self.Mode.Queue.Shuffle)

    @property
    def Repeat(self) -> str:
        """Off, All or One. Audio mode's loop counts as One."""
        Queue = getattr(self.Mode, "Queue", None)
        if Queue is not None:
            return Queue.Repeat
        return "One" if self.ThePlayer.Loop else "Off"

    def _PlayerEvent(self, Name: str, Kind: str) -> None:
        if self.Backend is None:
            return
        if Kind == "Started" and Name != self.ActiveName:
            self.ActiveName = Name
            self.Page.run_task(self._TrackChanged)
            return
        if Name != self.ActiveName:
            return
        if Kind == "Changed":
            self.Backend.StateChanged()
        elif Kind == "Seeked":
            self.Backend.Seeked()
        elif Kind == "Volume":
            self.Backend.VolumeChanged()

    def MediaChanged(self) -> None:
        """A mode's NowPlaying changed (Main hooks this to App.MediaChanged)."""
        if self.Backend is not None:
            self.Page.run_task(self._TrackChanged)

    def ModesChanged(self) -> None:
        """Shuffle, repeat or loop changed in the app."""
        if self.Backend is not None:
            self.Backend.StateChanged()

    async def _TrackChanged(self) -> None:
        Info = self.Info
        Key = (self.ActiveName, Info.Path if Info else None, id(Info))
        if Key != self._Shown:
            self._Shown = Key
            self.TrackNumber += 1
            self.ArtPath = await asyncio.to_thread(CoverPath, Info.Path, ArtSize) if Info else None
            if self._Shown != Key:
                return  # changed again while the cover was being made
        if self.Backend is not None:
            self.Backend.TrackChanged()

    # --- Commands from the system -------------------------------------------

    def Run(self, Coroutine, *Arguments) -> None:
        """Backends call this from their callbacks; the work runs on Flet's loop."""
        self.Page.run_task(Coroutine, *Arguments)

    async def Play(self) -> None:
        if not self.ThePlayer.Playing:
            await self.ThePlayer.Toggle()

    async def Pause(self) -> None:
        await self.ThePlayer.Pause()

    async def Toggle(self) -> None:
        await self.ThePlayer.Toggle()

    async def StopPlaying(self) -> None:
        """MPRIS Stop: pause and go back to the start, keeping the track."""
        await self.ThePlayer.Pause()
        await self.ThePlayer.Seek(0)

    async def Next(self) -> None:
        await self.Mode.Next()

    async def Previous(self) -> None:
        await self.Mode.Previous()

    async def SeekBy(self, Milliseconds: int) -> None:
        Target = self.ThePlayer.Position + Milliseconds
        if self.ThePlayer.Duration and Target >= self.ThePlayer.Duration:
            await self.Mode.Next()  # MPRIS: seeking past the end acts like Next
        else:
            await self.ThePlayer.Seek(max(0, Target))

    async def SetPosition(self, Milliseconds: int) -> None:
        await self.ThePlayer.Seek(Milliseconds)

    def SetVolume(self, Volume: float) -> None:
        self.SetVolumeEverywhere(max(0.0, min(1.0, Volume)))
        for Mode in self.Modes.values():
            Mode.Bar.SetVolume(self.ThePlayer.Volume)
            Mode._Show(Mode.Bar)

    def SetShuffle(self, Enabled: bool) -> None:
        """The modes tell the backend back through ModesChanged."""
        if hasattr(self.Mode, "SetShuffle"):
            self.Mode.SetShuffle(Enabled)
        else:
            self.ModesChanged()  # Audio mode has no shuffle: put the system's switch back

    def SetRepeat(self, Mode: str) -> None:
        if hasattr(self.Mode, "SetRepeat"):
            self.Mode.SetRepeat(Mode)
        else:
            self.Mode.SetLoop(Mode == "One")

    async def Raise(self) -> None:
        await self.Page.window.to_front()

    async def Quit(self) -> None:
        await self._Quit()