curds: Make playlist notifications more generic
We might want to use these outside of the playlist code, so let's move this to a generic place so it can be easily used. Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
64ab13718c
commit
752ebece3d
|
@ -1,11 +1,13 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import data
|
||||
from . import notify
|
||||
from . import playlist
|
||||
from . import tags
|
||||
from . import threadqueue
|
||||
|
||||
Album = tags.Album
|
||||
DataFile = data.DataFile
|
||||
Notify = notify.Notify
|
||||
Playlist = playlist.playlist.Playlist
|
||||
ThreadQueue = threadqueue.ThreadQueue
|
||||
Track = tags.Track
|
||||
|
|
|
@ -0,0 +1,24 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
|
||||
class Notify:
|
||||
notifications = {}
|
||||
|
||||
def __init__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def clear():
|
||||
Notify.notifications.clear()
|
||||
|
||||
def notify(name, *args):
|
||||
for func in Notify.notifications.get(name, []):
|
||||
func(*args)
|
||||
|
||||
def notify_cancel(name, func):
|
||||
cb = Notify.notifications.get(name, [])
|
||||
if func in cb:
|
||||
cb.remove(func)
|
||||
|
||||
def notify_me(name, func):
|
||||
cb = Notify.notifications.setdefault(name, [])
|
||||
if not func in cb:
|
||||
cb.append(func)
|
|
@ -1,5 +1,6 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import playlist
|
||||
from .. import notify
|
||||
from .. import tags
|
||||
from .. import threadqueue
|
||||
import os
|
||||
|
@ -18,7 +19,7 @@ class LibraryPlaylist(playlist.Playlist):
|
|||
track = tags.Track.add(path)
|
||||
if track != None:
|
||||
self.add(track)
|
||||
self.notify("on-scan", track)
|
||||
notify.Notify.notify("on-scan", self, track)
|
||||
|
||||
def thread_scan(self):
|
||||
for dirname, subdirs, files in os.walk(self.name):
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import library
|
||||
from . import playlist
|
||||
from .. import notify
|
||||
import os
|
||||
|
||||
class LibraryManager(dict):
|
||||
|
@ -16,7 +17,7 @@ class PlaylistManager(dict):
|
|||
self["Collection"] = playlist.Playlist("collection")
|
||||
self["Library"] = LibraryManager()
|
||||
|
||||
playlist.Playlist.notify_me("on-scan", self.on_scan)
|
||||
notify.Notify.notify_me("on-scan", self.on_scan)
|
||||
|
||||
def on_scan(self, plist, track):
|
||||
self["Collection"].add(track)
|
||||
|
|
|
@ -1,27 +1,14 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from .. import notify
|
||||
|
||||
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 never_mind(name, func):
|
||||
cb = Playlist.notifications.get(name, [])
|
||||
if func in cb:
|
||||
cb.remove(func)
|
||||
|
||||
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)
|
||||
notify.Notify.notify("on-add", self, track)
|
||||
|
||||
def runtime(self):
|
||||
m, s = divmod(sum([ track.length for track in self ]), 60)
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import library
|
||||
from . import playlist
|
||||
from .. import notify
|
||||
from .. import tags
|
||||
from .. import threadqueue
|
||||
import os
|
||||
|
@ -10,6 +11,9 @@ import unittest
|
|||
test_library = os.path.abspath("./trier/Test Library")
|
||||
|
||||
class TestLibraryPlaylist(unittest.TestCase):
|
||||
def setUpClass():
|
||||
notify.Notify.clear()
|
||||
|
||||
def setUp(self):
|
||||
library.library_thread.stop()
|
||||
library.library_thread = threadqueue.ThreadQueue()
|
||||
|
@ -30,7 +34,7 @@ class TestLibraryPlaylist(unittest.TestCase):
|
|||
self.scan_count += 1
|
||||
|
||||
def test_playlist_library_scan(self):
|
||||
library.LibraryPlaylist.notify_me("on-scan", self.on_scan)
|
||||
notify.Notify.notify_me("on-scan", self.on_scan)
|
||||
|
||||
plist = library.LibraryPlaylist(test_library)
|
||||
self.assertGreater(library.library_thread.qsize(), 0)
|
||||
|
|
|
@ -2,6 +2,7 @@
|
|||
from . import library
|
||||
from . import manager
|
||||
from . import playlist
|
||||
from .. import notify
|
||||
from .. import tags
|
||||
from .. import threadqueue
|
||||
import os
|
||||
|
@ -12,6 +13,7 @@ test_library = os.path.abspath("./trier/Test Library")
|
|||
|
||||
class TestPlaylistManager(unittest.TestCase):
|
||||
def setUp(self):
|
||||
notify.Notify.clear()
|
||||
tags.tag_map.clear()
|
||||
if not library.library_thread.is_alive():
|
||||
library.library_thread = threadqueue.ThreadQueue()
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import playlist
|
||||
from .. import notify
|
||||
from .. import tags
|
||||
import os
|
||||
import unittest
|
||||
|
@ -9,6 +10,7 @@ test_library = os.path.abspath("./trier/Test Library")
|
|||
class TestPlaylist(unittest.TestCase):
|
||||
def setUp(self):
|
||||
tags.tag_map.clear()
|
||||
notify.Notify.clear()
|
||||
self.cb_plist = None
|
||||
self.cb_track = None
|
||||
|
||||
|
@ -58,21 +60,16 @@ class TestPlaylist(unittest.TestCase):
|
|||
track.length = 172800
|
||||
self.assertEqual(plist.runtime(), "2 days")
|
||||
|
||||
def on_add1(self, plist, track):
|
||||
def on_add(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)
|
||||
notify.Notify.notify_me("on-add", self.on_add)
|
||||
|
||||
plist.add(track)
|
||||
self.assertEqual(self.cb_plist, plist)
|
||||
|
@ -80,13 +77,3 @@ class TestPlaylist(unittest.TestCase):
|
|||
plist.add(None)
|
||||
self.assertEqual(self.cb_plist, plist)
|
||||
self.assertEqual(self.cb_track, track)
|
||||
|
||||
self.cb_plist = None
|
||||
self.cb_track = None
|
||||
track = tags.Track.add(os.path.join(path, "02 - Test Track 02.ogg"))
|
||||
playlist.Playlist.never_mind("on-add", self.on_add1)
|
||||
self.assertEqual(len(playlist.Playlist.notifications["on-add"]), 1)
|
||||
playlist.Playlist.never_mind("on-add", self.on_add1)
|
||||
plist.add(track)
|
||||
self.assertIsNone(self.cb_plist)
|
||||
self.assertEqual(self.cb_track, track)
|
||||
|
|
|
@ -0,0 +1,45 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
from . import notify
|
||||
import unittest
|
||||
|
||||
class TestNotify(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.test_count = 0
|
||||
self.test_arg1 = None
|
||||
self.test_arg2 = None
|
||||
|
||||
def on_test1(self, arg1, arg2):
|
||||
self.test_count += 1
|
||||
def on_test2(self, arg1, arg2):
|
||||
self.test_arg1 = arg1
|
||||
def on_test3(self, arg1, arg2):
|
||||
self.test_arg2 = arg2
|
||||
|
||||
def test_notify(self):
|
||||
self.assertRaises(NotImplementedError, notify.Notify)
|
||||
self.assertEqual(notify.Notify.notifications, {})
|
||||
notify.Notify.notify_me("on-test", self.on_test1)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ self.on_test1 ]})
|
||||
notify.Notify.notify_me("on-test", self.on_test1)
|
||||
notify.Notify.notify_me("on-test", self.on_test2)
|
||||
notify.Notify.notify_me("on-test", self.on_test3)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ self.on_test1, self.on_test2, self.on_test3 ]})
|
||||
|
||||
notify.Notify.notify("on-test", "It Worked", "CoolCoolCool")
|
||||
self.assertEqual(self.test_count, 1)
|
||||
self.assertEqual(self.test_arg1, "It Worked")
|
||||
self.assertEqual(self.test_arg2, "CoolCoolCool")
|
||||
|
||||
notify.Notify.notify("no-cb", "Please", "Don't", "Crash")
|
||||
|
||||
notify.Notify.notify_cancel("on-test", self.on_test2)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ self.on_test1, self.on_test3 ]})
|
||||
notify.Notify.notify_cancel("on-test", self.on_test1)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ self.on_test3 ]})
|
||||
notify.Notify.notify_cancel("on-test", self.on_test3)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ ]})
|
||||
notify.Notify.notify_cancel("on-test", self.on_test1)
|
||||
self.assertEqual(notify.Notify.notifications, {"on-test" : [ ]})
|
||||
|
||||
notify.Notify.clear()
|
||||
self.assertEqual(notify.Notify.notifications, { })
|
|
@ -6,6 +6,7 @@ class TestEmmental(unittest.TestCase):
|
|||
def test_import_curds(self):
|
||||
self.assertEqual(curds.Album, curds.tags.Album)
|
||||
self.assertEqual(curds.DataFile, curds.data.DataFile)
|
||||
self.assertEqual(curds.Notify, curds.notify.Notify)
|
||||
self.assertEqual(curds.Playlist, curds.playlist.playlist.Playlist)
|
||||
self.assertEqual(curds.ThreadQueue, curds.threadqueue.ThreadQueue)
|
||||
self.assertEqual(curds.Track, curds.tags.Track)
|
||||
|
|
Loading…
Reference in New Issue