anna/emmental
anna
/
emmental
1
0
Fork 0
Code Issues 13 Pull Requests Projects 1 Releases 12 Wiki Activity

curds: Declassify notifications 

Keeping this as a class that can never be instantiated seems like
overkill when we could just use plain functions.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2019-03-27 09:49:40 -04:00
parent 89a4daedf3
commit 918f76663c
21 changed files with 82 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
curds/__init__.py
View File

@ -8,7 +8,6 @@ from . import threadqueue
Album = tags.Album
DataFile = data.DataFile
LibraryPlaylist = playlist.library.LibraryPlaylist
Notify = notify.Notify
Playlist = playlist.playlist.Playlist
ThreadQueue = threadqueue.ThreadQueue
Track = tags.Track

58
curds/notify.py
View File

@ -1,35 +1,31 @@
# Copyright 2019 (c) Anna Schumaker.
class Notify:
notifications = {}
queue = [ ]
registered = { }
queued = [ ]
def __init__(self):
raise NotImplementedError
def notify(name, *args):
for (func, queue) in Notify.notifications.get(name, []):
if queue == True:
Notify.queue.append((func, args))
else:
func(*args)
def cancel(name, func, queue=False):
cb = Notify.notifications.get(name, [])
if (func, queue) in cb:
cb.remove((func, queue))
if len(cb) == 0:
Notify.notifications.pop(name)
def register(name, func, queue=False):
cb = Notify.notifications.setdefault(name, [])
if func in [ n[0] for n in cb ]:
return
cb.append((func, queue))
def run_queued():
if len(Notify.queue) > 0:
(func, args) = Notify.queue.pop(0)
def notify(name, *args):
for (func, queue) in registered.get(name, []):
if queue == True:
queued.append((func, args))
else:
func(*args)
return True
return False
def cancel(name, func, queue=False):
cb = registered.get(name, [])
if (func, queue) in cb:
cb.remove((func, queue))
if len(cb) == 0:
registered.pop(name)
def register(name, func, queue=False):
cb = registered.setdefault(name, [])
if func in [ n[0] for n in cb ]:
return
cb.append((func, queue))
def run_queued():
if len(queued) > 0:
(func, args) = queued.pop(0)
func(*args)
return True
return False

2
curds/playlist/collection.py
View File

@ -5,7 +5,7 @@ from .. import notify
class CollectionPlaylist(playlist.Playlist):
def __init__(self):
playlist.Playlist.__init__(self, "Collection", "media-playback-start")
notify.Notify.register("new-track", self.add)
notify.register("new-track", self.add)
self.loop = True
def set_loop(self, loop):

2
curds/playlist/manager.py
View File

@ -21,7 +21,7 @@ class LibraryManager(list):
plist = library.LibraryPlaylist(path)
self.append(plist)
self.sort()
notify.Notify.notify("new-library-playlist", plist)
notify.notify("new-library-playlist", plist)
return plist

2
curds/playlist/playlist.py
View File

@ -38,7 +38,7 @@ class Playlist():
def add(self, track):
if track is not None and track not in self:
self.list.append(track)
notify.Notify.notify("add-track", self, track, self.list.index(track))
notify.notify("add-track", self, track, self.list.index(track))
def next(self):
max = len(self) - 1

2
curds/playlist/previous.py
View File

@ -10,7 +10,7 @@ class PreviousPlaylist(playlist.Playlist):
if track is not None and track not in self:
self.list.insert(0, track)
self.current = 0
notify.Notify.notify("add-track", self, track, 0)
notify.notify("add-track", self, track, 0)
def set_loop(self, loop):
return playlist.Playlist.set_loop(self, False)

2
curds/playlist/test_collection.py
View File

@ -26,4 +26,4 @@ class TestCollectionPlaylist(unittest.TestCase):
self.assertTrue(plist.set_loop(False))
self.assertTrue(plist.loop)
notify.Notify.cancel("new-track", plist.add)
notify.cancel("new-track", plist.add)

6
curds/playlist/test_manager.py
View File

@ -17,14 +17,14 @@ class TestPlaylistManager(unittest.TestCase):
def setUp(self):
tags.clear()
library.reset()
notify.Notify.register("new-library-playlist", self.on_new_library)
notify.register("new-library-playlist", self.on_new_library)
self.playman = manager.PlaylistManager()
self.cb_plist = None
def tearDown(self):
notify.Notify.cancel("new-track", self.playman["Collection"].add)
notify.Notify.cancel("new-library-playlist", self.on_new_library)
notify.cancel("new-track", self.playman["Collection"].add)
notify.cancel("new-library-playlist", self.on_new_library)
library.stop()
def on_new_library(self, plist):

4
curds/playlist/test_playlist.py
View File

@ -11,13 +11,13 @@ test_library = os.path.abspath("./trier/Test Library")
class TestPlaylist(unittest.TestCase):
def setUp(self):
tags.clear()
notify.Notify.register("add-track", self.on_add)
notify.register("add-track", self.on_add)
self.cb_plist = None
self.cb_track = None
self.cb_index = None
def tearDown(self):
notify.Notify.cancel("add-track", self.on_add)
notify.cancel("add-track", self.on_add)
def on_add(self, plist, track, index):
self.cb_plist = plist

4
curds/playlist/test_previous.py
View File

@ -10,13 +10,13 @@ test_album = os.path.abspath("./trier/Test Library/Test Artist 01/Test Album 1")
class TestPreviousPlaylist(unittest.TestCase):
def setUp(self):
notify.Notify.register("add-track", self.on_add_track)
notify.register("add-track", self.on_add_track)
self.cb_plist = None
self.cb_track = None
self.cb_index = 0
def tearDown(self):
notify.Notify.cancel("add-track", self.on_add_track)
notify.cancel("add-track", self.on_add_track)
def on_add_track(self, plist, track, index):
if isinstance(plist, previous.PreviousPlaylist):

2
curds/tags.py
View File

@ -19,7 +19,7 @@ class Tag:
def lookup(tag, cb):
ret = tag_map.setdefault(hash(tag), tag)
if ret == tag:
notify.Notify.notify(cb, ret)
notify.notify(cb, ret)
return ret

54
curds/test_notify.py
View File

@ -4,8 +4,8 @@ import unittest
class TestNotify(unittest.TestCase):
def setUp(self):
notify.Notify.notifications.clear()
notify.Notify.queue.clear()
notify.registered.clear()
notify.queued.clear()
self.test_count = 0
self.test_arg1 = None
self.test_arg2 = None
@ -18,37 +18,37 @@ class TestNotify(unittest.TestCase):
self.test_arg2 = arg2
def test_notify(self):
self.assertRaises(NotImplementedError, notify.Notify)
self.assertEqual(notify.Notify.notifications, {})
self.assertEqual(notify.Notify.queue, [])
self.assertEqual(notify.registered, {})
self.assertEqual(notify.queued, [])
notify.Notify.register("on-test", self.on_test1)
self.assertEqual(notify.Notify.notifications, {"on-test" : [ (self.on_test1, False) ]})
notify.Notify.register("on-test", self.on_test1, True)
notify.Notify.register("on-test", self.on_test2)
notify.Notify.register("on-test", self.on_test3, True)
self.assertEqual(notify.Notify.notifications, {"on-test" : [ (self.on_test1, False),
(self.on_test2, False),
(self.on_test3, True ) ]})
notify.register("on-test", self.on_test1)
self.assertEqual(notify.registered, {"on-test" : [ (self.on_test1, False) ]})
notify.register("on-test", self.on_test1, True)
notify.register("on-test", self.on_test2)
notify.register("on-test", self.on_test3, True)
self.assertEqual(notify.registered, {"on-test" : [ (self.on_test1, False),
(self.on_test2, False),
(self.on_test3, True ) ]})
notify.Notify.notify("on-test", "It Worked", "CoolCoolCool")
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, None)
self.assertEqual(notify.Notify.queue, [ (self.on_test3, ("It Worked", "CoolCoolCool")) ])
self.assertTrue(notify.Notify.run_queued())
self.assertEqual(notify.queued, [ (self.on_test3, ("It Worked", "CoolCoolCool")) ])
self.assertTrue(notify.run_queued())
self.assertEqual(self.test_arg2, "CoolCoolCool")
self.assertEqual(notify.queued, [ ])
notify.Notify.notify("no-cb", "Please", "Don't", "Crash")
self.assertFalse(notify.Notify.run_queued())
notify.notify("no-cb", "Please", "Don't", "Crash")
self.assertFalse(notify.run_queued())
notify.Notify.cancel("on-test", self.on_test2)
self.assertEqual(notify.Notify.notifications, {"on-test" : [ (self.on_test1, False),
(self.on_test3, True) ]})
notify.Notify.cancel("on-test", self.on_test1)
self.assertEqual(notify.Notify.notifications, {"on-test" : [ (self.on_test3, True) ]})
notify.Notify.cancel("on-test", self.on_test3, True)
self.assertEqual(notify.Notify.notifications, { })
notify.Notify.cancel("on-test", self.on_test1)
self.assertEqual(notify.Notify.notifications, { })
notify.cancel("on-test", self.on_test2)
self.assertEqual(notify.registered, {"on-test" : [ (self.on_test1, False),
(self.on_test3, True) ]})
notify.cancel("on-test", self.on_test1)
self.assertEqual(notify.registered, {"on-test" : [ (self.on_test3, True) ]})
notify.cancel("on-test", self.on_test3, True)
self.assertEqual(notify.registered, { })
notify.cancel("on-test", self.on_test1)
self.assertEqual(notify.registered, { })

4
curds/test_tags_album.py
View File

@ -9,11 +9,11 @@ album_info = {"album" : [ "Test Album" ], "albumartist" : [ "Test Artist" ],
class TestAlbumTag(unittest.TestCase):
def setUp(self):
tags.clear()
notify.Notify.register("new-album", self.on_new_album)
notify.register("new-album", self.on_new_album)
self.cb_album = None
def tearDown(self):
notify.Notify.cancel("new-album", self.on_new_album)
notify.cancel("new-album", self.on_new_album)
def on_new_album(self, album):
self.cb_album = album

4
curds/test_tags_track.py
View File

@ -9,11 +9,11 @@ test_tracks = os.path.abspath("./trier/Test Album")
class TestTrackTag(unittest.TestCase):
def setUp(self):
tags.clear()
notify.Notify.register("new-track", self.on_new_track)
notify.register("new-track", self.on_new_track)
self.cb_track = None
def tearDown(self):
notify.Notify.cancel("new-track", self.on_new_track)
notify.cancel("new-track", self.on_new_track)
def on_new_track(self, track):
self.cb_track = track

2
rind/gst.py
View File

@ -120,7 +120,7 @@ class EmmentalAudio:
if state == Gst.State.READY:
Title.set_markup("<big>Emmental</big>")
Artist.set_markup("<big>The Cheesy Music Player</big>")
curds.Notify.notify("state-changed", state)
curds.notify.notify("state-changed", state)
def update_progress(self):
(p_m, p_s) = divmod(int(self.position()), 60)

2
rind/gtk.py
View File

@ -28,7 +28,7 @@ class EmmentalApplication(Gtk.Application):
self.idle_id = GLib.idle_add(self.on_idle)
def on_idle(self):
curds.Notify.run_queued()
curds.notify.run_queued()
return GLib.SOURCE_CONTINUE
Application = EmmentalApplication()

4
rind/manager.py
View File

@ -9,8 +9,8 @@ toplevel = [ "Collection", "Previous", None, "Library" ]
class ManagerModel(GObject.GObject, Gtk.TreeModel):
def __init__(self, *args, **kwargs):
GObject.GObject.__init__(self)
curds.Notify.register("add-track", self.on_add_track, queue=True)
curds.Notify.register("new-library-playlist", self.on_new_library, queue=True)
curds.notify.register("add-track", self.on_add_track, queue=True)
curds.notify.register("new-library-playlist", self.on_new_library, queue=True)
def can_select_path(self, selection, model, path, current):
plist = self.iter_playlist(self.get_iter(path))

2
rind/playlist.py
View File

@ -10,7 +10,7 @@ class PlaylistModel(GObject.GObject, Gtk.TreeModel):
def __init__(self, *args, **kwargs):
GObject.GObject.__init__(self)
self.playlist = curds.PlaylistManager["Collection"]
curds.Notify.register("add-track", self.on_add_track, queue=True)
curds.notify.register("add-track", self.on_add_track, queue=True)
RandomButton.connect("toggled", self.on_random_toggled)
def do_get_column_type(self, col):

4
rind/test_gst.py
View File

@ -13,11 +13,11 @@ class TestGst(unittest.TestCase):
def setUp(self):
curds.PlaylistManager["Collection"].list.clear()
curds.PlaylistManager["Library"].clear()
curds.Notify.register("new-track", curds.PlaylistManager["Collection"].add)
curds.notify.register("new-track", curds.PlaylistManager["Collection"].add)
self.audio = gst.EmmentalAudio()
self.state = None
curds.Notify.register("state-changed", self.on_state_changed)
curds.notify.register("state-changed", self.on_state_changed)
def setUpClass():
gst.Audio.disconnect()

6
rind/test_manager.py
View File

@ -19,9 +19,9 @@ class TestManager(unittest.TestCase):
curds.PlaylistManager["Collection"].list.clear()
curds.PlaylistManager["Previous"].list.clear()
curds.PlaylistManager["Library"].clear()
curds.Notify.register("new-track", curds.PlaylistManager["Collection"].add)
curds.Notify.register("add-track", manager.MgrModel.on_add_track)
curds.Notify.register("new-library-playlist", manager.MgrModel.on_new_library)
curds.notify.register("new-track", curds.PlaylistManager["Collection"].add)
curds.notify.register("add-track", manager.MgrModel.on_add_track)
curds.notify.register("new-library-playlist", manager.MgrModel.on_new_library)
curds.playlist.library.reset()
def tearDown(self):

2
test_emmental.py
View File

@ -8,11 +8,11 @@ class TestEmmental(unittest.TestCase):
self.assertEqual(curds.Album, curds.tags.Album)
self.assertEqual(curds.DataFile, curds.data.DataFile)
self.assertEqual(curds.LibraryPlaylist, curds.playlist.library.LibraryPlaylist)
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)
self.assertIsNotNone(curds.notify)
self.assertIsInstance(curds.PlaylistManager, curds.playlist.manager.PlaylistManager)
def test_import_rind(self):