mirror of
https://git.srhtnin.garden/sarah/SaraTune.git
synced 2026-10-14 10:18:11 +02:00
232 lines
9.2 KiB
Python
232 lines
9.2 KiB
Python
"""One audio player. Each mode has its own, so switching modes never loses
|
|
what the other was playing; Main pauses one when the other starts.
|
|
|
|
Every track gets a fresh flet_audio.Audio service: changing an existing
|
|
one's src only works the first time (later play calls never answer), and
|
|
sources are file:// URIs, since plain paths are read as URLs and break on
|
|
names with # or %. Events from a replaced service are ignored.
|
|
|
|
The Flet client only reports the position about once a second, so the
|
|
position is kept here and moved along with the clock in between."""
|
|
|
|
import asyncio
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
import flet as ft
|
|
import flet_audio as fta
|
|
|
|
CallTimeout = 8 # seconds; a call the client never answers shouldn't hang the app
|
|
|
|
|
|
class Player:
|
|
def __init__(self, Page: ft.Page, Volume: float):
|
|
self.Page = Page
|
|
self.Volume = Volume
|
|
self.Audio: Optional[fta.Audio] = None
|
|
self.Source: Optional[str] = None
|
|
self.Playing = False
|
|
self.Duration = 0 # milliseconds; 0 until known
|
|
self._Position = 0
|
|
self._PositionAt = time.monotonic()
|
|
self._Ready = asyncio.Event() # the current service has loaded its source
|
|
self.Loop = False
|
|
# Set by the owner. OnEnded: the track ran out by itself.
|
|
self.OnStarted: Callable[[], None] = lambda: None
|
|
self.OnEnded: Callable[[], None] = lambda: None
|
|
self.OnChanged: Callable[[], None] = lambda: None # playing/paused, duration, source
|
|
self.OnError: Callable[[str], None] = lambda Message: None
|
|
# Others following along (the system media controls): called with
|
|
# "Changed", "Started" or "Seeked" after the owner's own callbacks.
|
|
self.Watchers: list[Callable[[str], None]] = []
|
|
|
|
# --- Position -----------------------------------------------------------
|
|
|
|
@property
|
|
def Position(self) -> int:
|
|
if not self.Playing:
|
|
return self._Position
|
|
Moved = self._Position + (time.monotonic() - self._PositionAt) * 1000
|
|
return int(min(Moved, self.Duration) if self.Duration else Moved)
|
|
|
|
def _SetPosition(self, Milliseconds: int) -> None:
|
|
self._Position = max(0, int(Milliseconds))
|
|
self._PositionAt = time.monotonic()
|
|
|
|
def _Tell(self, Kind: str) -> None:
|
|
if Kind == "Changed":
|
|
self.OnChanged()
|
|
for Watcher in self.Watchers:
|
|
Watcher(Kind)
|
|
|
|
# --- Events from the client ---------------------------------------------
|
|
|
|
def _StateChanged(self, Audio: fta.Audio, e: fta.AudioStateChangeEvent) -> None:
|
|
if Audio is not self.Audio:
|
|
return
|
|
if e.state == fta.AudioState.PLAYING:
|
|
if not self.Playing:
|
|
self._SetPosition(self._Position)
|
|
self.Playing = True
|
|
self.OnStarted()
|
|
self._Tell("Started")
|
|
elif e.state == fta.AudioState.COMPLETED:
|
|
self.Playing = False
|
|
self._SetPosition(0)
|
|
if self.Loop:
|
|
self.Page.run_task(self.Play, 0)
|
|
else:
|
|
self.OnEnded()
|
|
elif e.state in (fta.AudioState.PAUSED, fta.AudioState.STOPPED):
|
|
if self.Playing:
|
|
self._SetPosition(self.Position)
|
|
self.Playing = False
|
|
self._Tell("Changed")
|
|
|
|
def _PositionChanged(self, Audio: fta.Audio, e: fta.AudioPositionChangeEvent) -> None:
|
|
if Audio is self.Audio and self.Playing:
|
|
self._SetPosition(e.position)
|
|
|
|
def _DurationChanged(self, Audio: fta.Audio, e: fta.AudioDurationChangeEvent) -> None:
|
|
if Audio is not self.Audio:
|
|
return
|
|
Duration = e.duration
|
|
Value = int(Duration.in_milliseconds if isinstance(Duration, ft.Duration) else Duration or 0)
|
|
if Value > 0:
|
|
self.Duration = Value
|
|
self._Ready.set()
|
|
self._Tell("Changed")
|
|
|
|
def _Loaded(self, Audio: fta.Audio) -> None:
|
|
if Audio is self.Audio:
|
|
self._Ready.set()
|
|
|
|
# --- Control ------------------------------------------------------------
|
|
|
|
async def _Call(self, Method, *Arguments) -> bool:
|
|
try:
|
|
await asyncio.wait_for(Method(*Arguments), CallTimeout)
|
|
return True
|
|
except (asyncio.TimeoutError, RuntimeError) as Error:
|
|
# e.g. GStreamer can't decode it: the client raises a PlatformException.
|
|
Text = str(Error)
|
|
self.OnError("The player didn't answer." if isinstance(Error, asyncio.TimeoutError) else
|
|
Text.split(",")[1].strip() if "PlatformException" in Text and "," in Text else Text)
|
|
return False
|
|
|
|
async def _Unload(self) -> None:
|
|
Old, self.Audio = self.Audio, None
|
|
if Old is None:
|
|
return
|
|
# Taking a service off the page doesn't stop its native player: it
|
|
# would play on, unseen. Stop and free it first.
|
|
for Method in (Old.pause, Old.release):
|
|
try:
|
|
await asyncio.wait_for(Method(), 2)
|
|
except (asyncio.TimeoutError, RuntimeError):
|
|
pass
|
|
if Old in self.Page.services:
|
|
self.Page.services.remove(Old)
|
|
self.Page.update()
|
|
|
|
async def Load(self, FilePath: str, Start: int = 0, Play: bool = True, Duration: int = 0) -> bool:
|
|
"""Switches to FilePath and (if Play) starts at Start (ms). Duration is
|
|
a hint (e.g. from the tags) until the client reports the real one.
|
|
False when the file doesn't exist or can't be played."""
|
|
if not Path(FilePath).is_file():
|
|
return False
|
|
await self._Unload()
|
|
self.Source = FilePath
|
|
self.Duration = Duration
|
|
self.Playing = False
|
|
self._SetPosition(Start)
|
|
self._Ready = asyncio.Event()
|
|
Audio = fta.Audio(
|
|
src=Path(FilePath).absolute().as_uri(),
|
|
volume=self.Volume,
|
|
release_mode=fta.ReleaseMode.STOP,
|
|
)
|
|
# Handlers know which service they belong to, so a replaced one's late events are dropped.
|
|
Audio.on_state_change = lambda e: self._StateChanged(Audio, e)
|
|
Audio.on_position_change = lambda e: self._PositionChanged(Audio, e)
|
|
Audio.on_duration_change = lambda e: self._DurationChanged(Audio, e)
|
|
Audio.on_loaded = lambda e: self._Loaded(Audio)
|
|
self.Audio = Audio
|
|
self.Page.services.append(Audio)
|
|
self.Page.update()
|
|
self._Tell("Changed")
|
|
if Play:
|
|
return await self.Play(Start)
|
|
return True
|
|
|
|
async def _WaitReady(self) -> None:
|
|
try:
|
|
await asyncio.wait_for(self._Ready.wait(), 3)
|
|
except asyncio.TimeoutError:
|
|
pass # play anyway; it answers with an error if the file is bad
|
|
|
|
async def Play(self, Start: Optional[int] = None) -> bool:
|
|
"""From Start (ms), or from where it was."""
|
|
if not self.Source:
|
|
return False
|
|
if self.Audio is None: # stopped: load it again
|
|
return await self.Load(self.Source, self._Position if Start is None else Start, Duration=self.Duration)
|
|
Audio = self.Audio
|
|
await self._WaitReady()
|
|
Position = self._Position if Start is None else Start
|
|
self._SetPosition(Position)
|
|
if Position > 0:
|
|
# play()'s own position isn't reliable before the first play: start, then seek.
|
|
if not await self._Call(Audio.resume):
|
|
return False
|
|
if not await self._Call(Audio.seek, ft.Duration(milliseconds=int(Position))):
|
|
return False
|
|
self._SetPosition(Position) # it played from 0 until the seek landed
|
|
self._Tell("Seeked")
|
|
return True
|
|
return await self._Call(Audio.play, ft.Duration(milliseconds=0))
|
|
|
|
async def Pause(self) -> None:
|
|
if self.Playing and self.Audio is not None:
|
|
self._SetPosition(self.Position)
|
|
self.Playing = False
|
|
await self._Call(self.Audio.pause)
|
|
self._Tell("Changed")
|
|
|
|
async def Toggle(self) -> None:
|
|
if self.Playing:
|
|
await self.Pause()
|
|
else:
|
|
# A finished track (position at the end) starts over.
|
|
AtEnd = self.Duration and self._Position >= self.Duration - 250
|
|
await self.Play(0 if AtEnd else None)
|
|
|
|
async def Seek(self, Milliseconds: int) -> None:
|
|
if not self.Source:
|
|
return
|
|
Milliseconds = max(0, min(int(Milliseconds), self.Duration or int(Milliseconds)))
|
|
self._SetPosition(Milliseconds)
|
|
if self.Playing and self.Audio is not None:
|
|
await self._Call(self.Audio.seek, ft.Duration(milliseconds=Milliseconds))
|
|
else:
|
|
# Paused or not started: remember it, Play starts from there.
|
|
self._Tell("Changed")
|
|
self._Tell("Seeked")
|
|
|
|
async def Stop(self) -> None:
|
|
"""Back to nothing loaded."""
|
|
await self._Unload()
|
|
self.Source = None
|
|
self.Playing = False
|
|
self.Duration = 0
|
|
self._SetPosition(0)
|
|
self._Tell("Changed")
|
|
|
|
def SetVolume(self, Volume: float) -> None:
|
|
self.Volume = max(0.0, min(1.0, Volume))
|
|
if self.Audio is not None:
|
|
self.Audio.volume = self.Volume
|
|
self.Audio.update()
|
|
self._Tell("Volume")
|