Source code for baclassic.macmusicapp

# Released under the MIT License. See LICENSE for details.
#
"""Music playback functionality using the Mac Music (formerly iTunes) app."""
from __future__ import annotations

import logging
import threading
from collections import deque
from typing import TYPE_CHECKING, override

import babase

from baclassic._music import MusicPlayer

if TYPE_CHECKING:
    from typing import Callable, Any

    import bauiv1


[docs] class MacMusicAppMusicPlayer(MusicPlayer): """A music-player that utilizes the macOS Music.app for playback. Allows selecting playlists as entries. """ def __init__(self) -> None: super().__init__() self._thread = _MacMusicAppThread() self._thread.start()
[docs] @override def on_select_entry( self, callback: Callable[[Any], None], current_entry: Any, selection_target_name: str, ) -> bauiv1.MainWindow: # pylint: disable=cyclic-import from bauiv1lib.soundtrack import entrytypeselect as etsel return etsel.SoundtrackEntryTypeSelectWindow( callback, current_entry, selection_target_name )
[docs] @override def on_set_volume(self, volume: float) -> None: self._thread.set_volume(volume)
[docs] def get_playlists(self, callback: Callable) -> None: """Asynchronously fetch the list of available iTunes playlists.""" self._thread.get_playlists(callback)
[docs] @override def on_play(self, entry: Any) -> None: assert babase.app.classic is not None music = babase.app.classic.music entry_type = music.get_soundtrack_entry_type(entry) if entry_type == 'iTunesPlaylist': self._thread.play_playlist(music.get_soundtrack_entry_name(entry)) else: print( 'MacMusicAppMusicPlayer passed unrecognized entry type:', entry_type, )
[docs] @override def on_stop(self) -> None: self._thread.play_playlist(None)
[docs] @override def on_app_shutdown(self) -> None: self._thread.shutdown()
class _MacMusicAppThread(threading.Thread): """Thread which wrangles Music.app playback""" def __init__(self) -> None: super().__init__() self._commands_available = threading.Event() self._commands = deque[list]() self._volume = 1.0 self._current_playlist: str | None = None self._orig_volume: int | None = None @override def run(self) -> None: """Run the Music.app thread.""" babase.set_thread_name('BA_MacMusicAppThread') # Let's mention to the user we're launching Music.app in case # it causes any funny business (this used to background the app # sometimes, though I think that is fixed now) def do_print() -> None: babase.apptimer( 0.5, babase.Call( babase.screenmessage, babase.Lstr(resource='usingItunesText'), (0, 1, 0), ), ) babase.pushcall(do_print, from_other_thread=True) babase.mac_music_app_init() done = False while not done: self._commands_available.wait() self._commands_available.clear() # We're not protecting this list with a mutex but we're # just using it as a simple queue so it should be fine. while self._commands: cmd = self._commands.popleft() if cmd[0] == 'DIE': self._handle_die_command() done = True break if cmd[0] == 'PLAY': self._handle_play_command(target=cmd[1]) elif cmd[0] == 'GET_PLAYLISTS': self._handle_get_playlists_command(target=cmd[1]) del cmd # Allows the command data/callback/etc to be freed. def set_volume(self, volume: float) -> None: """Set volume to a value between 0 and 1.""" old_volume = self._volume self._volume = volume # If we've got nothing we're supposed to be playing, # don't touch itunes/music. if self._current_playlist is None: return # If volume is going to zero, stop actually playing # but don't clear playlist. if old_volume > 0.0 and volume == 0.0: try: assert self._orig_volume is not None babase.mac_music_app_stop() babase.mac_music_app_set_volume(self._orig_volume) except Exception as exc: print('Error stopping iTunes music:', exc) elif self._volume > 0: # If volume was zero, store pre-playing volume and start # playing. if old_volume == 0.0: self._orig_volume = babase.mac_music_app_get_volume() self._update_mac_music_app_volume() if old_volume == 0.0: self._play_current_playlist() def play_playlist(self, musictype: str | None) -> None: """Play the given playlist.""" self._commands.append(['PLAY', musictype]) self._commands_available.set() def shutdown(self) -> None: """Request that the player shuts down.""" self._commands.append(['DIE']) self._commands_available.set() self.join() def get_playlists(self, callback: Callable[[Any], None]) -> None: """Request the list of playlists.""" self._commands.append(['GET_PLAYLISTS', callback]) self._commands_available.set() def _handle_get_playlists_command( self, target: Callable[[list[str]], None] ) -> None: try: playlists = babase.mac_music_app_get_playlists() playlists = [ p for p in playlists if p not in [ 'Music', 'Movies', 'TV Shows', 'Podcasts', 'iTunes\xa0U', 'Books', 'Genius', 'iTunes DJ', 'Music Videos', 'Home Videos', 'Voice Memos', 'Audiobooks', ] ] playlists.sort(key=lambda x: x.lower()) except Exception as exc: print('Error getting iTunes playlists:', exc) playlists = [] babase.pushcall(babase.Call(target, playlists), from_other_thread=True) def _handle_play_command(self, target: str | None) -> None: if target is None: if self._current_playlist is not None and self._volume > 0: try: assert self._orig_volume is not None babase.mac_music_app_stop() babase.mac_music_app_set_volume(self._orig_volume) except Exception as exc: print('Error stopping iTunes music:', exc) self._current_playlist = None else: # If we've got something playing with positive # volume, stop it. if self._current_playlist is not None and self._volume > 0: try: assert self._orig_volume is not None babase.mac_music_app_stop() babase.mac_music_app_set_volume(self._orig_volume) except Exception as exc: print('Error stopping iTunes music:', exc) # Set our playlist and play it if our volume is up. self._current_playlist = target if self._volume > 0: self._orig_volume = babase.mac_music_app_get_volume() self._update_mac_music_app_volume() self._play_current_playlist() def _handle_die_command(self) -> None: # Only stop if we've actually played something # (we don't want to kill music the user has playing). if self._current_playlist is not None and self._volume > 0: try: assert self._orig_volume is not None babase.mac_music_app_stop() babase.mac_music_app_set_volume(self._orig_volume) except Exception as exc: print('Error stopping iTunes music:', exc) def _play_current_playlist(self) -> None: try: assert self._current_playlist is not None if babase.mac_music_app_play_playlist(self._current_playlist): pass else: babase.pushcall( babase.Call( babase.screenmessage, babase.app.lang.get_resource('playlistNotFoundText') + ': \'' + self._current_playlist + '\'', (1, 0, 0), ), from_other_thread=True, ) except Exception: logging.exception( "Error playing playlist '%s'.", self._current_playlist ) def _update_mac_music_app_volume(self) -> None: babase.mac_music_app_set_volume( max(0, min(100, int(100.0 * self._volume))) )