diff --git a/beetsplug/mpdstats.py b/beetsplug/mpdstats.py index fa0791527c..77bafb2829 100644 --- a/beetsplug/mpdstats.py +++ b/beetsplug/mpdstats.py @@ -2,9 +2,10 @@ import os import time -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict, overload import mpd +from typing_extensions import NotRequired from beets import config, plugins, ui from beets.dbcore import types @@ -15,7 +16,62 @@ if TYPE_CHECKING: import optparse - from beets.library import Library + from beets.library import Item, Library + from beets.logging import BeetsLogger as Logger + + from ._typing import JSONDict + + +#: When playlist is empty and status is "stop", it is an empty dictionary. +MPDCurrentSong = TypedDict( + "MPDCurrentSong", + { + "added": NotRequired[str], + "artist": NotRequired[str], + "date": NotRequired[str], + "duration": NotRequired[str], + "file": NotRequired[str], + "format": NotRequired[str], + "id": NotRequired[str], + "last-modified": NotRequired[str], + "pos": NotRequired[str], + "time": NotRequired[str], + "title": NotRequired[str], + }, +) + + +class MPDStatus(TypedDict): + state: Literal["play", "pause", "stop"] + volume: str + repeat: str + random: str + single: str + consume: str + partition: str + playlist: str + playlistlength: str + mixrampdb: str + lastloadedplaylist: str + song: str + songid: str + # below are only set when status is "play" or "pause" + time: NotRequired[str] + elapsed: NotRequired[str] + bitrate: NotRequired[str] + duration: NotRequired[str] + audio: NotRequired[str] + nextsong: NotRequired[str] + nextsongid: NotRequired[str] + + +class NowPlaying(TypedDict): + started: float + elapsed_at_start: int + duration: int + path: str + id: str + beets_item: Item | None # If we lose the connection, how many times do we want to retry and how @@ -28,7 +84,7 @@ mpd_config = config["mpd"] -def is_url(path): +def is_url(path: str) -> bool: """Try to determine if the path is an URL.""" if isinstance(path, bytes): # if it's bytes, then it's a path return False @@ -36,7 +92,7 @@ def is_url(path): class MPDClientWrapper: - def __init__(self, log): + def __init__(self, log: Logger) -> None: self._log = log self.music_directory = mpd_config["music_directory"].as_str() @@ -51,7 +107,7 @@ def __init__(self, log): self.client = mpd.MPDClient() - def connect(self): + def connect(self) -> None: """Connect to the MPD.""" host = mpd_config["host"].as_str() port = mpd_config["port"].get(int) @@ -72,12 +128,26 @@ def connect(self): except mpd.CommandError as e: raise UserError(f"could not authenticate to MPD: {e}") - def disconnect(self): + def disconnect(self) -> None: """Disconnect from the MPD.""" self.client.close() self.client.disconnect() - def get(self, command, retries=RETRIES): + @overload + def get( + self, command: Literal["currentsong"], retries: int = RETRIES + ) -> MPDCurrentSong: ... + @overload + def get( + self, command: Literal["status"], retries: int = RETRIES + ) -> MPDStatus: ... + @overload + def get( + self, command: Literal["idle"], retries: int = RETRIES + ) -> list[str]: ... + @overload + def get(self, command: str, retries: int = RETRIES) -> Any: ... + def get(self, command: str, retries: int = RETRIES) -> Any: """Wrapper for requests to the MPD server. Tries to re-connect if the connection was lost (f.ex. during MPD's library refresh). """ @@ -100,7 +170,7 @@ def get(self, command, retries=RETRIES): self.connect() return self.get(command, retries=retries - 1) - def currentsong(self): + def currentsong(self) -> tuple[str | None, str | None]: """Return the path to the currently playing song, along with its songid. Prefixes paths with the music_directory, to get the absolute path. @@ -108,24 +178,20 @@ def currentsong(self): we replace 'strip_path' with ''. `strip_path` defaults to ''. """ - result = None entry = self.get("currentsong") - if "file" in entry: - if not is_url(entry["file"]): - file = entry["file"] - if file.startswith(self.strip_path): - file = file[len(self.strip_path) :] - result = os.path.join(self.music_directory, file) - else: - result = entry["file"] - self._log.debug("returning: {}", result) - return result, entry.get("id") - - def status(self): + file, id_ = entry.get("file"), entry.get("id") + if file and not is_url(file): + if file.startswith(self.strip_path): + file = file[len(self.strip_path) :] + file = os.path.join(self.music_directory, file) + self._log.debug("returning: {}", file) + return file, id_ + + def status(self) -> MPDStatus: """Return the current status of the MPD.""" return self.get("status") - def events(self): + def events(self) -> list[str]: """Return list of events. This may block a long time while waiting for an answer from MPD. """ @@ -133,7 +199,9 @@ def events(self): class MPDStats: - def __init__(self, lib, log): + now_playing: NowPlaying | None = None + + def __init__(self, lib: Library, log: Logger) -> None: self.lib = lib self._log = log @@ -142,11 +210,11 @@ def __init__(self, lib, log): self.played_ratio_threshold = mpd_config["played_ratio_threshold"].get( float ) - - self.now_playing = None self.mpd = MPDClientWrapper(log) - def rating(self, play_count, skip_count, rating, skipped): + def rating( + self, play_count: int, skip_count: int, rating: float, skipped: bool + ) -> float: """Calculate a new rating for a song based on play count, skip count, old rating and the fact if it was skipped or not. """ @@ -157,16 +225,22 @@ def rating(self, play_count, skip_count, rating, skipped): stable = (play_count + 1.0) / (play_count + skip_count + 2.0) return self.rating_mix * stable + (1.0 - self.rating_mix) * rolling - def get_item(self, path): + def get_item(self, path: str) -> Item | None: """Return the beets item related to path.""" - query = PathQuery("path", path) + query = PathQuery("path", os.fsencode(path)) item = self.lib.items(query).get() if item: return item self._log.info("item not found: {}", displayable_path(path)) return None - def update_item(self, item, attribute, value=None, increment=None): + def update_item( + self, + item: Item | None, + attribute: str, + value: float | None = None, + increment: float | None = None, + ) -> None: """Update the beets item. Set attribute to value or increment the value of attribute. If the increment argument is used the value is cast to the corresponding type. @@ -189,7 +263,7 @@ def update_item(self, item, attribute, value=None, increment=None): item, ) - def update_rating(self, item, skipped): + def update_rating(self, item: Item | None, skipped: bool) -> None: """Update the rating for a beets item. The `item` can either be a beets `Item` or None. If the item is None, nothing changes. """ @@ -206,7 +280,7 @@ def update_rating(self, item, skipped): self.update_item(item, "rating", rating) - def handle_song_change(self, song): + def handle_song_change(self, song: NowPlaying) -> bool: """Determine if a song was skipped or not and update its attributes. To this end the difference between the song's supposed end time and the current time is calculated. If it's greater than a threshold, @@ -226,17 +300,17 @@ def handle_song_change(self, song): return skipped - def handle_played(self, song): + def handle_played(self, song: NowPlaying) -> None: """Updates the play count of a song.""" self.update_item(song["beets_item"], "play_count", increment=1) self._log.info("played {}", displayable_path(song["path"])) - def handle_skipped(self, song): + def handle_skipped(self, song: NowPlaying) -> None: """Updates the skip count of a song.""" self.update_item(song["beets_item"], "skip_count", increment=1) self._log.info("skipped {}", displayable_path(song["path"])) - def on_stop(self, status): + def on_stop(self, status: JSONDict) -> None: self._log.info("stop") # if the current song stays the same it means that we stopped on the @@ -246,13 +320,13 @@ def on_stop(self, status): self.now_playing = None - def on_pause(self, status): + def on_pause(self, status: JSONDict) -> None: self._log.info("pause") self.now_playing = None - def on_play(self, status): + def on_play(self, status: JSONDict) -> None: path, songid = self.mpd.currentsong() - if not path: + if not path or not songid: return played, duration = map(int, status["time"].split(":", 1)) @@ -294,20 +368,14 @@ def on_play(self, status): value=int(time.time()), ) - def run(self): + def run(self) -> None: self.mpd.connect() events = ["player"] while True: if "player" in events: status = self.mpd.status() - - handler = getattr(self, f"on_{status['state']}", None) - - if handler: - handler(status) - else: - self._log.debug('unhandled status "{}"', status) + getattr(self, f"on_{status['state']}")(status) events = self.mpd.events() @@ -320,7 +388,7 @@ class MPDStatsPlugin(plugins.BeetsPlugin): "rating": types.FLOAT, } - def __init__(self): + def __init__(self) -> None: super().__init__() mpd_config.add( { @@ -336,7 +404,7 @@ def __init__(self): ) mpd_config["password"].redact = True - def commands(self): + def commands(self) -> list[ui.Subcommand]: cmd = ui.Subcommand( "mpdstats", help="run a MPD client to gather play statistics" ) diff --git a/beetsplug/thumbnails.py b/beetsplug/thumbnails.py index f25d8ab3ba..d8857b8361 100644 --- a/beetsplug/thumbnails.py +++ b/beetsplug/thumbnails.py @@ -12,7 +12,7 @@ import shutil from hashlib import md5 from pathlib import PurePosixPath -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from xdg import BaseDirectory @@ -40,7 +40,7 @@ def __init__(self) -> None: if self.config["auto"] and self._check_local_ok(): self.register_listener("art_set", self.process_album) - def commands(self): + def commands(self) -> list[Subcommand]: thumbnails_command = Subcommand( "thumbnails", help="Create album thumbnails" ) @@ -72,7 +72,7 @@ def process_query( for album in lib.albums(args): self.process_album(album) - def _check_local_ok(self): + def _check_local_ok(self) -> bool: """Check that everything is ready: - local capability to resize images - thumbnail dirs exist (create them if needed) @@ -97,7 +97,7 @@ def _check_local_ok(self): ) self._log.debug("using {.shared.method} to write metadata", ArtResizer) - uri_getter = GioURI() + uri_getter: URIGetter = GioURI() if not uri_getter.available: uri_getter = PathlibURI() self._log.debug("using {.name} to compute URIs", uri_getter) @@ -108,14 +108,16 @@ def _check_local_ok(self): def process_album(self, album: Album) -> None: """Produce thumbnails for the album folder.""" self._log.debug("generating thumbnail for {}", album) - if not album.artpath: - self._log.info("album {} has no art", album) + + artpath = album.artpath + if not artpath: + self._log.warning("album {} has no art", album) return if self.config["dolphin"]: - self.make_dolphin_cover_thumbnail(album) + self.make_dolphin_cover_thumbnail(album.path, artpath) - size = ArtResizer.shared.get_size(album.artpath) + size = ArtResizer.shared.get_size(artpath) if not size: self._log.warning( "problem getting the picture size for {.artpath}", album @@ -124,15 +126,17 @@ def process_album(self, album: Album) -> None: wrote = True if max(size) >= 256: - wrote &= self.make_cover_thumbnail(album, 256, LARGE_DIR) - wrote &= self.make_cover_thumbnail(album, 128, NORMAL_DIR) + wrote &= self.make_cover_thumbnail(album, artpath, 256, LARGE_DIR) + wrote &= self.make_cover_thumbnail(album, artpath, 128, NORMAL_DIR) if wrote: self._log.info("wrote thumbnail for {}", album) else: self._log.info("nothing to do for {}", album) - def make_cover_thumbnail(self, album, size, target_dir): + def make_cover_thumbnail( + self, album: Album, artpath: bytes, size: int, target_dir: bytes + ) -> bool: """Make a thumbnail of given size for `album` and put it in `target_dir`. """ @@ -141,7 +145,7 @@ def make_cover_thumbnail(self, album, size, target_dir): if ( os.path.exists(syspath(target)) and os.stat(syspath(target)).st_mtime - > os.stat(syspath(album.artpath)).st_mtime + > os.stat(syspath(artpath)).st_mtime ): if self.config["force"]: self._log.debug( @@ -157,12 +161,12 @@ def make_cover_thumbnail(self, album, size, target_dir): album, ) return False - resized = ArtResizer.shared.resize(size, album.artpath, target) - self.add_tags(album, resized) + resized = ArtResizer.shared.resize(size, artpath, target) + self.add_tags(artpath, resized) shutil.move(syspath(resized), syspath(target)) return True - def thumbnail_file_name(self, path): + def thumbnail_file_name(self, path: bytes) -> bytes: """Compute the thumbnail file name See https://standards.freedesktop.org/thumbnail-spec/latest/x227.html """ @@ -170,13 +174,13 @@ def thumbnail_file_name(self, path): hash_ = md5(uri.encode("utf-8")).hexdigest() return bytestring_path(f"{hash_}.png") - def add_tags(self, album, image_path): + def add_tags(self, artpath: bytes, image_path: bytes) -> None: """Write required metadata to the thumbnail See https://standards.freedesktop.org/thumbnail-spec/latest/x142.html """ - mtime = os.stat(syspath(album.artpath)).st_mtime + mtime = os.stat(syspath(artpath)).st_mtime metadata = { - "Thumb::URI": self.get_uri(album.artpath), + "Thumb::URI": self.get_uri(artpath), "Thumb::MTime": str(mtime), } try: @@ -186,11 +190,13 @@ def add_tags(self, album, image_path): "could not write metadata to {}", displayable_path(image_path) ) - def make_dolphin_cover_thumbnail(self, album): - outfilename = os.path.join(album.path, b".directory") + def make_dolphin_cover_thumbnail( + self, album_path: bytes, artpath: bytes + ) -> None: + outfilename = os.path.join(album_path, b".directory") if os.path.exists(syspath(outfilename)): return - artfile = os.path.split(album.artpath)[1] + artfile = os.path.split(artpath)[1] with open(syspath(outfilename), "w") as f: f.write("[Desktop Entry]\n") f.write(f"Icon=./{artfile.decode('utf-8')}") @@ -202,7 +208,7 @@ class URIGetter: available = False name = "Abstract base" - def uri(self, path): + def uri(self, path: bytes) -> str: raise NotImplementedError() @@ -210,11 +216,11 @@ class PathlibURI(URIGetter): available = True name = "Python Pathlib" - def uri(self, path): + def uri(self, path: bytes) -> str: return PurePosixPath(os.fsdecode(path)).as_uri() -def copy_c_string(c_string): +def copy_c_string(c_string: Any) -> bytes | None: """Copy a `ctypes.POINTER(ctypes.c_char)` value into a new Python string and return it. The old memory is then safe to free. """ @@ -232,7 +238,7 @@ class GioURI(URIGetter): def __init__(self) -> None: self.libgio = self.get_library() self.available = bool(self.libgio) - if self.available: + if self.libgio: self.libgio.g_type_init() # for glib < 2.36 self.libgio.g_file_new_for_path.argtypes = [ctypes.c_char_p] @@ -243,28 +249,32 @@ def __init__(self) -> None: self.libgio.g_object_unref.argtypes = [ctypes.c_void_p] - def get_library(self): + def get_library(self) -> ctypes.CDLL | None: lib_name = ctypes.util.find_library("gio-2") try: if not lib_name: - return False + return None return ctypes.cdll.LoadLibrary(lib_name) except OSError: - return False + return None + + def uri(self, path: bytes) -> str: + libgio = self.libgio + if libgio is None: + raise RuntimeError("GIO library is unavailable") - def uri(self, path): - g_file_ptr = self.libgio.g_file_new_for_path(path) + g_file_ptr = libgio.g_file_new_for_path(path) if not g_file_ptr: raise RuntimeError( f"No gfile pointer received for {displayable_path(path)}" ) try: - uri_ptr = self.libgio.g_file_get_uri(g_file_ptr) + uri_ptr = libgio.g_file_get_uri(g_file_ptr) finally: - self.libgio.g_object_unref(g_file_ptr) + libgio.g_object_unref(g_file_ptr) if not uri_ptr: - self.libgio.g_free(uri_ptr) + libgio.g_free(uri_ptr) raise RuntimeError( f"No URI received from the gfile pointer for {displayable_path(path)}" ) @@ -272,7 +282,10 @@ def uri(self, path): try: uri = copy_c_string(uri_ptr) finally: - self.libgio.g_free(uri_ptr) + libgio.g_free(uri_ptr) + + if uri is None: + raise RuntimeError("GIO returned NULL for filename") try: return os.fsdecode(uri) diff --git a/test/plugins/test_mpdstats.py b/test/plugins/test_mpdstats.py index a96a8ed687..329e2b1ebe 100644 --- a/test/plugins/test_mpdstats.py +++ b/test/plugins/test_mpdstats.py @@ -33,7 +33,6 @@ def test_get_item(self): assert "item not found:" in log.info.call_args[0][0] STATUSES: ClassVar[list[dict[str, Any]]] = [ - {"state": "some-unknown-one"}, {"state": "pause"}, {"state": "play", "songid": 1, "time": "0:1"}, {"state": "stop"}, @@ -62,7 +61,6 @@ def test_run_mpdstats(self, mpd_mock): except KeyboardInterrupt: pass - log.debug.assert_has_calls([call('unhandled status "{}"', ANY)]) log.info.assert_has_calls( [call("pause"), call("playing {}", ANY), call("stop")] ) diff --git a/test/plugins/test_thumbnails.py b/test/plugins/test_thumbnails.py index 51e84a8f15..8f58ca213f 100644 --- a/test/plugins/test_thumbnails.py +++ b/test/plugins/test_thumbnails.py @@ -23,16 +23,16 @@ def test_add_tags(self, mock_stat, mock_artresizer): plugin.get_uri = Mock( side_effect={b"/path/to/cover": "COVER_URI"}.__getitem__ ) - album = Mock(artpath=b"/path/to/cover") + artpath = b"/path/to/cover" mock_stat.return_value.st_mtime = 12345 - plugin.add_tags(album, b"/path/to/thumbnail") + plugin.add_tags(artpath, b"/path/to/thumbnail") metadata = {"Thumb::URI": "COVER_URI", "Thumb::MTime": "12345"} mock_artresizer.shared.write_metadata.assert_called_once_with( b"/path/to/thumbnail", metadata ) - mock_stat.assert_called_once_with(syspath(album.artpath)) + mock_stat.assert_called_once_with(syspath(artpath)) @patch("beetsplug.thumbnails.os") @patch("beetsplug.thumbnails.ArtResizer") @@ -88,26 +88,26 @@ def exists(path): def test_make_cover_thumbnail(self, mock_shutils, mock_os, mock_artresizer): thumbnail_dir = os.path.normpath(b"/thumbnail/dir") md5_file = os.path.join(thumbnail_dir, b"md5") - path_to_art = os.path.normpath(b"/path/to/art") + artpath = os.path.normpath(b"/path/to/art") path_to_resized_art = os.path.normpath(b"/path/to/resized/artwork") mock_os.path.join = os.path.join # don't mock that function plugin = ThumbnailsPlugin() plugin.add_tags = Mock() - album = Mock(artpath=path_to_art) + album = Mock(artpath=artpath) plugin.thumbnail_file_name = Mock(return_value=b"md5") mock_os.path.exists.return_value = False mock_resize = mock_artresizer.shared.resize mock_resize.return_value = path_to_resized_art - plugin.make_cover_thumbnail(album, 12345, thumbnail_dir) + plugin.make_cover_thumbnail(album, artpath, 12345, thumbnail_dir) mock_os.path.exists.assert_called_once_with(syspath(md5_file)) - mock_resize.assert_called_once_with(12345, path_to_art, md5_file) - plugin.add_tags.assert_called_once_with(album, path_to_resized_art) + mock_resize.assert_called_once_with(12345, artpath, md5_file) + plugin.add_tags.assert_called_once_with(artpath, path_to_resized_art) mock_shutils.move.assert_called_once_with( syspath(path_to_resized_art), syspath(md5_file) ) @@ -120,19 +120,19 @@ def test_make_cover_thumbnail(self, mock_shutils, mock_os, mock_artresizer): def os_stat(target): if target == syspath(md5_file): return Mock(st_mtime=3) - if target == syspath(path_to_art): + if target == syspath(artpath): return Mock(st_mtime=2) raise ValueError(f"invalid target {target}") mock_os.stat.side_effect = os_stat - plugin.make_cover_thumbnail(album, 12345, thumbnail_dir) + plugin.make_cover_thumbnail(album, artpath, 12345, thumbnail_dir) assert mock_resize.call_count == 0 # and with force plugin.config["force"] = True - plugin.make_cover_thumbnail(album, 12345, thumbnail_dir) - mock_resize.assert_called_once_with(12345, path_to_art, md5_file) + plugin.make_cover_thumbnail(album, artpath, 12345, thumbnail_dir) + mock_resize.assert_called_once_with(12345, artpath, md5_file) @patch("beetsplug.thumbnails.ThumbnailsPlugin._check_local_ok", Mock()) def test_make_dolphin_cover_thumbnail(self): @@ -141,13 +141,13 @@ def test_make_dolphin_cover_thumbnail(self): album = Mock( path=os.fsencode(tmp), artpath=os.fsencode(tmp / "cover.jpg") ) - plugin.make_dolphin_cover_thumbnail(album) + plugin.make_dolphin_cover_thumbnail(album.path, album.artpath) filename = tmp / ".directory" assert filename.read_text() == "[Desktop Entry]\nIcon=./cover.jpg" # not rewritten when it already exists (yup that's a big limitation) album.artpath = b"/my/awesome/art.tiff" - plugin.make_dolphin_cover_thumbnail(album) + plugin.make_dolphin_cover_thumbnail(album.path, album.artpath) assert filename.read_text() == "[Desktop Entry]\nIcon=./cover.jpg" @patch("beetsplug.thumbnails.ThumbnailsPlugin._check_local_ok", Mock()) @@ -179,19 +179,24 @@ def test_process_album(self, mock_artresizer): plugin.config["dolphin"] = True plugin.process_album(album) - make_dolphin.assert_called_once_with(album) + make_dolphin.assert_called_once_with(album.path, album.artpath) # small art get_size.return_value = 200, 200 plugin.process_album(album) - make_cover.assert_called_once_with(album, 128, NORMAL_DIR) + make_cover.assert_called_once_with( + album, album.artpath, 128, NORMAL_DIR + ) # big art make_cover.reset_mock() get_size.return_value = 500, 500 plugin.process_album(album) make_cover.assert_has_calls( - [call(album, 128, NORMAL_DIR), call(album, 256, LARGE_DIR)], + [ + call(album, album.artpath, 128, NORMAL_DIR), + call(album, album.artpath, 256, LARGE_DIR), + ], any_order=True, )