179 lines
7.3 KiB
Python
179 lines
7.3 KiB
Python
# Copyright 2023 (c) Anna Schumaker.
|
|
"""An object for managing the visible and active playlists."""
|
|
from gi.repository import GObject
|
|
from . import playlist
|
|
from . import previous
|
|
from .. import db
|
|
|
|
|
|
class Factory(GObject.GObject):
|
|
"""Our playlist model factory."""
|
|
|
|
sql = GObject.Property(type=db.Connection)
|
|
|
|
db_active = GObject.Property(type=db.playlist.Playlist)
|
|
db_previous = GObject.Property(type=db.playlist.Playlist)
|
|
db_visible = GObject.Property(type=db.playlist.Playlist)
|
|
|
|
active_playlist = GObject.Property(type=playlist.Playlist)
|
|
previous_playlist = GObject.Property(type=previous.Previous)
|
|
visible_playlist = GObject.Property(type=playlist.Playlist)
|
|
|
|
can_go_next = GObject.Property(type=bool, default=False)
|
|
can_go_previous = GObject.Property(type=bool, default=False)
|
|
|
|
def __init__(self, sql: db.Connection):
|
|
"""Initialize the Playlist Factory."""
|
|
super().__init__(sql=sql)
|
|
self.sql.bind_property("active-playlist", self, "db-active")
|
|
self.connect("notify", self.__notify_db_playlists)
|
|
self.sql.playlists.connect("notify::have-collection-tracks",
|
|
self.__update_can_go_next)
|
|
|
|
def __get_playlists(self) -> list[playlist.Playlist]:
|
|
plists = [self.active_playlist, self.previous_playlist,
|
|
self.visible_playlist]
|
|
return [p for p in plists if p is not None]
|
|
|
|
def __search_playlists(self, db_plist: db.playlist.Playlist) \
|
|
-> playlist.Playlist:
|
|
for plist in self.__get_playlists():
|
|
if plist.playlist == db_plist:
|
|
return plist
|
|
|
|
def __update_can_go(self, which: str, newval: bool) -> None:
|
|
if self.get_property(f"can-go-{which}") != newval:
|
|
self.set_property(f"can-go-{which}", newval)
|
|
|
|
def __update_can_go_next(self, *args) -> None:
|
|
prev = self.previous_playlist
|
|
active = self.active_playlist
|
|
self.__update_can_go("next",
|
|
self.sql.playlists.have_collection_tracks or
|
|
(active is not None and active.can_go_next) or
|
|
(prev is not None and prev.can_go_forward))
|
|
|
|
def __update_can_go_prev(self, *args) -> None:
|
|
self.__update_can_go("previous", self.previous_playlist.can_go_next)
|
|
|
|
def __playlist_track_requested(self, playlist: playlist.Playlist,
|
|
track: db.tracks.Track) -> None:
|
|
album = isinstance(playlist.playlist, db.albums.Album) or \
|
|
isinstance(playlist.playlist, db.media.Medium)
|
|
self.emit("track-requested", track,
|
|
"album" if album else "track",
|
|
playlist.playlist == self.sql.playlists.previous)
|
|
|
|
def __make_playlist(self,
|
|
db_plist: db.playlist.Playlist) -> playlist.Playlist:
|
|
if db_plist == self.sql.playlists.previous:
|
|
res = previous.Previous(self.sql, db_plist)
|
|
res.connect("notify::can-go-next", self.__update_can_go_prev)
|
|
res.connect("notify::can-go-forward", self.__update_can_go_next)
|
|
else:
|
|
res = playlist.Playlist(self.sql, db_plist)
|
|
res.connect("notify::can-go-next", self.__update_can_go_next)
|
|
res.connect("track-requested", self.__playlist_track_requested)
|
|
return res
|
|
|
|
def __free_playlist(self, plist: playlist.Playlist) -> None:
|
|
plist.playlist = None
|
|
if isinstance(plist, previous.Previous):
|
|
plist.disconnect_by_func(self.__update_can_go_prev)
|
|
plist.disconnect_by_func(self.__update_can_go_next)
|
|
plist.disconnect_by_func(self.__playlist_track_requested)
|
|
|
|
def __run_factory(self, label: str) -> None:
|
|
db_plist = self.get_property(f"db-{label}")
|
|
plist = self.get_property(f"{label}-playlist")
|
|
|
|
print(f"factory: {label} playlist is:",
|
|
"<None>" if db_plist is None else db_plist.name)
|
|
|
|
if db_plist is None:
|
|
if self.__get_playlists().count(plist) == 1:
|
|
self.__free_playlist(plist)
|
|
new = None
|
|
elif plist is None or self.__get_playlists().count(plist) > 1:
|
|
if (new := self.__search_playlists(db_plist)) is None:
|
|
new = self.__make_playlist(db_plist)
|
|
elif (new := self.__search_playlists(db_plist)) is None:
|
|
plist.playlist = db_plist
|
|
return
|
|
else:
|
|
plist.playlist = None
|
|
|
|
self.set_property(f"{label}-playlist", new)
|
|
|
|
def __notify_db_playlists(self, factory: GObject.GObject, param) -> None:
|
|
match param.name:
|
|
case "db-active":
|
|
self.__run_factory("active")
|
|
self.notify("active-loop")
|
|
self.notify("active-shuffle")
|
|
case "db-previous":
|
|
self.__run_factory("previous")
|
|
case "db-visible":
|
|
self.__run_factory("visible")
|
|
|
|
def next_track(self, *, user: bool = False) \
|
|
-> tuple[db.tracks.Track | None, str, bool]:
|
|
"""Get the next track."""
|
|
track = None
|
|
restart = False
|
|
|
|
if user is True:
|
|
track = self.previous_playlist.previous_track()
|
|
restart = track is not None
|
|
if track is None and self.active_playlist is not None:
|
|
track = self.active_playlist.next_track()
|
|
if track is None:
|
|
self.sql.set_active_playlist(self.sql.playlists.collection)
|
|
track = self.active_playlist.next_track()
|
|
|
|
album = isinstance(self.db_active, db.albums.Album) or \
|
|
isinstance(self.db_active, db.media.Medium)
|
|
rg_auto = "album" if album else "track"
|
|
return (track, rg_auto, restart)
|
|
|
|
def previous_track(self) -> tuple[db.tracks.Track | None, bool]:
|
|
"""Get the previous Track."""
|
|
if self.previous_playlist is None:
|
|
return None
|
|
return self.previous_playlist.next_track()
|
|
|
|
@GObject.Property(type=str, flags=playlist.FLAGS)
|
|
def active_loop(self) -> str:
|
|
"""Get the loop state of the active playlist."""
|
|
if self.active_playlist is not None:
|
|
return self.active_playlist.loop
|
|
return "None"
|
|
|
|
@active_loop.setter
|
|
def active_loop(self, newval: str) -> None:
|
|
"""Set the loop state of the active playlist."""
|
|
if self.active_playlist is not None:
|
|
if self.active_playlist.loop != newval:
|
|
self.active_playlist.loop = newval
|
|
self.notify("active-loop")
|
|
|
|
@GObject.Property(type=bool, default=False, flags=playlist.FLAGS)
|
|
def active_shuffle(self) -> bool:
|
|
"""Get the shuffle state of the active playlist."""
|
|
if self.active_playlist is not None:
|
|
return self.active_playlist.shuffle
|
|
return False
|
|
|
|
@active_shuffle.setter
|
|
def active_shuffle(self, newval: bool) -> None:
|
|
"""Set the shuffle state of the active playlist."""
|
|
if self.active_playlist is not None:
|
|
if self.active_playlist.shuffle != newval:
|
|
self.active_playlist.shuffle = newval
|
|
self.notify("active-shuffle")
|
|
|
|
@GObject.Signal(arg_types=(db.tracks.Track, str, bool))
|
|
def track_requested(self, track: db.tracks.Track,
|
|
rg_auto: str, restarted: bool) -> None:
|
|
"""Signal that a track has been requested by the user."""
|