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

Remove old curds/ directory 

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2021-07-26 10:24:15 -04:00
parent e248fd3658
commit 08d866b047
32 changed files with 0 additions and 2659 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
curds/__init__.py
View File

@ -1,26 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import data
from . import notify
from . import playlist
from . import threadqueue
import os
import trackdb
trackdb.load()
playlist.load()
if os.environ.get("EMMENTAL_TESTING"):
playlist.reset()
DataFile = data.DataFile
Playlist = playlist.playlist.Playlist
ThreadQueue = threadqueue.ThreadQueue
Track = trackdb.track.Track
def reset():
notify.clear()
trackdb.reset()
playlist.reset()
def stop():
playlist.library.stop()

52
curds/data.py
View File

@ -1,52 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
import os
import pickle
import xdg.BaseDirectory
__resource = "emmental"
if os.environ.get("EMMENTAL_TESTING"):
__resource = "emmental-testing"
READ = 'rb'
WRITE = 'wb'
emmental_data = xdg.BaseDirectory.save_data_path(__resource)
class DataFile:
def __init__(self, path, mode):
self.path = os.path.join(emmental_data, path)
self.temp = os.path.join(emmental_data, f".{path}.tmp")
self.mode = mode
self.file = None
def __enter__(self):
if self.mode == WRITE:
self.file = open(self.temp, self.mode)
elif self.mode == READ and self.exists():
self.file = open(self.path, self.mode)
return self
def __exit__(self, exp_type, exp_value, traceback):
if self.file:
self.file.close()
self.file = None
if self.mode == WRITE:
os.rename(self.temp, self.path)
return True
def exists(self):
return os.path.exists(self.path)
def pickle(self, obj):
if self.file:
pickle.dump(obj, self.file, pickle.HIGHEST_PROTOCOL)
def remove(self):
if not self.file and self.exists():
os.remove(self.path)
return True
return False
def unpickle(self):
if self.file:
return pickle.load(self.file)

45
curds/notify.py
View File

@ -1,45 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
import threading
from gi.repository import GLib
events = { }
idle_queue = [ ]
event_lock = threading.Lock()
def cancel(name, func):
cb = events.get(name, [])
for event in cb:
if event[0] == func:
cb.remove(event)
if len(cb) == 0:
events.pop(name)
def clear():
events.clear()
idle_queue.clear()
def notify(name, *args):
for (func, idle) in events.get(name, []):
if idle == True:
with event_lock:
if (func, args) not in idle_queue:
idle_queue.append((func, args))
if len(idle_queue) == 1:
GLib.idle_add(notify_idle)
else:
func(*args)
def notify_idle():
with event_lock:
if len(idle_queue) == 0:
return False
(func, args) = idle_queue.pop(0)
more = len(idle_queue) > 0
func(*args)
return more
def register(name, func, idle=False):
cb = events.setdefault(name, [])
if func in [ n[0] for n in cb ]:
return
cb.append((func, idle))

116
curds/playlist/__init__.py
View File

@ -1,116 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import library
from . import playlist
from . import root
from .. import data
from .. import notify
import random
import threading
import trackdb
Current = [ ]
Library = library.LibraryPlaylist
Lock = threading.Lock()
Root = None
Track = None
def current():
with Lock:
return Current[0]
def lookup(name):
with Lock:
return Root.lookup(name)
def next():
global Track
with Lock:
orig = Current[0]
Track = orig.next()
while Track == None and len(Current) > 1:
Current.pop(0)
Track = Current[0].next()
if len(Current[0]) == 0 and len(Current) > 1:
Current.pop(0)
if orig != Current[0]:
orig.changed()
Current[0].changed()
Root.lookup("Previous").add(Track)
return Track
def peek(n):
res = [ ]
with Lock:
state = random.getstate()
for pl in Current:
tracks = pl.peek(n)
res += tracks
if (n := n - len(tracks)) == 0:
break
random.setstate(state)
return res
def previous():
global Track
with Lock:
Track = Root.lookup("Previous").next()
return Track
def select(plist):
with Lock:
orig = Current[0]
if plist == Root.lookup("Collection"):
Current.clear()
elif plist == Root.lookup("Previous"):
return
elif plist in Current:
Current.remove(plist)
Current.insert(0, plist)
orig.changed()
plist.changed()
def save():
with Lock:
with data.DataFile("playlists.pickle", data.WRITE) as f:
trackid = Track.trackid if Track != None else None
f.pickle([ Root, trackid, Current ])
def load():
global Current
global Root
global Track
with data.DataFile("playlists.pickle", data.READ) as f:
if f.exists():
[ Root, trackid, Current ] = f.unpickle()
Track = trackdb.get_track(trackid)
if Root == None:
Root = root.PlaylistRoot()
Track = None
Current = [ Root.lookup("Collection") ]
init()
def reset():
global Current
global Track
Root.reset()
Track = None
Current = [ Root.lookup("Collection") ]
init()
def init():
global Starred
global UpNext
Starred = Root.lookup("Playlists").lookup("Starred")
UpNext = Root.lookup("Up Next")
notify.register("save-playlists", save, idle=True)
notify.register("save-data", save)

29
curds/playlist/artist.py
View File

@ -1,29 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import tree
ALBUM_ICON = "media-optical-cd-audio"
ALBUM_SORT = [ "discnumber", "tracknumber" ]
ARTIST_ICON = "avatar-default-symbolic"
ARTIST_SORT = [ "album", "discnumber", "tracknumber" ]
class ArtistPlaylist(playlist.Playlist):
def alloc_child(self, name):
return playlist.Playlist(name, ALBUM_ICON, ALBUM_SORT)
def add(self, track):
playlist.Playlist.add(self, track)
self.lookup(track["album"]).add(track)
class ArtistNode(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "Artists", ARTIST_ICON)
def alloc_child(self, name):
return ArtistPlaylist(name, ARTIST_ICON, ARTIST_SORT)
def new_track(self, track, lib):
self.lookup(track["artist"]).add(track)

30
curds/playlist/decade.py
View File

@ -1,30 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import tree
DECADE_ICON = "x-office-calendar"
DECADE_SORT = [ "date", "artist", "album", "discnumber", "tracknumber" ]
YEAR_ICON = "x-office-calendar"
YEAR_SORT = [ "artist", "album", "discnumber", "tracknumber" ]
class DecadePlaylist(playlist.Playlist):
def alloc_child(self, name):
return playlist.Playlist(name, YEAR_ICON, YEAR_SORT)
def add(self, track):
playlist.Playlist.add(self, track)
self.lookup(str(track["year"])).add(track)
class DecadeNode(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "Decades", DECADE_ICON)
def alloc_child(self, name):
return DecadePlaylist(name, DECADE_ICON, DECADE_SORT)
def new_track(self, track, lib):
decade = str(track.decade()) + "s"
self.lookup(decade).add(track)

16
curds/playlist/genre.py
View File

@ -1,16 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import tree
GENRE_ICON = "emblem-generic"
GENRE_SORT = [ "artist", "date", "album", "discnumber", "tracknumber" ]
class GenreNode(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "Genres", GENRE_ICON)
def alloc_child(self, name):
return playlist.Playlist(name, GENRE_ICON, GENRE_SORT)
def new_track(self, track, lib):
self.lookup(track.tags["genre"][0]).add(track)

43
curds/playlist/library.py
View File

@ -1,43 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import notify
from .. import threadqueue
from .. import tree
import os
import trackdb
LIBRARY_ICON = "folder-music"
LIBRARY_SORT = [ "artist", "date", "album", "discnumber", "tracknumber" ]
class LibraryPlaylist(playlist.Playlist):
def __init__(self, path):
playlist.Playlist.__init__(self, path, icon=LIBRARY_ICON, sort=LIBRARY_SORT)
self.lib = trackdb.add_path(path)
def scan(self):
self.lib.scan()
def new_track(self, track, lib):
if lib == self.lib:
self.add(track)
class LibraryNode(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "Libraries", LIBRARY_ICON)
def alloc_child(self, path):
return LibraryPlaylist(path)
def lookup(self, path):
library = tree.ETree.lookup(self, os.path.abspath(path))
library.scan()
return library
def join():
for lib in trackdb.library_paths:
lib.join()
def stop():
pass

204
curds/playlist/playlist.py
View File

@ -1,204 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from .. import notify
from .. import sort
from .. import tree
import random
import trackdb
import threading
from gi.repository import GLib
class Playlist(tree.ETree):
def __init__(self, name, icon="", sort=[], can_loop=True, can_random=True):
self.list = [ ]
self.sort_order = sort
self.current = -1
self.can_loop = can_loop
self.loop = False
self.can_random = can_random
self.random = False
self.visible = False
self.plist_lock = threading.Lock()
tree.ETree.__init__(self, name, icon)
def __find_track__(self, track):
if track == None:
return (False, None)
if len(self.sort_order) == 0:
try:
return (True, self.list.index(track))
except:
return (False, len(self.list))
key = self.track_key(track)
(index, obj) = sort.bisect(self.list, key, self.track_key)
return (obj == track, index)
def __getitem__(self, i):
with self.plist_lock:
return self.list[i] if i < len(self.list) else None
def __getstate__(self):
with self.plist_lock:
state = tree.ETree.__getstate__(self)
state["list"] = [ t.trackid for t in self.list ]
del state["plist_lock"]
return state
def __len__(self):
with self.plist_lock:
return len(self.list)
def __next__(self, cur):
if (max := len(self.list) - 1) == -1:
return None
elif max == 0:
return None if (self.loop == False and cur == 0) else 0
elif self.random == True:
return (cur + random.randint(1, max)) % max
elif cur < max:
return cur + 1
elif self.loop == True:
return 0
return None
def __setstate__(self, state):
tree.ETree.__setstate__(self, state)
self.list = [ trackdb.get_track(id) for id in state["list"] ]
self.sort_order = state["sort_order"]
self.current = state["current"]
self.can_loop = state["can_loop"]
self.loop = state["loop"]
self.can_random = state["can_random"]
self.random = state["random"]
self.visible = False
self.plist_lock = threading.Lock()
def add(self, track):
with self.plist_lock:
(has, index) = self.__find_track__(track)
if has == True or index == None:
return
self.list[index:index] = [ track ]
self.track_added(track, index)
def changed(self):
notify.notify("playlist-changed", self)
notify.notify("save-playlists")
def contains(self, track):
with self.plist_lock:
return self.__find_track__(track)[0]
def get_markup(self):
t = GLib.markup_escape_text(self.name)
l = len(self.list)
return f"{t}\n{l} Track{'s' if l != 1 else ''}"
def index(self, track):
with self.plist_lock:
(has, index) = self.__find_track__(track)
return index if has == True else None
def move(self, track, new):
with self.plist_lock:
(has, old) = self.__find_track__(track)
if has == False:
return
self.list.pop(old)
self.list.insert(new, track)
self.sort_order.clear()
if old == self.current:
self.current = new
elif new < old and new <= self.current:
self.current += 1
elif new > old and new == self.current:
self.current += 1
elif new > old and new > self.current:
self.current -= 1
def next(self):
with self.plist_lock:
if (n := self.__next__(self.current)) != None:
self.current = n
return self.list[self.current]
return None
def peek(self, n):
with self.plist_lock:
cur = self.current
res = [ ]
for i in range(n):
if (cur := self.__next__(cur)) == None:
break
res.append(self.list[cur])
return res
def remove(self, track):
with self.plist_lock:
(has, pos) = self.__find_track__(track)
if has == False:
return
track = self.list.pop(pos)
if pos <= self.current:
self.current -= 1
self.track_removed(track, pos)
def reset(self):
with self.plist_lock:
self.list.clear()
self.current = -1
self.loop = False
self.random = False
tree.ETree.reset(self)
def runtime(self):
with self.plist_lock:
total = sum([ track.tags["length"] for track in self.list ])
m, s = divmod(total, 60)
h, m = divmod(m, 60)
d, h = divmod(h, 24)
res = [ ]
if d > 0: res.append(f"{int(d)} day{'s' if d != 1 else ''}")
if h > 0: res.append(f"{int(h)} hour{'s' if h != 1 else ''}")
if m > 0: res.append(f"{int(m)} minute{'s' if m != 1 else ''}")
if s > 0: res.append(f"{int(s)} second{'s' if s != 1 else ''}")
return ", ".join(res)
def set_loop(self, loop):
if self.can_loop == True:
self.loop = loop
return self.loop
def set_random(self, random):
if self.can_random == True:
self.random = random
return self.random
def sort(self, field):
with self.plist_lock:
if field in self.sort_order:
self.sort_order.remove(field)
else:
self.sort_order.append(field)
self.list.sort(key=self.track_key)
return field in self.sort_order
def track_added(self, track, index):
if self.visible == True:
notify.notify("add-track", self, track, index)
self.changed()
def track_key(self, track):
ret = [ ]
for field in self.sort_order:
val = track[field]
if val != None:
ret.append(sort.key(val))
return ret
def track_removed(self, track, index):
if self.visible == True:
notify.notify("remove-track", self, track, index)
self.changed()

35
curds/playlist/root.py
View File

@ -1,35 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import artist
from . import decade
from . import genre
from . import library
from . import special
from . import user
from .. import sort
from .. import tree
class PlaylistRoot(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "root")
with self.tree_lock:
self.__insert__(0, special.CollectionPlaylist())
self.__insert__(1, special.UpNextPlaylist())
self.__insert__(2, special.PreviousPlaylist())
self.__insert__(3, user.UserNode())
self.__insert__(4, artist.ArtistNode())
self.__insert__(5, genre.GenreNode())
self.__insert__(6, decade.DecadeNode())
self.__insert__(7, library.LibraryNode())
def lookup(self, name):
with self.tree_lock:
for node in self.children:
if node.sort_key() == sort.key(name):
return node
return None
def reset(self):
for plist in self.children:
plist.reset()

64
curds/playlist/special.py
View File

@ -1,64 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import notify
COLL_ICON = "media-playback-start"
COLL_SORT = [ "artist", "date", "album", "discnumber", "tracknumber" ]
NEXT_ICON = "edit-redo"
NEXT_SORT = [ ]
PREV_ICON = "edit-undo"
PREV_SORT = [ ]
class CollectionPlaylist(playlist.Playlist):
def __init__(self):
playlist.Playlist.__init__(self, "Collection", COLL_ICON, COLL_SORT, can_loop=False)
self.loop = True
def new_track(self, track, lib):
self.add(track)
def reset(self):
playlist.Playlist.reset(self)
self.loop = True
class PreviousPlaylist(playlist.Playlist):
def __init__(self):
playlist.Playlist.__init__(self, "Previous", PREV_ICON, PREV_SORT,
can_loop=False, can_random=False)
def __getstate__(self):
state = playlist.Playlist.__getstate__(self)
state["list"] = [ ]
state["current"] = -1
return state
def add(self, track):
if track is not None:
self.remove(track)
with self.plist_lock:
self.list.insert(0, track)
self.current = 0
self.track_added(track, 0)
def sort(self, field):
return False
class UpNextPlaylist(playlist.Playlist):
def __init__(self):
playlist.Playlist.__init__(self, "Up Next", NEXT_ICON, NEXT_SORT, can_loop=False)
def next(self):
track = playlist.Playlist.next(self)
self.remove(track)
if len(self) == 0:
self.reset()
return track
def track_removed(self, track, index):
notify.notify("remove-track", self, track, index)
self.changed()

70
curds/playlist/test_artist.py
View File

@ -1,70 +0,0 @@
# Copyright 2019 (c) Anna Schumaker
from . import artist
from . import playlist
from .. import notify
from .. import tree
import os
import pathlib
import trackdb
import unittest
test_album = pathlib.Path("./trier/Test Library/Test Artist 01/Test Album 1")
class TestArtistPlaylist(unittest.TestCase):
def setUp(self):
notify.clear()
trackdb.reset()
self.lib = trackdb.add_path(test_album)
def test_artist_node(self):
anode = artist.ArtistNode()
self.assertIsInstance(anode, tree.ETree)
self.assertEqual(anode.name, "Artists")
self.assertEqual(anode.icon, "avatar-default-symbolic")
def test_artist_alloc_child(self):
anode = artist.ArtistNode()
plist = anode.alloc_child("Test Artist")
self.assertIsInstance(plist, artist.ArtistPlaylist)
self.assertEqual(plist.name, "Test Artist")
self.assertEqual(plist.icon, "avatar-default-symbolic")
self.assertEqual(plist.sort_order, [ "album", "discnumber", "tracknumber" ])
def test_artist_node_new_track(self):
anode = artist.ArtistNode()
self.assertEqual(anode.n_children(), 0)
track = self.lib.add_track("01 - Test Track 01.ogg")
self.assertEqual(anode.n_children(), 1)
plist = anode.nth_child(0)
self.assertEqual(plist.name, "Test Artist 01")
self.assertEqual(plist.icon, "avatar-default-symbolic")
self.assertEqual(plist[0], track)
def test_artist_node_reset(self):
anode = artist.ArtistNode()
anode.children.append(42)
trackdb.library.TrackAdded.unregister(anode.new_track)
anode.reset()
self.assertEqual(len(anode.children), 0)
self.assertIn(anode.new_track, trackdb.library.TrackAdded.subscribers)
def test_artist_playlist_alloc(self):
anode = artist.ArtistPlaylist("Test Playlist", artist.ARTIST_ICON)
album = anode.alloc_child("Test Album")
self.assertIsInstance(album, playlist.Playlist)
self.assertEqual(album.name, "Test Album")
self.assertEqual(album.icon, "media-optical-cd-audio")
self.assertEqual(album.sort_order, [ "discnumber", "tracknumber" ])
def test_artist_playlist_add(self):
anode = artist.ArtistNode()
track = self.lib.add_track("01 - Test Track 01.ogg")
plist = anode.nth_child(0)
self.assertEqual(plist.n_children(), 1)
album = plist.nth_child(0)
self.assertIsInstance(album, playlist.Playlist)
self.assertEqual(album.name, "Test Album 1")

53
curds/playlist/test_collection.py
View File

@ -1,53 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from . import special
from .. import notify
import os
import pathlib
import trackdb
import unittest
test_album = pathlib.Path("./trier/Test Album")
class TestCollectionPlaylist(unittest.TestCase):
def setUp(self):
trackdb.reset()
self.lib = trackdb.add_path(test_album)
def test_collection_plist(self):
plist = special.CollectionPlaylist()
self.assertIsInstance(plist, special.CollectionPlaylist)
self.assertIsInstance(plist, playlist.Playlist)
self.assertEqual(plist.name, "Collection")
self.assertEqual(plist.icon, "media-playback-start")
self.assertEqual(plist.sort_order,
[ "artist", "date", "album", "discnumber", "tracknumber" ])
def test_collection_new_track(self):
plist = special.CollectionPlaylist()
self.assertEqual(len(plist), 0)
track1 = self.lib.add_track("01 - Test Track.ogg")
self.assertEqual(len(plist), 1)
self.assertEqual(plist[0], track1)
def test_collection_reset(self):
plist = special.CollectionPlaylist()
plist.list.append(42)
plist.random = True
plist.loop = False
trackdb.library.TrackAdded.unregister(plist.new_track)
plist.reset()
self.assertEqual(len(plist), 0)
self.assertFalse(plist.random)
self.assertTrue(plist.loop)
self.assertIn(plist.new_track, trackdb.library.TrackAdded.subscribers)
def test_collection_loop(self):
plist = special.CollectionPlaylist()
self.assertFalse(plist.can_loop)
self.assertTrue( plist.loop)
self.assertTrue( plist.set_loop(False))
self.assertTrue( plist.loop)

72
curds/playlist/test_decade.py
View File

@ -1,72 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import decade
from . import playlist
from .. import notify
from .. import tree
import os
import pathlib
import trackdb
import unittest
test_library = pathlib.Path("./trier/Test Library/Test Artist 01")
class TestDecadePlaylist(unittest.TestCase):
def setUp(self):
notify.clear()
trackdb.reset()
self.lib = trackdb.add_path(test_library)
def test_decade_node(self):
dnode = decade.DecadeNode()
self.assertIsInstance(dnode, tree.ETree)
self.assertEqual(dnode.name, "Decades")
self.assertEqual(dnode.icon, "x-office-calendar")
def test_decade_alloc_child(self):
dnode = decade.DecadeNode()
plist = dnode.alloc_child("1980s")
self.assertIsInstance(plist, decade.DecadePlaylist)
self.assertEqual(plist.name, "1980s")
self.assertEqual(plist.icon, "x-office-calendar")
self.assertEqual(plist.sort_order,
[ "date", "artist", "album", "discnumber", "tracknumber"])
def test_decade_node_new_track(self):
dnode = decade.DecadeNode()
self.assertEqual(dnode.n_children(), 0)
track = self.lib.add_track("Test Album 1", "01 - Test Track 01.ogg")
self.assertEqual(dnode.n_children(), 1)
plist = dnode.nth_child(0)
self.assertEqual(plist.name, "1970s")
self.assertEqual(plist.icon, "x-office-calendar")
self.assertEqual(plist[0], track)
def test_decade_node_reset(self):
dnode = decade.DecadeNode()
dnode.children.append(42)
trackdb.library.TrackAdded.unregister(dnode.new_track)
dnode.reset()
self.assertEqual(len(dnode.children), 0)
self.assertIn(dnode.new_track, trackdb.library.TrackAdded.subscribers)
def test_decade_playlist_alloc(self):
dnode = decade.DecadePlaylist("Test Decade", decade.DECADE_ICON)
year = dnode.alloc_child("1973")
self.assertIsInstance(year, playlist.Playlist)
self.assertEqual(year.name, "1973")
self.assertEqual(year.icon, "x-office-calendar")
self.assertEqual(year.sort_order,
[ "artist", "album", "discnumber", "tracknumber" ])
def test_decade_playlist_add(self):
dnode = decade.DecadeNode()
track = self.lib.add_track("Test Album 1", "01 - Test Track 01.ogg")
plist = dnode.nth_child(0)
self.assertEqual(plist.n_children(), 1)
year = plist.nth_child(0)
self.assertIsInstance(year, playlist.Playlist)
self.assertEqual(year.name, "1973")

70
curds/playlist/test_genre.py
View File

@ -1,70 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import genre
from . import playlist
from .. import notify
from .. import tree
import os
import pathlib
import trackdb
import unittest
test_library = pathlib.Path("./trier/Test Library/Test Artist 01")
class TestGenrePlaylist(unittest.TestCase):
def setUp(self):
notify.clear()
trackdb.reset()
self.lib = trackdb.add_path(test_library)
def test_genre_node(self):
gnode = genre.GenreNode()
self.assertIsInstance(gnode, tree.ETree)
self.assertEqual(gnode.name, "Genres")
self.assertEqual(gnode.icon, "emblem-generic")
def test_genre_playlist(self):
gnode = genre.GenreNode()
track = self.lib.add_track("Test Album 1", "01 - Test Track 01.ogg")
self.assertEqual(gnode.n_children(), 1)
plist = gnode.nth_child(0)
self.assertEqual(plist.name, "Test Genre 1")
self.assertEqual(plist.icon, "emblem-generic")
self.assertEqual(plist[0], track)
def test_genre_alloc_child(self):
gnode = genre.GenreNode()
plist = gnode.alloc_child("Test Genre")
self.assertIsInstance(plist, playlist.Playlist)
self.assertEqual(plist.name, "Test Genre")
self.assertEqual(plist.icon, "emblem-generic")
self.assertEqual(plist.sort_order,
[ "artist", "date", "album", "discnumber", "tracknumber" ])
def test_genre_new_tracks(self):
gnode = genre.GenreNode()
track1 = self.lib.add_track("Test Album 1", "01 - Test Track 01.ogg")
track2 = self.lib.add_track("Test Album 2", "02 - Test Track 02.ogg")
track3 = self.lib.add_track("Test Album 2", "03 - Test Track 03.ogg")
self.assertEqual(gnode.n_children(), 2)
plist = gnode.nth_child(0)
self.assertEqual(plist.name, "Test Genre 1")
self.assertEqual(len(plist), 1)
self.assertEqual(plist[0], track1)
plist = gnode.nth_child(1)
self.assertEqual(plist.name, "Test Genre 2")
self.assertEqual(len(plist), 2)
self.assertEqual(plist[0], track2)
self.assertEqual(plist[1], track3)
def test_genre_reset(self):
gnode = genre.GenreNode()
plist = playlist.Playlist("Test Playlist")
gnode.insert_child(plist)
trackdb.library.TrackAdded.unregister(gnode.new_track)
gnode.reset()
self.assertEqual(gnode.n_children(), 0)
self.assertIn(gnode.new_track, trackdb.library.TrackAdded.subscribers)

55
curds/playlist/test_library.py
View File

@ -1,55 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import library
from . import playlist
from .. import notify
from .. import tree
import lib
import os
import time
import trackdb
import unittest
test_library = os.path.abspath("./trier/Test Library")
test_album = os.path.abspath("./trier/Test Album")
class TestLibraryPlaylist(unittest.TestCase):
def setUp(self):
notify.clear()
trackdb.reset()
def tearDownClass():
library.stop()
trackdb.reset()
def test_library_node(self):
lnode = library.LibraryNode()
self.assertIsInstance(lnode, tree.ETree)
self.assertEqual(lnode.name, "Libraries")
self.assertEqual(lnode.icon, "folder-music")
def test_library_alloc_child(self):
lnode = library.LibraryNode()
plist = lnode.alloc_child(test_library)
self.assertIsInstance(plist, library.LibraryPlaylist)
self.assertEqual(plist.name, test_library)
self.assertEqual(plist.icon, "folder-music")
self.assertEqual(plist.sort_order,
[ "artist", "date", "album", "discnumber", "tracknumber"])
self.assertIsInstance(plist.lib, trackdb.library.LibraryPath)
library.join()
self.assertEqual(len(plist), 0)
def test_library_lookup(self):
lnode = library.LibraryNode()
plist = lnode.lookup(test_library)
self.assertEqual(lnode.n_children(), 1)
self.assertIsInstance(plist, library.LibraryPlaylist)
library.join()
self.assertEqual(len(plist), 1250)
self.assertEqual(lnode.lookup(test_library + "/"), plist)
plist = lnode.lookup(test_album)
self.assertEqual(lnode.n_children(), 2)
library.join()
self.assertEqual(len(plist), 12)

43
curds/playlist/test_new.py
View File

@ -1,43 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from . import user
import os
import pathlib
import trackdb
import unittest
test_album = pathlib.Path("./trier/Test Album/")
class TestNewPlaylist(unittest.TestCase):
def setUp(self):
trackdb.reset()
self.lib = trackdb.add_path(test_album)
def test_new_plist(self):
plist = user.NewPlaylist()
self.assertIsInstance(plist, playlist.Playlist)
self.assertEqual(plist.name, "New Tracks")
self.assertEqual(plist.icon, "document-new")
self.assertEqual(plist.sort_order,
[ "artist", "date", "album", "discnumber", "tracknumber" ])
def test_new_plist_new_track(self):
plist = user.NewPlaylist()
self.assertEqual(len(plist), 0)
track = self.lib.add_track("01 - Test Track.ogg")
self.assertEqual(len(plist), 1)
self.assertEqual(plist[0], track)
def test_new_plist_get_state(self):
plist = user.NewPlaylist()
track = self.lib.add_track("01 - Test Track.ogg")
plist.add(track)
plist.current = 3
state = plist.__getstate__()
self.assertEqual(state["name"], "New Tracks")
self.assertEqual(state["icon"], "document-new")
self.assertEqual(state["list"], [ ])
self.assertEqual(state["current"], -1)

399
curds/playlist/test_playlist.py
View File

@ -1,399 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import notify
from .. import tree
import os
import pathlib
import random
import trackdb
import unittest
test_library = pathlib.Path("./trier/Test Library")
test_album = test_library / "Test Artist 01" / "Test Album 1"
class TestPlaylist(unittest.TestCase):
def setUp(self):
trackdb.reset()
notify.register("add-track", self.on_add_remove)
notify.register("remove-track", self.on_add_remove)
notify.register("playlist-changed", self.on_plist_changed)
self.cb_plist = None
self.cb_index = None
self.cb_changed = None
self.plist = playlist.Playlist("Test Playlist", sort=[])
self.lib = trackdb.add_path(test_album)
def tearDown(self):
notify.cancel("add-track", self.on_add_remove)
notify.cancel("remove-track", self.on_add_remove)
notify.cancel("playlist-changed", self.on_plist_changed)
def add_tracks(self, tracknos):
tracks = [ ]
for t in tracknos:
track = self.lib.add_track(f"{t:02} - Test Track {t:02}.ogg")
tracks.append(track)
self.plist.add(track)
return tracks
def on_add_remove(self, plist, track, index):
self.cb_plist = plist
self.cb_track = track
self.cb_index = index
def on_plist_changed(self, plist):
self.cb_changed = plist
def test_playlist__init__(self):
self.assertIsInstance(self.plist, tree.ETree)
self.assertEqual(self.plist.name, "Test Playlist")
self.assertEqual(self.plist.icon, "")
self.assertEqual(self.plist.list, [ ])
self.assertEqual(self.plist.sort_order, [ ])
self.assertEqual(self.plist.current, -1)
self.assertTrue( self.plist.can_loop)
self.assertFalse(self.plist.loop)
self.assertTrue( self.plist.can_random)
self.assertFalse(self.plist.random)
self.assertFalse(self.plist.visible)
self.assertFalse(self.plist.plist_lock.locked())
def test_playlist__getitem__(self):
self.assertIsNone(self.plist[0])
tracks = self.add_tracks([1, 2])
self.assertEqual(self.plist[0], tracks[0])
self.assertEqual(self.plist[1], tracks[1])
def test_playlist__getstate__(self):
tracks = self.add_tracks([1, 2])
state = self.plist.__getstate__()
self.assertEqual(state["name"], "Test Playlist")
self.assertEqual(state["icon"], "")
self.assertEqual(state["list"], [ tracks[0].trackid, tracks[1].trackid ])
self.assertNotIn("plist_lock", state)
def test_playlist__len__(self):
self.assertEqual(len(self.plist), 0)
self.add_tracks([1, 2])
self.assertEqual(len(self.plist), 2)
def test_playlist__setstate__(self):
tracks = self.add_tracks([1, 2])
state = { "name" : "Test", "icon" : "test", "children" : [ ],
"sibling" : None, "parent" : None,
"list" : [ tracks[0].trackid, tracks[1].trackid ],
"sort_order" : [ 1 ], "current" : 2, "can_loop" : False,
"loop" : True, "can_random" : False, "random" : True,
"visible" : True }
self.plist.reset()
self.plist.plist_lock = None
self.plist.__setstate__(state)
self.assertEqual(self.plist.name, "Test")
self.assertEqual(self.plist.icon, "test")
self.assertEqual(self.plist.list, tracks)
self.assertEqual(self.plist.sort_order, [ 1 ])
self.assertEqual(self.plist.current, 2)
self.assertFalse(self.plist.can_loop)
self.assertTrue( self.plist.loop)
self.assertFalse(self.plist.can_random)
self.assertTrue( self.plist.random)
self.assertFalse(self.plist.visible)
self.assertFalse(self.plist.plist_lock.locked())
def test_playlist_add_append(self):
tracks = self.add_tracks(range(1, 6))
self.assertEqual(len(self.plist), 5)
self.assertEqual(self.cb_changed, self.plist)
self.assertIsNone(self.cb_plist)
self.assertIsNone(self.cb_index)
for i, track in enumerate(tracks):
self.assertEqual(self.plist[i], track)
self.plist.visible = True
self.add_tracks([6])
self.assertEqual(len(self.plist), 6)
self.assertEqual(self.cb_plist, self.plist)
self.assertEqual(self.cb_index, 5)
self.plist.add(None)
self.assertEqual(len(self.plist), 6)
def test_playlist_add_sorted(self):
self.plist.sort("tracknumber")
tracknos = [ 5, 6, 4, 7, 3, 8, 2, 9, 1, 10 ]
tracks = self.add_tracks(tracknos)
self.assertEqual(len(self.plist), 10)
for i, track in enumerate(self.plist.list):
self.assertEqual(track.tags["tracknumber"], i + 1)
def test_playlist_contains(self):
tracks = self.add_tracks(range(9, 0, -1))
for track in tracks:
self.assertTrue(self.plist.contains(track))
track = self.lib.add_track(f"10 - Test Track 10.ogg")
self.assertFalse(self.plist.contains(track))
def test_playlist_get_markup(self):
self.assertEqual(self.plist.get_markup(), "Test Playlist\n0 Tracks")
self.add_tracks([1])
self.assertEqual(self.plist.get_markup(), "Test Playlist\n1 Track")
self.add_tracks([2])
self.assertEqual(self.plist.get_markup(), "Test Playlist\n2 Tracks")
self.plist.name = "Test & Playlist"
self.assertEqual(self.plist.get_markup(), "Test &amp; Playlist\n2 Tracks")
def test_playlist_index(self):
tracks = self.add_tracks(range(9, 0, -1))
for i, track in enumerate(tracks):
self.assertEqual(self.plist.index(track), i)
track = self.lib.add_track(f"10 - Test Track 10.ogg")
self.assertIsNone(self.plist.index(track))
self.assertIsNone(self.plist.index(None))
def test_playlist_index_sorted(self):
self.plist.sort("tracknumber")
tracks = self.add_tracks(range(1, 10))
for i, track in enumerate(tracks):
self.assertEqual(self.plist.index(track), i)
track = self.lib.add_track(f"10 - Test Track 10.ogg")
self.assertIsNone(self.plist.index(track))
self.assertIsNone(self.plist.index(None))
def test_playlist_move_up(self):
tracks = self.add_tracks(range(1, 11))
self.plist.current = 5
self.plist.sort("tracknumber")
self.plist.move(tracks[8], 6)
self.assertEqual(self.plist.current, 5)
self.assertEqual(self.plist.list[6], tracks[8])
self.assertEqual(self.plist.sort_order, [ ])
self.plist.move(tracks[8], 5)
self.assertEqual(self.plist.current, 6)
self.assertEqual(self.plist.list[5], tracks[8])
self.plist.move(tracks[9], 0)
self.assertEqual(self.plist.current, 7)
self.assertEqual(self.plist.list[0], tracks[9])
def test_playlist_move_down(self):
tracks = self.add_tracks(range(1, 11))
self.plist.current = 5
self.plist.move(tracks[1], 4)
self.assertEqual(self.plist.current, 5)
self.assertEqual(self.plist.list[4], tracks[1])
self.plist.move(tracks[1], 5)
self.assertEqual(self.plist.current, 6)
self.assertEqual(self.plist.list[5], tracks[1])
self.plist.move(tracks[0], 9)
self.assertEqual(self.plist.current, 5)
self.assertEqual(self.plist.list[9], tracks[0])
self.plist.move(42, 0)
def test_playlist_move_current(self):
tracks = self.add_tracks(range(1, 11))
self.plist.current = 5
self.plist.move(tracks[5], 8)
self.assertEqual(self.plist.current, 8)
self.assertEqual(self.plist.list[8], tracks[5])
self.plist.move(tracks[5], 0)
self.assertEqual(self.plist.current, 0)
self.assertEqual(self.plist.list[0], tracks[5])
def test_playlist_next_empty_one(self):
self.assertIsNone(self.plist.next())
self.assertEqual(self.plist.current, -1)
track = self.add_tracks([1])[0]
self.assertEqual(self.plist.current, -1)
self.assertEqual(self.plist.next(), track)
self.assertEqual(self.plist.current, 0)
self.assertEqual(self.plist.next(), None)
self.assertEqual(self.plist.current, 0)
self.plist.set_loop(True)
self.assertEqual(self.plist.next(), track)
self.assertEqual(self.plist.current, 0)
def test_playlist_next(self):
tracks = self.add_tracks(range(1, 11))
for i, t in enumerate(tracks):
track = self.plist.next()
self.assertEqual(track, t)
self.assertEqual(self.plist.current, i)
self.assertIsNone(self.plist.next())
self.assertEqual(self.plist.current, len(tracks) - 1)
def test_playlist_next_loop(self):
tracks = self.add_tracks(range(1, 11))
self.assertTrue(self.plist.set_loop(True))
for i, t in enumerate(tracks):
track = self.plist.next()
self.assertEqual(track, t)
self.assertEqual(self.plist.current, i)
for i, t in enumerate(tracks):
track = self.plist.next()
self.assertEqual(track, t)
self.assertEqual(self.plist.current, i)
def test_playlist_next_random(self):
current = -1
tracks = self.add_tracks(range(1, 11))
max = len(tracks) - 1
random.seed(1)
offsets = [ random.randint(1, max) for i in range(max * 2) ]
self.assertTrue(self.plist.set_random(True))
random.seed(1)
for offset in offsets:
current = (current + offset) % max
track = self.plist.next()
self.assertEqual(self.plist.current, current)
self.assertEqual(track, tracks[current])
def test_playlist_peek(self):
tracks = self.add_tracks(range(1, 5))
peek = self.plist.peek(0)
self.assertEqual(peek, [ ])
self.assertEqual(self.plist.current, -1)
peek = self.plist.peek(2)
self.assertEqual(peek, tracks[:2])
self.assertEqual(self.plist.current, -1)
peek = self.plist.peek(10)
self.assertEqual(peek, tracks)
self.assertEqual(self.plist.current, -1)
def test_playlist_remove(self):
tracks = self.add_tracks(range(1, 11))
self.plist.current = 5
self.cb_plist = None
self.cb_track = None
self.cb_index = None
self.plist.remove(tracks[8])
self.assertEqual(self.plist.current, 5)
self.assertEqual(len(self.plist), 9)
self.assertNotIn(tracks[8], self.plist.list)
self.assertEqual(self.cb_changed, self.plist)
self.assertIsNone(self.cb_plist)
self.assertIsNone(self.cb_track)
self.assertIsNone(self.cb_index)
self.plist.visible = True
self.plist.remove(tracks[5])
self.assertEqual(self.plist.current, 4)
self.assertEqual(len(self.plist), 8)
self.assertNotIn(tracks[5], self.plist.list)
self.assertEqual(self.cb_plist, self.plist)
self.assertEqual(self.cb_track, tracks[5])
self.assertEqual(self.cb_index, 5)
self.plist.remove(tracks[2])
self.plist.remove(tracks[2])
self.assertEqual(self.plist.current, 3)
self.assertEqual(len(self.plist), 7)
self.assertNotIn(tracks[2], self.plist.list)
def test_playlist_reset(self):
tracks = self.add_tracks(range(1, 11))
self.plist.current = 2
self.plist.loop = True
self.plist.random = True
self.plist.reset()
self.assertEqual(len(self.plist.list), 0)
self.assertEqual(self.plist.current, -1)
self.assertFalse(self.plist.loop)
self.assertFalse(self.plist.random)
def test_playlist_runtime(self):
track = self.add_tracks([1])[0]
self.assertEqual(self.plist.runtime(), "1 second")
track.tags["length"] = 2
self.assertEqual(self.plist.runtime(), "2 seconds")
track.tags["length"] = 60
self.assertEqual(self.plist.runtime(), "1 minute")
track.tags["length"] = 120
self.assertEqual(self.plist.runtime(), "2 minutes")
track.tags["length"] = 3600
self.assertEqual(self.plist.runtime(), "1 hour")
track.tags["length"] = 7200
self.assertEqual(self.plist.runtime(), "2 hours")
track.tags["length"] = 86400
self.assertEqual(self.plist.runtime(), "1 day")
track.tags["length"] = 172800
self.assertEqual(self.plist.runtime(), "2 days")
def test_playlist_set_loop(self):
self.assertTrue( self.plist.can_loop)
self.assertFalse(self.plist.loop)
self.assertTrue( self.plist.set_loop(True))
self.assertTrue( self.plist.loop)
self.assertFalse( self.plist.set_loop(False))
self.assertFalse(self.plist.loop)
self.plist = playlist.Playlist("Test Playlist", can_loop=False)
self.assertFalse(self.plist.can_loop)
self.assertFalse(self.plist.set_loop(True))
self.assertFalse(self.plist.loop)
def test_playlist_set_random(self):
self.assertTrue( self.plist.can_random)
self.assertFalse(self.plist.random)
self.assertTrue( self.plist.set_random(True))
self.assertTrue( self.plist.random)
self.assertFalse( self.plist.set_random(False))
self.assertFalse(self.plist.random)
self.plist = playlist.Playlist("Test Playlist", can_random=False)
self.assertFalse(self.plist.can_random)
self.assertFalse(self.plist.set_random(True))
self.assertFalse(self.plist.random)
def test_playlist_sort(self):
tracks = self.add_tracks(range(1, 11))
random.shuffle(self.plist.list)
self.assertEqual(self.plist.sort_order, [ ])
self.assertTrue( self.plist.sort("tracknumber"))
self.assertEqual(self.plist.sort_order, [ "tracknumber" ])
for i, track in enumerate(tracks):
self.assertEqual(self.plist[i], track)
self.assertTrue( self.plist.sort("artist"))
self.assertEqual(self.plist.sort_order, [ "tracknumber", "artist" ])
self.assertFalse(self.plist.sort("tracknumber"))
self.assertEqual(self.plist.sort_order, [ "artist" ])
self.assertFalse(self.plist.sort("artist"))
self.assertEqual(self.plist.sort_order, [ ])
def test_playlist_track_key(self):
track = self.add_tracks([1])[0]
trackno = track["tracknumber"]
artist = track["artist"].lower().split()
self.assertEqual(self.plist.track_key(track), [ ])
self.assertTrue( self.plist.sort("tracknumber"))
self.assertEqual(self.plist.track_key(track), [ [trackno] ])
self.assertTrue( self.plist.sort("artist"))
self.assertEqual(self.plist.track_key(track), [ [trackno], artist ])

108
curds/playlist/test_prev.py
View File

@ -1,108 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from . import special
from .. import notify
import os
import pathlib
import trackdb
import unittest
test_album = pathlib.Path("./trier/Test Library/Test Artist 01/Test Album 1")
class TestPreviousPlaylist(unittest.TestCase):
def setUp(self):
trackdb.reset()
notify.register("add-track", self.on_add_track)
self.cb_plist = None
self.cb_track = None
self.cb_index = 0
self.lib = trackdb.add_path(test_album)
def tearDown(self):
notify.cancel("add-track", self.on_add_track)
def on_add_track(self, plist, track, index):
if isinstance(plist, special.PreviousPlaylist):
self.cb_plist = plist
self.cb_track = track
self.cb_index = index
def test_previous_playlist(self):
plist = special.PreviousPlaylist()
self.assertIsInstance(plist, special.PreviousPlaylist)
self.assertIsInstance(plist, playlist.Playlist)
self.assertEqual(plist.name, "Previous")
self.assertEqual(plist.icon, "edit-undo")
def test_previous_getstate(self):
plist = special.PreviousPlaylist()
track = self.lib.add_track("01 - Test Track 01.ogg")
plist.add(track)
plist.current = 3
state = plist.__getstate__()
self.assertEqual(state["name"], "Previous")
self.assertEqual(state["icon"], "edit-undo")
self.assertEqual(state["list"], [ ])
self.assertEqual(state["current"], -1)
def test_previous_add(self):
plist = special.PreviousPlaylist()
plist.visible = True
track1 = self.lib.add_track("01 - Test Track 01.ogg")
track2 = self.lib.add_track("02 - Test Track 02.ogg")
track3 = self.lib.add_track("03 - Test Track 03.ogg")
self.assertEqual(plist.current, -1)
self.assertEqual(plist.next(), None)
plist.add(track1)
self.assertEqual(self.cb_plist, plist)
self.assertEqual(self.cb_track, track1)
self.assertEqual(self.cb_index, 0)
self.assertEqual(plist.current, 0)
self.assertEqual(plist.next(), None)
plist.add(track2)
self.assertEqual(self.cb_plist, plist)
self.assertEqual(self.cb_track, track2)
self.assertEqual(self.cb_index, 0)
self.assertEqual(plist.current, 0)
self.assertEqual(len(plist), 2)
self.assertEqual(plist[0], track2)
self.assertEqual(plist[1], track1)
self.assertEqual(plist.next(), track1)
self.assertEqual(plist.current, 1)
plist.add(track3)
self.assertEqual(plist.current, 0)
plist.add(track1)
self.assertEqual(len(plist), 3)
self.assertEqual(plist[0], track1)
self.assertEqual(plist[1], track3)
self.assertEqual(plist[2], track2)
def test_previous_loop(self):
plist = special.PreviousPlaylist()
self.assertFalse(plist.can_loop)
self.assertFalse(plist.loop)
self.assertFalse(plist.set_loop(True))
self.assertFalse(plist.loop)
def test_previous_random(self):
plist = special.PreviousPlaylist()
self.assertFalse(plist.can_random)
self.assertFalse(plist.random)
self.assertFalse(plist.set_random(True))
self.assertFalse(plist.random)
def test_previous_sort(self):
plist = special.PreviousPlaylist()
self.assertEqual(plist.sort_order, [ ])
self.assertFalse(plist.sort("date"))
self.assertEqual(plist.sort_order, [ ])

86
curds/playlist/test_root.py
View File

@ -1,86 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import artist
from . import decade
from . import genre
from . import library
from . import playlist
from . import root
from . import special
from . import user
from .. import notify
from .. import threadqueue
from .. import tree
import os
import random
import trackdb
import unittest
test_album = os.path.abspath("./trier/Test Album")
test_library = os.path.abspath("./trier/Test Library")
class TestPlaylistRoot(unittest.TestCase):
def setUp(self):
trackdb.reset()
notify.register("next-track", self.on_next_track)
notify.register("playlist-changed", self.on_playlist_changed)
try:
self.playman.reset()
except:
self.playman = root.PlaylistRoot()
self.cb_plist = None
self.cb_index = None
self.cb_first = None
self.cb_track = None
self.cb_change = [ ]
def tearDown(self):
trackdb.reset()
notify.cancel("next-track", self.on_next_track)
notify.register("playlist-changed", self.on_playlist_changed)
library.stop()
def on_playlist_changed(self, plist):
self.cb_change.append(plist)
def on_next_track(self, track):
self.cb_track = track
def test_root_init(self):
self.assertIsInstance(self.playman, root.PlaylistRoot)
self.assertIsInstance(self.playman, tree.ETree)
self.assertIsInstance(self.playman.lookup("Collection"), special.CollectionPlaylist)
self.assertIsInstance(self.playman.lookup("Up Next"), special.UpNextPlaylist)
self.assertIsInstance(self.playman.lookup("Previous"), special.PreviousPlaylist)
self.assertIsInstance(self.playman.lookup("Playlists"), user.UserNode)
self.assertIsInstance(self.playman.lookup("Artists"), artist.ArtistNode)
self.assertIsInstance(self.playman.lookup("Genres"), genre.GenreNode)
self.assertIsInstance(self.playman.lookup("Decades"), decade.DecadeNode)
self.assertIsInstance(self.playman.lookup("Libraries"), library.LibraryNode)
self.assertEqual(self.playman.nth_child(0), self.playman.lookup("Collection"))
self.assertEqual(self.playman.nth_child(1), self.playman.lookup("Up Next"))
self.assertEqual(self.playman.nth_child(2), self.playman.lookup("Previous"))
self.assertEqual(self.playman.nth_child(3), self.playman.lookup("Playlists"))
self.assertEqual(self.playman.nth_child(4), self.playman.lookup("Artists"))
self.assertEqual(self.playman.nth_child(5), self.playman.lookup("Genres"))
self.assertEqual(self.playman.nth_child(6), self.playman.lookup("Decades"))
self.assertEqual(self.playman.nth_child(7), self.playman.lookup("Libraries"))
for i, plist in enumerate(self.playman.children):
self.assertEqual(plist.sibling, self.playman.nth_child(i + 1))
self.assertEqual(self.playman.name, "root")
def test_root_on_scan(self):
plist = self.playman.lookup("Libraries").lookup(test_library)
alist = self.playman.lookup("Libraries").lookup(test_album)
library.join()
self.assertEqual(len(plist), 1250)
self.assertEqual(len(alist), 12)
self.assertEqual(len(self.playman.lookup("Collection")), len(alist) + len(plist))
self.playman.reset()
self.assertEqual(len(self.playman.lookup("Collection")), 0)

71
curds/playlist/test_up_next.py
View File

@ -1,71 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from .. import notify
from . import special
import os
import pathlib
import trackdb
import unittest
test_album = pathlib.Path("./trier/Test Library/Test Artist 01/Test Album 1")
class TestUpNextPlaylist(unittest.TestCase):
def setUp(self):
trackdb.reset()
notify.register("remove-track", self.on_remove)
self.cb_plist = None
self.cb_track = None
self.cb_index = None
self.lib = trackdb.add_path(test_album)
def tearDown(self):
notify.cancel("remove-track", self.on_remove)
def on_remove(self, plist, track, index):
self.cb_plist = plist
self.cb_track = track
self.cb_index = index
def add_tracks(self, plist):
tracks = [ ]
for i in range(1, 4,):
ret = self.lib.add_track(f"0{i} - Test Track 0{i}.ogg")
tracks.append(ret)
plist.add(ret)
return tracks
def test_up_next_playlist(self):
plist = special.UpNextPlaylist()
self.assertEqual(plist.name, "Up Next")
self.assertEqual(plist.icon, "edit-redo")
self.assertEqual(plist.sort_order, [ ])
self.assertFalse(plist.can_loop)
self.assertTrue( plist.can_random)
def test_up_next_next(self):
plist = special.UpNextPlaylist()
tracks = self.add_tracks(plist)
self.assertEqual(len(plist), 3)
track = plist.next()
self.assertEqual(len(plist), 2)
self.assertEqual(plist.current, -1)
self.assertEqual(self.cb_plist, plist)
self.assertEqual(self.cb_track, tracks[0])
self.assertEqual(self.cb_index, 0)
track = plist.next()
self.assertEqual(len(plist), 1)
self.assertEqual(plist.current, -1)
self.assertEqual(self.cb_plist, plist)
self.assertEqual(self.cb_track, tracks[1])
self.assertEqual(self.cb_index, 0)
plist.random = True
plist.sort_order = [ 42 ]
track = plist.next()
self.assertEqual(len(plist), 0)
self.assertEqual(plist.current, -1)
self.assertEqual(self.cb_plist, plist)
self.assertEqual(self.cb_track, tracks[2])
self.assertEqual(self.cb_index, 0)

45
curds/playlist/test_user.py
View File

@ -1,45 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import user
from .. import tree
import unittest
class TestUserPlaylists(unittest.TestCase):
def test_user_node(self):
unode = user.UserNode()
self.assertIsInstance(unode, tree.ETree)
self.assertEqual(unode.name, "Playlists")
self.assertEqual(unode.icon, "audio-x-generic")
self.assertEqual(unode.n_children(), 2)
self.assertEqual(unode.nth_child(0).name, "New Tracks")
self.assertEqual(unode.nth_child(0).icon, user.NEW_ICON)
self.assertEqual(unode.nth_child(1).name, "Starred")
self.assertEqual(unode.nth_child(1).icon, user.STAR_ICON)
self.assertIsInstance(unode.lookup("New Tracks"), user.NewPlaylist)
def test_user_reset(self):
unode = user.UserNode()
new = unode.lookup("New Tracks")
star = unode.lookup("Starred")
unode.reset()
self.assertEqual(unode.n_children(), 2)
self.assertNotEqual(unode.nth_child(0), new)
self.assertEqual(unode.nth_child(0).name, "New Tracks")
self.assertEqual(unode.nth_child(0).icon, user.NEW_ICON)
self.assertNotEqual(unode.nth_child(1), star)
self.assertEqual(unode.nth_child(1).name, "Starred")
self.assertEqual(unode.nth_child(1).icon, user.STAR_ICON)
def test_user_starred(self):
unode = user.UserNode()
star = unode.lookup("Starred")
self.assertEqual(star.name, "Starred")
self.assertEqual(star.icon, "starred")
self.assertEqual(star.sort_order,
[ "artist", "date", "album", "discnumber", "tracknumber" ])

35
curds/playlist/user.py
View File

@ -1,35 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import playlist
from .. import tree
NEW_ICON = "document-new"
NEW_SORT = [ "artist", "date", "album", "discnumber", "tracknumber" ]
STAR_ICON = "starred"
STAR_SORT = [ "artist", "date", "album", "discnumber", "tracknumber" ]
class NewPlaylist(playlist.Playlist):
def __init__(self):
playlist.Playlist.__init__(self, "New Tracks", NEW_ICON, NEW_SORT)
def __getstate__(self):
state = playlist.Playlist.__getstate__(self)
state["list"] = [ ]
state["current"] = -1
return state
def new_track(self, track, lib):
self.add(track)
class UserNode(tree.ETree):
def __init__(self):
tree.ETree.__init__(self, "Playlists", "audio-x-generic")
self.insert_child(NewPlaylist())
self.insert_child(playlist.Playlist("Starred", STAR_ICON, STAR_SORT))
def reset(self):
tree.ETree.reset(self)
self.insert_child(NewPlaylist())
self.insert_child(playlist.Playlist("Starred", STAR_ICON, STAR_SORT))

31
curds/sort.py
View File

@ -1,31 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
import unicodedata
import os
def bisect(lst, lhs, key_func):
begin = 0
end = len(lst)
while end - begin > 0:
pos = (end + begin) // 2
if lhs == (rhs := key_func(lst[pos])):
return (pos, lst[pos])
elif lhs < rhs:
end = pos
else:
begin = pos + 1
return (begin, None)
def key(text):
if os.path.exists(text):
return text.strip("/").split("/")
words = normalize(text).lower().split()
words = [ ''.join(filter(str.isalnum, w)) for w in words ]
words = [ w for w in words if w != "" ]
if len(words) > 0 and words[0] in [ "a", "the" ]:
return words[1:]
return words
def normalize(text):
decode = unicodedata.normalize("NFD", text)
return decode.encode("ascii", "ignore").decode("utf8")

70
curds/test_data.py
View File

@ -1,70 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import data
import os
import unittest
import xdg.BaseDirectory
xdg_data_home = xdg.BaseDirectory.xdg_data_home
testing_data = os.path.join(xdg_data_home, "emmental-testing")
testing_file = os.path.join(testing_data, "test.file")
testing_temp = os.path.join(testing_data, ".test.file.tmp")
class TestDataModule(unittest.TestCase):
def setUp(self):
if os.path.exists(testing_file): os.remove(testing_file)
def test_dir(self):
self.assertEqual(data.emmental_data, testing_data)
self.assertTrue(os.path.exists(testing_data))
self.assertTrue(os.path.isdir(testing_data))
self.assertEqual(data.READ, 'rb')
self.assertEqual(data.WRITE, 'wb')
def test_data_file_init(self):
f = data.DataFile("test.file", data.READ)
self.assertEqual(f.path, testing_file)
self.assertEqual(f.temp, testing_temp)
self.assertFalse(f.exists())
self.assertEqual(f.mode, data.READ)
self.assertIsNone(f.file)
f = data.DataFile("test.file", data.WRITE)
self.assertEqual(f.temp, testing_temp)
self.assertFalse(f.exists())
self.assertEqual(f.mode, data.WRITE)
self.assertIsNone(f.file)
def test_data_file_read_write(self):
test = [ "test", "saving", "a", "list" ]
with data.DataFile("test.file", data.READ) as f:
self.assertIsNone(f.file)
f.pickle(test)
self.assertIsNone(f.unpickle())
self.assertFalse(f.remove())
self.assertTrue(f.exists())
with data.DataFile("test.file", data.WRITE) as f:
self.assertIsNotNone(f.file)
self.assertEqual(f.file.name, testing_temp)
self.assertFalse(f.exists())
self.assertTrue(os.path.exists(testing_temp))
f.pickle(test)
self.assertFalse(f.remove())
self.assertTrue(f.exists())
self.assertIsNone(f.file)
self.assertFalse(os.path.exists(testing_temp))
self.assertTrue(f.exists())
with data.DataFile("test.file", data.READ) as f:
self.assertIsNotNone(f.file)
self.assertEqual(f.file.name, testing_file)
lst = f.unpickle()
self.assertEqual(test, lst)
self.assertFalse(f.remove())
self.assertTrue(f.exists())
f = data.DataFile("test.file", data.READ)
self.assertTrue(f.remove())
self.assertFalse(f.exists())
self.assertFalse(f.remove())

96
curds/test_notify.py
View File

@ -1,96 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import notify
import gi
import threading
import unittest
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
class TestNotify(unittest.TestCase):
def setUp(self):
notify.clear()
self.test_count = 0
self.test_arg1 = None
self.test_arg2 = None
self.task_queued = False
def tearDown(self):
notify.clear()
def notify_thread(self, event, *args):
thread = threading.Thread(target=notify.notify, args=(event, *args))
thread.start()
thread.join()
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 on_task_queued(self): self.task_queued = True
def on_test_queue(self, arg1): self.test_count += 1; self.test_arg1 = arg1
def test_notify_init(self):
self.assertEqual(notify.events, {})
self.assertEqual(notify.idle_queue, [])
self.assertFalse(notify.event_lock.locked())
def test_notify_register(self):
notify.register("on-test", self.on_test1)
self.assertEqual(notify.events, {"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.events, {"on-test" : [ (self.on_test1, False),
(self.on_test2, False),
(self.on_test3, True ) ]})
def test_notify_clear(self):
notify.events = { "abc" : [ (self.on_test1, False) ]}
notify.main_q = [ 1, 2, 3 ]
notify.clear()
self.assertEqual(notify.events, {})
self.assertEqual(notify.idle_queue, [])
def test_notify_cancel(self):
notify.events = { "on-test" : [ (self.on_test1, False),
(self.on_test2, False),
(self.on_test3, True) ]}
notify.cancel("on-test", self.on_test2)
self.assertEqual(notify.events, { "on-test" : [ (self.on_test1, False),
(self.on_test3, True) ]})
notify.cancel("on-test", self.on_test1)
self.assertEqual(notify.events, { "on-test" : [ (self.on_test3, True) ] })
notify.cancel("on-test", self.on_test3)
self.assertEqual(notify.events, { })
def test_notify_main_thread(self):
notify.register("on-test", self.on_test1)
notify.register("on-test", self.on_test2)
notify.register("on-test", self.on_test3, True)
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)
while Gtk.events_pending(): Gtk.main_iteration_do(True)
self.assertEqual(self.test_arg2, "CoolCoolCool")
def test_notify_bg_thread(self):
notify.register("on-test", self.on_test1, True)
notify.register("on-test", self.on_test2)
self.notify_thread("on-test", "It Worked", "CoolCoolCool")
self.assertEqual(self.test_count, 0)
self.assertEqual(self.test_arg1, "It Worked")
self.assertEqual(notify.idle_queue, [ (self.on_test1, ("It Worked", "CoolCoolCool")) ])
self.notify_thread("on-test", "It Worked", "CoolCoolCool")
self.assertEqual(self.test_count, 0)
self.assertEqual(notify.idle_queue, [ (self.on_test1, ("It Worked", "CoolCoolCool")) ])
while Gtk.events_pending(): Gtk.main_iteration_do(True)
self.assertEqual(self.test_count, 1)
self.assertEqual(notify.idle_queue, [ ])
self.assertFalse(notify.notify_idle())

196
curds/test_playlist.py
View File

@ -1,196 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import data
from . import notify
from . import playlist
import os
import pathlib
import random
import trackdb
import unittest
test_path = pathlib.Path("./trier/")
test_album = pathlib.Path("./trier/Test Album")
test_library = pathlib.Path("./trier/Test Library")
class TestPlaylist(unittest.TestCase):
def setUp(self):
data.DataFile("playlists.pickle", data.WRITE).remove()
trackdb.reset()
playlist.reset()
self.orig = playlist.Root
self.lib = trackdb.add_path(test_path)
def tearDown(self):
notify.clear()
playlist.Root = self.orig
playlist.Root.reset()
def test_current_shortcut(self):
self.assertEqual(playlist.current(), playlist.lookup("Collection"))
playlist.select(playlist.Starred)
self.assertEqual(playlist.current(), playlist.Starred)
def test_lookup_shortcut(self):
self.assertEqual(playlist.lookup("Collection"), playlist.Root.lookup("Collection"))
self.assertEqual(playlist.lookup("Playlists"), playlist.Root.lookup("Playlists"))
def test_starred_shortcut(self):
self.assertEqual(playlist.Starred, playlist.lookup("Playlists").lookup("Starred"))
playlist.Starred = None
playlist.reset()
self.assertEqual(playlist.Starred, playlist.lookup("Playlists").lookup("Starred"))
def test_up_next_shortcut(self):
self.assertEqual(playlist.UpNext, playlist.lookup("Up Next"))
playlist.UpNext = None
playlist.reset()
self.assertEqual(playlist.UpNext, playlist.lookup("Up Next"))
def test_next_prev(self):
self.assertIsNone(playlist.next())
self.assertIsNone(playlist.previous())
self.assertIsNone(playlist.Track)
clist = playlist.lookup("Collection")
prev = playlist.lookup("Previous")
upnxt = playlist.lookup("Up Next")
plist = playlist.lookup("Libraries").lookup(test_album)
playlist.library.join()
upnxt.add(clist[3])
playlist.Current.insert(0, upnxt)
track1 = playlist.next()
self.assertEqual(playlist.Track, track1)
self.assertEqual(playlist.Current, [ clist ])
self.assertEqual(len(prev), 1)
self.assertEqual(prev[0], track1)
track2 = playlist.next()
self.assertEqual(playlist.Current, [ clist ])
self.assertEqual(playlist.Track, track2)
self.assertEqual(clist.current, 0)
self.assertEqual(track2, clist[0])
self.assertEqual(len(prev), 2)
self.assertEqual(prev[0], track2)
self.assertEqual(prev[1], track1)
track3 = playlist.next()
self.assertEqual(playlist.Track, track3)
self.assertEqual(clist.current, 1)
self.assertEqual(track3, clist[1])
self.assertEqual(len(prev), 3)
self.assertEqual(prev[0], track3)
self.assertEqual(prev[1], track2)
self.assertEqual(prev[2], track1)
self.assertEqual(playlist.previous(), track2)
self.assertEqual(playlist.Track, track2)
playlist.reset()
self.assertEqual(len(prev), 0)
self.assertIsNone(playlist.Track)
def test_peek(self):
self.assertEqual(playlist.peek(3), [ ])
clist = playlist.lookup("Collection")
plist = playlist.lookup("Libraries").lookup(test_album)
playlist.library.join()
random.seed(1)
clist.set_random(False)
tracks = playlist.peek(0)
self.assertEqual(tracks, [ ])
self.assertEqual(clist.current, -1)
tracks = playlist.peek(5)
self.assertEqual(tracks, [ clist[i] for i in range(5) ])
self.assertEqual(clist.current, -1)
clist.set_random(True)
tracks = playlist.peek(5)
self.assertEqual(len(tracks), 5)
self.assertEqual(clist.current, -1)
for i in range(5):
self.assertEqual(playlist.next(), tracks[i])
def test_save_playlists(self):
dfile = data.DataFile("playlists.pickle", data.READ)
track = self.lib.add_track("01 - Test Track.ogg")
playlist.Root = None
playlist.Track = 1234
self.assertFalse(dfile.exists())
playlist.load()
self.assertIsInstance(playlist.Root, playlist.root.PlaylistRoot)
self.assertIsNone(playlist.Track)
self.assertEqual(playlist.Starred, playlist.lookup("Playlists").lookup("Starred"))
self.assertEqual(playlist.UpNext, playlist.lookup("Up Next"))
root = playlist.Root
playlist.Track = track
playlist.Current = [ playlist.lookup("Up Next"), playlist.lookup("Collection") ]
self.assertFalse(dfile.exists())
playlist.Starred.changed()
notify.notify_idle()
self.assertTrue(dfile.exists())
playlist.Current = [ ]
playlist.Root = None
playlist.Track = None
playlist.load()
self.assertIsNotNone(playlist.Root)
self.assertNotEqual(playlist.Root, root)
self.assertEqual(playlist.Track, track)
self.assertEqual(playlist.Current, [ playlist.lookup("Up Next"),
playlist.lookup("Collection") ])
self.assertEqual(playlist.Starred, playlist.lookup("Playlists").lookup("Starred"))
self.assertEqual(playlist.UpNext, playlist.lookup("Up Next"))
def test_select_playlist(self):
clist = playlist.lookup("Collection")
plist = playlist.lookup("Libraries").lookup(test_library)
playlist.library.join()
glist = playlist.lookup("Genres").lookup("Test Genre 1")
self.assertEqual(playlist.Current, [ clist ])
playlist.select(plist)
self.assertEqual(playlist.Current, [ plist, clist ])
playlist.select(glist)
self.assertEqual(playlist.Current, [ glist, plist, clist ])
playlist.select(plist)
self.assertEqual(playlist.Current, [ plist, glist, clist ])
playlist.select(playlist.lookup("Previous"))
self.assertEqual(playlist.Current, [ plist, glist, clist ])
plist.current = 1247
peek = playlist.peek(4)
self.assertEqual(len(peek), 4)
self.assertEqual(peek[0], plist[-2])
self.assertEqual(peek[1], plist[-1])
self.assertEqual(peek[2], glist[ 0])
self.assertEqual(peek[3], glist[ 1])
self.assertEqual(playlist.next(), plist[-2])
self.assertEqual(playlist.next(), plist[-1])
self.assertEqual(playlist.Current, [ plist, glist, clist ])
self.assertEqual(playlist.next(), glist[0])
self.assertEqual(playlist.Current, [ glist, clist ])
playlist.select(plist)
self.assertEqual(playlist.Current, [ plist, glist, clist ])
playlist.select(clist)
self.assertEqual(playlist.Current, [ clist ])
playlist.select(plist)
playlist.reset()
self.assertEqual(playlist.Current, [ clist ])

42
curds/test_sort.py
View File

@ -1,42 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import sort
import string
import unittest
class CharSort:
def __init__(self, char):
self.c = char
class TestSort(unittest.TestCase):
def key_func(self, obj):
return [ obj.c ]
def test_bisect(self):
chars = [ CharSort(c) for c in string.ascii_lowercase ]
self.assertEqual(sort.bisect([ ], "a", self.key_func), (0, None))
self.assertEqual(sort.bisect(chars, ["A"], self.key_func), (0, None))
self.assertEqual(sort.bisect(chars, ["|"], self.key_func), (26, None))
for i, c in enumerate(string.ascii_lowercase):
self.assertEqual(sort.bisect(chars, [c], self.key_func), (i, chars[i]))
def test_key(self):
self.assertEqual(sort.key("Test Key"), [ "test", "key" ])
self.assertEqual(sort.key("Test! Key!"), [ "test", "key" ])
self.assertEqual(sort.key("Test - Key"), [ "test", "key" ])
self.assertEqual(sort.key("The Test Key"), [ "test", "key" ])
self.assertEqual(sort.key("Test The Key"), [ "test", "the", "key" ])
self.assertEqual(sort.key("A Test Key"), [ "test", "key" ])
self.assertEqual(sort.key("Test A Key"), [ "test", "a", "key" ])
self.assertEqual(sort.key("Toäst Keäy"), [ "toast", "keay" ])
def test_key_path(self):
self.assertEqual(sort.key("/a/b/c/d/"), [ "abcd" ])
self.assertEqual(sort.key("trier/Test Album"), [ "trier", "Test Album" ])
self.assertEqual(sort.key("trier/Test Album/"), [ "trier", "Test Album" ])
def test_normalize(self):
self.assertEqual(sort.normalize("Test Text"), "Test Text")
self.assertEqual(sort.normalize("Toäst Keäy"), "Toast Keay")

49
curds/test_threadqueue.py
View File

@ -1,49 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import threadqueue
import queue
import unittest
import threading
test_a = 0
test_b = 0
test_c = 0
test_l = threading.Lock()
def set_abc(a, b, c):
global test_a, test_b, test_c
with test_l:
test_a = a
test_b = b
test_c = c
class TestThreadQueue(unittest.TestCase):
def setUp(self):
set_abc(0, 0, 0)
def test_threadqueue_init(self):
tq = threadqueue.ThreadQueue()
self.assertIsInstance(tq, queue.Queue)
self.assertIsInstance(tq, threading.Thread)
self.assertIsInstance(tq.stop_event, threading.Event)
self.assertTrue(tq.is_alive())
tq.stop()
self.assertFalse(tq.is_alive())
def test_threadqueue_push(self):
tq = threadqueue.ThreadQueue()
with test_l:
tq.push(set_abc, 1, 2, 3)
tq.join()
self.assertEqual(test_a, 1)
self.assertEqual(test_b, 2)
self.assertEqual(test_c, 3)
for i in range(1000):
tq.push(set_abc, i, i+1, i+2)
tq.join()
self.assertEqual(test_a, 999)
self.assertEqual(test_b, 1000)
self.assertEqual(test_c, 1001)
tq.stop()

240
curds/test_tree.py
View File

@ -1,240 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import notify
from . import tree
import lib
import threading
import unittest
class TestETree(unittest.TestCase):
def setUp(self):
notify.register("child-inserted", self.on_child_inserted)
notify.register("first-child", self.on_first_child)
def tearDown(self):
lib.idle.reset()
notify.cancel("child-inserted", self.on_child_inserted)
notify.cancel("first-child", self.on_first_child)
def on_child_inserted(self, child, path):
self.cb_child = child
self.cb_path = path
def on_first_child(self, parent, path):
self.cb_first = parent
self.cb_fpath = path
def test_etree_init(self):
etree = tree.ETree("Test Tree", "test-icon")
self.assertEqual(etree.name, "Test Tree")
self.assertEqual(etree.icon, "test-icon")
self.assertEqual( etree.children, [ ])
self.assertEqual( etree.child_ids, { })
self.assertFalse( etree.tree_lock.locked())
self.assertIsNone(etree.parent)
self.assertIsNone(etree.sibling)
def test_etree_init_empty(self):
etree = tree.ETree()
self.assertEqual(etree.icon, "")
self.assertEqual(etree.name, "")
def test_etree_getstate(self):
etree = tree.ETree("Test Tree", "test-icon")
state = etree.__getstate__()
self.assertEqual(state["icon"], "test-icon")
self.assertEqual(state["name"], "Test Tree")
self.assertEqual(state["children"], [ ])
self.assertEqual(state["parent"], None)
self.assertEqual(state["sibling"], None)
self.assertNotIn("child_ids", state)
self.assertNotIn("tree_lock", state)
def test_etree_setstate(self):
etree = tree.ETree()
ctree = tree.ETree()
state = { "icon" : "test-icon", "name" : "Test Tree",
"children" : [ ctree ], "parent" : 2, "sibling" : 3 }
etree.tree_lock = None
ctree.child_ids = { id(1) : 1 }
etree.__setstate__(state)
self.assertEqual(etree.icon, "test-icon")
self.assertEqual(etree.name, "Test Tree")
self.assertEqual(etree.children, [ ctree ])
self.assertEqual(etree.parent, 2)
self.assertEqual(etree.sibling, 3)
self.assertEqual(etree.child_ids, { id(1) : 1, id(ctree) : ctree })
self.assertFalse(etree.tree_lock.locked())
def test_etree_child_index(self):
root = tree.ETree()
a = root.lookup("A")
b = root.lookup("B")
c = a.lookup("C")
self.assertEqual( root.child_index(a), 0)
self.assertEqual( root.child_index(b), 1)
self.assertIsNone(root.child_index(c))
def test_etree_get_markup(self):
etree = tree.ETree("Test Tree", "test-icon")
self.assertEqual(etree.get_markup(), "<big>Test Tree</big>")
def test_etree_get_path(self):
root = tree.ETree()
a = root.lookup("A")
b = a.lookup("B")
c = b.lookup("C")
self.assertEqual(root.get_path(), [ ])
self.assertEqual(a.get_path(), [ 0 ])
self.assertEqual(b.get_path(), [ 0, 0 ])
self.assertEqual(c.get_path(), [ 0, 0, 0 ])
def test_etree_insert_child(self):
root = tree.ETree()
a = root.insert_child(tree.ETree("A", "icon"))
self.assertEqual(root.children, [ a ])
self.assertEqual(a.parent, root)
self.assertEqual(self.cb_child, a)
self.assertEqual(self.cb_path, [ 0 ])
self.assertEqual(self.cb_first, root)
self.assertEqual(self.cb_fpath, [ ])
self.cb_first = None
c = root.insert_child(tree.ETree("C", "icon"))
self.assertEqual(root.children, [ a, c ])
self.assertEqual(a.sibling, c)
self.assertEqual(self.cb_child, c)
self.assertEqual(self.cb_path, [ 1 ])
self.assertIsNone(self.cb_first)
b = root.insert_child(tree.ETree("B", "icon"))
self.assertEqual(root.children, [ a, b, c ])
self.assertEqual(a.sibling, b)
self.assertEqual(b.sibling, c)
self.assertEqual(self.cb_child, b)
self.assertEqual(self.cb_path, [ 1 ])
b = root.insert_child(b)
self.assertEqual(root.children, [ a, b, c ])
def test_etree_lookup(self):
root = tree.ETree()
b = root.lookup("B")
self.assertEqual(root.children, [ b ])
self.assertEqual(b.parent, root)
d = root.lookup("D")
self.assertEqual(root.children, [ b, d ])
self.assertEqual(b.sibling, d)
c = root.lookup("C")
self.assertEqual(root.children, [ b, c, d ])
self.assertEqual(b.sibling, c)
self.assertEqual(c.sibling, d)
a = root.lookup("A")
self.assertEqual(root.children, [ a, b, c, d ])
self.assertEqual(a.sibling, b)
self.assertEqual(root.lookup("A"), a)
self.assertEqual(root.lookup("B"), b)
self.assertEqual(root.lookup("C"), c)
self.assertEqual(root.lookup("D"), d)
self.assertEqual(root.children, [ a, b, c, d ])
self.assertEqual(root.child_ids, { id(a) : a, id(b) : b, id(c) : c, id(d) : d })
def test_etree_lookup_byid(self):
root = tree.ETree()
a = root.insert_child(tree.ETree("A", "icon"))
b = a.insert_child(tree.ETree("B", "icon"))
c = b.insert_child(tree.ETree("C", "icon"))
self.assertEqual(root.lookup_byid(id(a)), a)
self.assertEqual(root.lookup_byid(id(b)), b)
self.assertEqual(root.lookup_byid(id(c)), c)
self.assertEqual( a.lookup_byid(id(b)), b)
self.assertEqual( a.lookup_byid(id(c)), c)
self.assertEqual( b.lookup_byid(id(c)), c)
self.assertIsNone(c.lookup_byid(id(a)))
def test_etree_lookup_path(self):
root = tree.ETree()
a = root.lookup("A")
b = a.lookup("B")
c = b.lookup("C")
self.assertEqual( root.lookup_path([ ]), root)
self.assertEqual( root.lookup_path([ 0 ]), a)
self.assertIsNone(root.lookup_path([ 1 ]))
self.assertEqual( root.lookup_path([ 0, 0 ]), b)
self.assertIsNone(root.lookup_path([ 0, 1 ]))
self.assertIsNone(root.lookup_path([ 1, 0 ]))
self.assertEqual( root.lookup_path([ 0, 0, 0 ]), c)
self.assertIsNone(root.lookup_path([ 0, 0, 1 ]))
self.assertIsNone(root.lookup_path([ 0, 1, 0 ]))
self.assertIsNone(root.lookup_path([ 1, 0, 0 ]))
def test_etree_n_children(self):
root = tree.ETree()
self.assertEqual(root.n_children(), 0)
a = root.lookup("A")
self.assertEqual(root.n_children(), 1)
b = root.lookup("B")
self.assertEqual(root.n_children(), 2)
def test_etree_next_child(self):
root = tree.ETree()
a = root.lookup("A")
b = root.lookup("B")
self.assertIsNone(root.next_child())
self.assertEqual( a.next_child(), b)
self.assertIsNone(b.next_child())
def test_etree_nth_child(self):
root = tree.ETree()
a = root.lookup("A")
b = root.lookup("B")
self.assertIsNone(root.nth_child(-1))
self.assertEqual( root.nth_child(0), a)
self.assertEqual( root.nth_child(1), b)
self.assertIsNone(root.nth_child(2))
def test_etree_reset(self):
root = tree.ETree()
a = root.lookup("A")
b = root.lookup("B")
root.reset()
self.assertEqual(root.children, [ ])
def test_etree_sort_key(self):
etree = tree.ETree("Test Tree")
self.assertEqual(etree.sort_key(), [ "test", "tree" ])
etree.name = "Test - Tree!"
self.assertEqual(etree.sort_key(), [ "test", "tree" ])
def test_etree_walk(self):
root = tree.ETree()
a = root.lookup("A")
b = a.lookup("B")
c = a.lookup("C")
d = b.lookup("D")
vals = [ a, b, d, c ]
for i, n in enumerate(root.walk()):
self.assertEqual(n, vals[i])
self.assertEqual(i, 3)
for i, n in enumerate(a.walk()):
self.assertEqual(n, vals[i])
self.assertEqual(i, 3)

26
curds/threadqueue.py
View File

@ -1,26 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
import queue
import threading
class ThreadQueue(queue.Queue, threading.Thread):
def __init__(self):
queue.Queue.__init__(self)
threading.Thread.__init__(self)
self.stop_event = threading.Event()
self.start()
def push(self, func, *args):
self.put((func, args))
def run(self):
while not self.stop_event.is_set():
try:
(func, args) = self.get(block=True, timeout=0.1)
func(*args)
self.task_done()
except queue.Empty:
pass
def stop(self):
self.stop_event.set()
threading.Thread.join(self)

142
curds/tree.py
View File

@ -1,142 +0,0 @@
# Copyright 2019 (c) Anna Schumaker.
from . import notify
from . import sort
import threading
import trackdb
class ETree:
def __init__(self, name="", icon=""):
self.icon = icon
self.name = name
self.children = [ ]
self.child_ids = dict()
self.tree_lock = threading.Lock()
self.parent = None
self.sibling = None
self.__new_track__()
def __getstate__(self):
with self.tree_lock:
state = self.__dict__.copy()
del state["child_ids"]
del state["tree_lock"]
return state
def __insert__(self, index, child):
self.children.insert(index, child)
child.sibling = self.__nth_child__(index + 1)
child.parent = self
if (prev := self.__nth_child__(index - 1)) != None:
with prev.tree_lock:
prev.sibling = child
self.__map_child__(child)
if len(self.children) == 1:
notify.notify("first-child", self, self.get_path())
notify.notify("child-inserted", child, self.get_path() + [ index ])
return child
def __new_track__(self):
if self.new_track != ETree.new_track:
trackdb.library.TrackAdded.register(self.new_track)
def __nth_child__(self, n):
return self.children[n] if -1 < n < len(self.children) else None
def __map_child__(self, child):
self.child_ids[id(child)] = child
self.child_ids.update(child.child_ids)
if self.parent != None:
with self.parent.tree_lock:
self.parent.__map_child__(child)
def __setstate__(self, state):
self.parent = None
self.child_ids = dict()
self.tree_lock = threading.Lock()
self.icon = state["icon"]
self.name = state["name"]
self.children = state["children"]
for child in self.children:
self.__map_child__(child)
self.parent = state["parent"]
self.sibling = state["sibling"]
self.__new_track__()
def alloc_child(self, name):
return ETree(name)
def child_index(self, child):
with self.tree_lock:
if child.parent == self:
return self.children.index(child)
return None
def get_markup(self):
return f"<big>{self.name}</big>"
def get_path(self):
if self.parent == None:
return [ ]
return self.parent.get_path() + [ self.parent.child_index(self) ]
def insert_child(self, child):
with self.tree_lock:
(index, node) = sort.bisect(self.children, sort.key(child.name), self.sort_key)
if node == child:
return child
return self.__insert__(index, child)
def lookup(self, name):
with self.tree_lock:
(index, child) = sort.bisect(self.children, sort.key(name), self.sort_key)
if child != None:
return child
return self.__insert__(index, self.alloc_child(name))
def lookup_byid(self, id):
return self.child_ids.get(id, None)
def lookup_path(self, path):
if path == None:
return None
if len(path) == 0:
return self
if (child := self.nth_child(path[0])) and len(path) > 1:
child = child.lookup_path(path[1:])
return child
def n_children(self):
with self.tree_lock:
return len(self.children)
def new_track(self, track, lib):
pass
def next_child(self):
with self.tree_lock:
return self.sibling
def nth_child(self, n):
with self.tree_lock:
return self.__nth_child__(n)
def reset(self):
with self.tree_lock:
self.children.clear()
self.__new_track__()
def sort_key(self, node=None):
if node is None:
node = self
return sort.key(node.name)
def walk(self):
with self.tree_lock:
if self.parent != None:
yield self
for child in self.children:
yield from child.walk()