curds: Add function for album lookups

We want to share album objects whenever possible, so add a lookup
function for checking this.  Python built-in objects (like dictionaries)
are supposedly threadsafe, so we don't need a lock to access them but I
added the threadpool test anyway just in case.

Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
Anna Schumaker 2019-01-28 16:05:11 -05:00
parent 3b42ca233e
commit 306fa0f40c
2 changed files with 38 additions and 0 deletions

View File

@ -1,4 +1,7 @@
# Copyright 2019 (c) Anna Schumaker.
import hashlib
album_map = dict()
class Album:
def __init__(self, fileinfo):
@ -9,3 +12,15 @@ class Album:
self.albumartist = fileinfo.get("albumartist",
fileinfo.get("album artist",
fileinfo.get("artist", [ "Unknown Artist" ])))[0]
def hash(self):
md5 = hashlib.md5()
md5.update(self.album.encode('utf-8'))
md5.update(self.albumartist.encode('utf-8'))
md5.update(str(self.date).encode('utf-8'))
return md5.hexdigest()
def lookup(fileinfo):
album = Album(fileinfo)
return album_map.setdefault(album.hash(), album)

View File

@ -1,4 +1,6 @@
# Copyright 2019 (c) Anna Schumaker
import hashlib
from multiprocessing.pool import ThreadPool
import unittest
import album
@ -13,6 +15,7 @@ class TestAlbumClass(unittest.TestCase):
self.assertEqual(a.date, 2019)
self.assertEqual(a.albumartist, "Test Artist")
self.assertEqual(a.tracktotal, 1)
self.assertEqual(a.hash(), hashlib.md5("Test AlbumTest Artist2019".encode('utf-8')).hexdigest())
def test_init_empty(self):
a = album.Album({})
@ -21,6 +24,7 @@ class TestAlbumClass(unittest.TestCase):
self.assertEqual(a.date, 0)
self.assertEqual(a.albumartist, "Unknown Artist")
self.assertEqual(a.tracktotal, 0)
self.assertEqual(a.hash(), hashlib.md5("Unknown AlbumUnknown Artist0".encode('utf-8')).hexdigest())
def test_init_artist_fallback(self):
test_info = {"albumartist" : [ "1" ], "album artist" : [ "2" ], "artist" : [ "3" ]}
@ -32,5 +36,24 @@ class TestAlbumClass(unittest.TestCase):
test_info.pop("artist")
self.assertEqual(album.Album(test_info).albumartist, "Unknown Artist")
def test_album_lookup(self):
album.album_map.clear()
a = album.lookup(album_info)
self.assertIsNotNone(a)
for i in range(10):
b = album.lookup(album_info)
self.assertEqual(a, b)
self.assertEqual(len(album.album_map), 1)
def test_parallel_lookup(self):
album.album_map.clear()
with ThreadPool(processes=5) as pool:
res = pool.map(album.lookup, [ album_info ] * 20)
self.assertIsNotNone(res[0])
self.assertEqual(res.count(res[0]), 20)
self.assertEqual(len(album.album_map), 1)
if __name__ == '__main__':
unittest.main()