curds: Add a playlist notifications system
This is mostly needed by the UI but there are a few cases where we might need it internally, such as adding newly creating tracks to all the required playlists. Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
b9a563024f
commit
713e8b2dd3
|
@ -17,6 +17,7 @@ class LibraryPlaylist(playlist.Playlist):
|
|||
def thread_add(self, path):
|
||||
track = tags.Track.add(path)
|
||||
self.add(track)
|
||||
self.notify("on-scan", track)
|
||||
|
||||
def thread_scan(self):
|
||||
for dirname, subdirs, files in os.walk(self.name):
|
||||
|
|
|
@ -1,12 +1,22 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
|
||||
class Playlist(list):
|
||||
notifications = dict()
|
||||
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
|
||||
def add(self, track):
|
||||
if track is not None:
|
||||
self.append(track)
|
||||
self.notify("on-add", track)
|
||||
|
||||
def notify_me(name, func):
|
||||
Playlist.notifications.setdefault(name, []).append(func)
|
||||
|
||||
def notify(self, name, *args):
|
||||
for cb in Playlist.notifications.get(name, []):
|
||||
cb(self, *args)
|
||||
|
||||
def runtime(self):
|
||||
m, s = divmod(sum([ track.length for track in self ]), 60)
|
||||
|
|
|
@ -14,6 +14,7 @@ class TestLibraryPlaylist(unittest.TestCase):
|
|||
library.library_thread.stop()
|
||||
library.library_thread = threadqueue.ThreadQueue()
|
||||
tags.tag_map.clear()
|
||||
self.scan_count = 0
|
||||
|
||||
def tearDownClass():
|
||||
library.library_thread.stop()
|
||||
|
@ -25,9 +26,15 @@ class TestLibraryPlaylist(unittest.TestCase):
|
|||
self.assertIsInstance(plist, playlist.Playlist)
|
||||
self.assertEqual(plist.name, test_library)
|
||||
|
||||
def on_scan(self, plist, track):
|
||||
self.scan_count += 1
|
||||
|
||||
def test_playlist_library_scan(self):
|
||||
library.LibraryPlaylist.notify_me("on-scan", self.on_scan)
|
||||
|
||||
plist = library.LibraryPlaylist(test_library)
|
||||
self.assertGreater(library.library_thread.qsize(), 0)
|
||||
library.library_thread.join()
|
||||
self.assertEqual(len(plist), 1250)
|
||||
self.assertEqual(self.scan_count, 1250)
|
||||
self.assertEqual(plist.runtime(), "1 hour, 54 minutes, 35 seconds")
|
||||
|
|
|
@ -9,6 +9,8 @@ test_library = os.path.abspath("./trier/Test Library")
|
|||
class TestPlaylist(unittest.TestCase):
|
||||
def setUp(self):
|
||||
tags.tag_map.clear()
|
||||
self.cb_plist = None
|
||||
self.cb_track = None
|
||||
|
||||
def test_playlist_init(self):
|
||||
plist = playlist.Playlist("Test Playlist")
|
||||
|
@ -55,3 +57,26 @@ class TestPlaylist(unittest.TestCase):
|
|||
self.assertEqual(plist.runtime(), "1 day")
|
||||
track.length = 172800
|
||||
self.assertEqual(plist.runtime(), "2 days")
|
||||
|
||||
def on_add1(self, plist, track):
|
||||
self.cb_plist = plist
|
||||
def on_add2(self, plist, track):
|
||||
self.cb_track = track
|
||||
|
||||
def test_playlist_notifications(self):
|
||||
path = os.path.join(test_library, "Test Artist 01", "Test Album 1")
|
||||
track = tags.Track.add(os.path.join(path, "01 - Test Track 01.ogg"))
|
||||
plist = playlist.Playlist("Test Playlist")
|
||||
self.assertIsInstance(plist.notifications, dict)
|
||||
|
||||
playlist.Playlist.notify_me("on-add", self.on_add1)
|
||||
playlist.Playlist.notify_me("on-add", self.on_add2)
|
||||
self.assertIsInstance(playlist.Playlist.notifications["on-add"], list)
|
||||
self.assertEqual(len(playlist.Playlist.notifications["on-add"]), 2)
|
||||
|
||||
plist.add(track)
|
||||
self.assertEqual(self.cb_plist, plist)
|
||||
self.assertEqual(self.cb_track, track)
|
||||
plist.add(None)
|
||||
self.assertEqual(self.cb_plist, plist)
|
||||
self.assertEqual(self.cb_track, track)
|
||||
|
|
Loading…
Reference in New Issue