trackdb: Add a TrackAllocator() class

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>

trackdb/allocator.py

@@ -0,0 +1,42 @@
+# Copyright 2021 (c) Anna Schumaker.
+from lib import publisher
+from . import track
+import threading
+
+class TrackAllocator:
+    def __init__(self):
+        self.tracks = dict()
+        self.nextid = 0
+        self.lock = threading.Lock()
+        self.Added = publisher.Publisher()
+        self.Removed = publisher.Publisher()
+
+    def __alloc_track__(self, lib, filepath):
+        with self.lock:
+            trak = track.Track(self.nextid, filepath, lib)
+            self.tracks[self.nextid] = trak
+            self.nextid += 1
+        self.Added.publish(trak)
+        return trak
+
+    def __getitem__(self, id):
+        with self.lock:
+            return self.tracks.get(id, None)
+
+    def allocate(self, lib, filepath):
+        try:
+            return self.__alloc_track__(lib, filepath)
+        except Exception as e:
+            pass
+
+    def remove(self, track):
+        with self.lock:
+            del self.tracks[track.trackid]
+        self.Removed.publish(track)
+
+    def reset(self):
+        with self.lock:
+            self.nextid = 0
+            self.tracks.clear()
+            self.Added.reset()
+            self.Removed.reset()

trackdb/test_allocator.py

@@ -0,0 +1,60 @@
+# Copyright 2021 (c) Anna Schumaker.
+from lib import publisher
+from . import allocator
+from . import library
+import pathlib
+import threading
+import unittest
+
+test_tracks = pathlib.Path("./trier/Test Album")
+
+class TestTrackAllocator(unittest.TestCase):
+    def setUp(self):
+        self.added = None
+        self.removed = None
+
+    def on_track_added(self, track):
+        self.added = track
+
+    def on_track_removed(self, track):
+        self.removed = track
+
+    def test_allocator_init(self):
+        alloc = allocator.TrackAllocator()
+        self.assertEqual(alloc.tracks, { })
+        self.assertEqual(alloc.nextid, 0)
+        self.assertIsInstance(alloc.lock, type(threading.Lock()))
+        self.assertIsInstance(alloc.Added, publisher.Publisher)
+        self.assertIsInstance(alloc.Removed, publisher.Publisher)
+
+    def test_allocator(self):
+        lib = library.LibraryPath(test_tracks)
+        alloc = allocator.TrackAllocator()
+        alloc.Added.register(self.on_track_added)
+        alloc.Removed.register(self.on_track_removed)
+
+        track = alloc.allocate(lib, test_tracks / "01 - Test Track.ogg")
+        self.assertEqual(alloc.tracks[0], track)
+        self.assertEqual(alloc[0], track)
+        self.assertEqual(alloc.nextid, 1)
+        self.assertEqual(track.trackid, 0)
+        self.assertEqual(self.added, track)
+
+        track2 = alloc.allocate(lib, test_tracks / "02 - Test {Disc 2}.ogg")
+        self.assertEqual(alloc.tracks[1], track2)
+        self.assertEqual(alloc[1], track2)
+
+        self.assertIsNone(alloc.allocate(lib, test_tracks / "No Such File"))
+        self.assertEqual(self.added, track2)
+
+        alloc.remove(track)
+        self.assertEqual(alloc[0], None)
+        self.assertEqual(alloc.tracks, { 1 : track2 })
+        self.assertEqual(alloc.nextid, 2)
+        self.assertEqual(self.removed, track)
+
+        alloc.reset()
+        self.assertEqual(alloc.nextid, 0)
+        self.assertEqual(alloc.tracks, { })
+        self.assertEqual(alloc.Added.subscribers, set())
+        self.assertEqual(alloc.Removed.subscribers, set())