Source code for baclassic.osmusic

# Released under the MIT License. See LICENSE for details.
#
"""Music playback using OS functionality exposed through the C++ layer."""
from __future__ import annotations

import os
import random
import logging
import threading
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 OSMusicPlayer(MusicPlayer): """Music player that talks to internal C++ layer for functionality. (internal)""" def __init__(self) -> None: super().__init__() self._want_to_play = False self._actually_playing = False
[docs] @classmethod def get_valid_music_file_extensions(cls) -> list[str]: """Return file extensions for types playable on this device.""" # FIXME: should ask the C++ layer for these; just hard-coding for now. return ['mp3', 'ogg', 'm4a', 'wav', 'flac', 'mid']
[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.entrytypeselect import ( SoundtrackEntryTypeSelectWindow, ) return SoundtrackEntryTypeSelectWindow( callback, current_entry, selection_target_name )
[docs] @override def on_set_volume(self, volume: float) -> None: babase.music_player_set_volume(volume)
[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) name = music.get_soundtrack_entry_name(entry) assert name is not None if entry_type == 'musicFile': self._want_to_play = self._actually_playing = True babase.music_player_play(name) elif entry_type == 'musicFolder': # Launch a thread to scan this folder and give us a random # valid file within it. self._want_to_play = True self._actually_playing = False _PickFolderSongThread( name, self.get_valid_music_file_extensions(), self._on_play_folder_cb, ).start()
def _on_play_folder_cb( self, result: str | list[str], error: str | None = None ) -> None: if error is not None: rstr = babase.Lstr( resource='internal.errorPlayingMusicText' ).evaluate() if isinstance(result, str): err_str = ( rstr.replace('${MUSIC}', os.path.basename(result)) + '; ' + str(error) ) else: err_str = ( rstr.replace('${MUSIC}', '<multiple>') + '; ' + str(error) ) babase.screenmessage(err_str, color=(1, 0, 0)) return # There's a chance a stop could have been issued before our thread # returned. If that's the case, don't play. if not self._want_to_play: print('_on_play_folder_cb called with _want_to_play False') else: self._actually_playing = True babase.music_player_play(result)
[docs] @override def on_stop(self) -> None: self._want_to_play = False self._actually_playing = False babase.music_player_stop()
[docs] @override def on_app_shutdown(self) -> None: babase.music_player_shutdown()
class _PickFolderSongThread(threading.Thread): def __init__( self, path: str, valid_extensions: list[str], callback: Callable[[str | list[str], str | None], None], ): super().__init__() self._valid_extensions = valid_extensions self._callback = callback self._path = path @override def run(self) -> None: do_log_error = True try: babase.set_thread_name('BA_PickFolderSongThread') all_files: list[str] = [] valid_extensions = ['.' + x for x in self._valid_extensions] for root, _subdirs, filenames in os.walk(self._path): for fname in filenames: if any( fname.lower().endswith(ext) for ext in valid_extensions ): all_files.insert( random.randrange(len(all_files) + 1), root + '/' + fname, ) if not all_files: do_log_error = False raise RuntimeError( babase.Lstr( resource='internal.noMusicFilesInFolderText' ).evaluate() ) babase.pushcall( babase.Call(self._callback, all_files, None), from_other_thread=True, ) except Exception as exc: if do_log_error: logging.exception('Error in _PickFolderSongThread') try: err_str = str(exc) except Exception: err_str = '<ENCERR4523>' babase.pushcall( babase.Call(self._callback, self._path, err_str), from_other_thread=True, )