andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

audio: Convert audio_data_to_track to tags_to_track

Browse Source 
The new function only uses tags as input. In other words we now need to set
length, uri and mtime ourselves. Users of scan APIs have been updated.
This commit is contained in:
Thomas Adamcik 2014-12-15 23:05:15 +01:00
parent 4948dee4b9
commit 4f8244c499
4 changed files with 91 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
mopidy/audio/scan.py
View File

@ -132,9 +132,7 @@ def _artists(tags, artist_name, artist_id=None):
return [Artist(name=name) for name in tags[artist_name]]
def audio_data_to_track(data):
"""Convert taglist data + our extras to a track."""
tags = data['tags']
def tags_to_track(tags):
album_kwargs = {}
track_kwargs = {}
@ -169,13 +167,9 @@ def audio_data_to_track(data):
if tags.get(gst.TAG_DATE) and tags.get(gst.TAG_DATE)[0]:
track_kwargs['date'] = tags[gst.TAG_DATE][0].isoformat()
track_kwargs['last_modified'] = int(data.get('mtime') or 0)
track_kwargs['length'] = data.get('duration')
# Clear out any empty values we found
track_kwargs = {k: v for k, v in track_kwargs.items() if v}
album_kwargs = {k: v for k, v in album_kwargs.items() if v}
track_kwargs['uri'] = data['uri']
track_kwargs['album'] = Album(**album_kwargs)
return Track(**track_kwargs)

6
mopidy/local/commands.py
View File

@ -130,10 +130,8 @@ class ScanCommand(commands.Command):
tags, duration = scanner.scan(file_uri)
# TODO: reuse mtime from above...
mtime = os.path.getmtime(os.path.join(media_dir, relpath))
track = scan.audio_data_to_track({'uri': uri,
'tags': tags,
'duration': duration,
'mtime': mtime})
track = scan.tags_to_track(tags).copy(
uri=uri, length=duration, last_modified=mtime)
track = translator.add_musicbrainz_coverart_to_track(track)
library.add(track)
logger.debug('Added %s', track.uri)

3
mopidy/stream/actor.py
View File

@ -45,8 +45,7 @@ class StreamLibraryProvider(backend.LibraryProvider):
try:
tags, duration = self._scanner.scan(uri)
track = scan.audio_data_to_track({
'uri': uri, 'tags': tags, 'duration': duration, 'mtime': None})
track = scan.tags_to_track(tags).copy(uri=uri, length=duration)
except exceptions.ScannerError as e:
logger.warning('Problem looking up %s: %s', uri, e)
track = Track(uri=uri)

196
tests/audio/test_scan.py
View File

@ -16,33 +16,28 @@ from tests import path_to_data_dir
# TODO: keep ids without name?
class TranslatorTest(unittest.TestCase):
class TagsToTrackTest(unittest.TestCase):
def setUp(self): # noqa
self.data = {
'uri': 'uri',
'duration': 4531,
'mtime': 1234,
'tags': {
'album': ['album'],
'track-number': [1],
'artist': ['artist'],
'composer': ['composer'],
'performer': ['performer'],
'album-artist': ['albumartist'],
'title': ['track'],
'track-count': [2],
'album-disc-number': [2],
'album-disc-count': [3],
'date': [datetime.date(2006, 1, 1,)],
'container-format': ['ID3 tag'],
'genre': ['genre'],
'comment': ['comment'],
'musicbrainz-trackid': ['trackid'],
'musicbrainz-albumid': ['albumid'],
'musicbrainz-artistid': ['artistid'],
'musicbrainz-albumartistid': ['albumartistid'],
'bitrate': [1000],
},
self.tags = {
'album': ['album'],
'track-number': [1],
'artist': ['artist'],
'composer': ['composer'],
'performer': ['performer'],
'album-artist': ['albumartist'],
'title': ['track'],
'track-count': [2],
'album-disc-number': [2],
'album-disc-count': [3],
'date': [datetime.date(2006, 1, 1,)],
'container-format': ['ID3 tag'],
'genre': ['genre'],
'comment': ['comment'],
'musicbrainz-trackid': ['trackid'],
'musicbrainz-albumid': ['albumid'],
'musicbrainz-artistid': ['artistid'],
'musicbrainz-albumartistid': ['albumartistid'],
'bitrate': [1000],
}
artist = Artist(name='artist', musicbrainz_id='artistid')
@ -54,223 +49,215 @@ class TranslatorTest(unittest.TestCase):
album = Album(name='album', num_tracks=2, num_discs=3,
musicbrainz_id='albumid', artists=[albumartist])
self.track = Track(uri='uri', name='track', date='2006-01-01',
genre='genre', track_no=1, disc_no=2, length=4531,
self.track = Track(name='track', date='2006-01-01',
genre='genre', track_no=1, disc_no=2,
comment='comment', musicbrainz_id='trackid',
last_modified=1234, album=album, bitrate=1000,
artists=[artist], composers=[composer],
performers=[performer])
album=album, bitrate=1000, artists=[artist],
composers=[composer], performers=[performer])
def check(self, expected):
actual = scan.audio_data_to_track(self.data)
actual = scan.tags_to_track(self.tags)
self.assertEqual(expected, actual)
def test_track(self):
self.check(self.track)
def test_none_track_length(self):
self.data['duration'] = None
self.check(self.track.copy(length=None))
def test_none_track_last_modified(self):
self.data['mtime'] = None
self.check(self.track.copy(last_modified=None))
def test_missing_track_no(self):
del self.data['tags']['track-number']
del self.tags['track-number']
self.check(self.track.copy(track_no=None))
def test_multiple_track_no(self):
self.data['tags']['track-number'].append(9)
self.tags['track-number'].append(9)
self.check(self.track)
def test_missing_track_disc_no(self):
del self.data['tags']['album-disc-number']
del self.tags['album-disc-number']
self.check(self.track.copy(disc_no=None))
def test_multiple_track_disc_no(self):
self.data['tags']['album-disc-number'].append(9)
self.tags['album-disc-number'].append(9)
self.check(self.track)
def test_missing_track_name(self):
del self.data['tags']['title']
del self.tags['title']
self.check(self.track.copy(name=None))
def test_multiple_track_name(self):
self.data['tags']['title'] = ['name1', 'name2']
self.tags['title'] = ['name1', 'name2']
self.check(self.track.copy(name='name1; name2'))
def test_missing_track_musicbrainz_id(self):
del self.data['tags']['musicbrainz-trackid']
del self.tags['musicbrainz-trackid']
self.check(self.track.copy(musicbrainz_id=None))
def test_multiple_track_musicbrainz_id(self):
self.data['tags']['musicbrainz-trackid'].append('id')
self.tags['musicbrainz-trackid'].append('id')
self.check(self.track)
def test_missing_track_bitrate(self):
del self.data['tags']['bitrate']
del self.tags['bitrate']
self.check(self.track.copy(bitrate=None))
def test_multiple_track_bitrate(self):
self.data['tags']['bitrate'].append(1234)
self.tags['bitrate'].append(1234)
self.check(self.track)
def test_missing_track_genre(self):
del self.data['tags']['genre']
del self.tags['genre']
self.check(self.track.copy(genre=None))
def test_multiple_track_genre(self):
self.data['tags']['genre'] = ['genre1', 'genre2']
self.tags['genre'] = ['genre1', 'genre2']
self.check(self.track.copy(genre='genre1; genre2'))
def test_missing_track_date(self):
del self.data['tags']['date']
del self.tags['date']
self.check(self.track.copy(date=None))
def test_multiple_track_date(self):
self.data['tags']['date'].append(datetime.date(2030, 1, 1))
self.tags['date'].append(datetime.date(2030, 1, 1))
self.check(self.track)
def test_missing_track_comment(self):
del self.data['tags']['comment']
del self.tags['comment']
self.check(self.track.copy(comment=None))
def test_multiple_track_comment(self):
self.data['tags']['comment'] = ['comment1', 'comment2']
self.tags['comment'] = ['comment1', 'comment2']
self.check(self.track.copy(comment='comment1; comment2'))
def test_missing_track_artist_name(self):
del self.data['tags']['artist']
del self.tags['artist']
self.check(self.track.copy(artists=[]))
def test_multiple_track_artist_name(self):
self.data['tags']['artist'] = ['name1', 'name2']
self.tags['artist'] = ['name1', 'name2']
artists = [Artist(name='name1'), Artist(name='name2')]
self.check(self.track.copy(artists=artists))
def test_missing_track_artist_musicbrainz_id(self):
del self.data['tags']['musicbrainz-artistid']
del self.tags['musicbrainz-artistid']
artist = list(self.track.artists)[0].copy(musicbrainz_id=None)
self.check(self.track.copy(artists=[artist]))
def test_multiple_track_artist_musicbrainz_id(self):
self.data['tags']['musicbrainz-artistid'].append('id')
self.tags['musicbrainz-artistid'].append('id')
self.check(self.track)
def test_missing_track_composer_name(self):
del self.data['tags']['composer']
del self.tags['composer']
self.check(self.track.copy(composers=[]))
def test_multiple_track_composer_name(self):
self.data['tags']['composer'] = ['composer1', 'composer2']
self.tags['composer'] = ['composer1', 'composer2']
composers = [Artist(name='composer1'), Artist(name='composer2')]
self.check(self.track.copy(composers=composers))
def test_missing_track_performer_name(self):
del self.data['tags']['performer']
del self.tags['performer']
self.check(self.track.copy(performers=[]))
def test_multiple_track_performe_name(self):
self.data['tags']['performer'] = ['performer1', 'performer2']
self.tags['performer'] = ['performer1', 'performer2']
performers = [Artist(name='performer1'), Artist(name='performer2')]
self.check(self.track.copy(performers=performers))
def test_missing_album_name(self):
del self.data['tags']['album']
del self.tags['album']
album = self.track.album.copy(name=None)
self.check(self.track.copy(album=album))
def test_multiple_album_name(self):
self.data['tags']['album'].append('album2')
self.tags['album'].append('album2')
self.check(self.track)
def test_missing_album_musicbrainz_id(self):
del self.data['tags']['musicbrainz-albumid']
del self.tags['musicbrainz-albumid']
album = self.track.album.copy(musicbrainz_id=None,
images=[])
self.check(self.track.copy(album=album))
def test_multiple_album_musicbrainz_id(self):
self.data['tags']['musicbrainz-albumid'].append('id')
self.tags['musicbrainz-albumid'].append('id')
self.check(self.track)
def test_missing_album_num_tracks(self):
del self.data['tags']['track-count']
del self.tags['track-count']
album = self.track.album.copy(num_tracks=None)
self.check(self.track.copy(album=album))
def test_multiple_album_num_tracks(self):
self.data['tags']['track-count'].append(9)
self.tags['track-count'].append(9)
self.check(self.track)
def test_missing_album_num_discs(self):
del self.data['tags']['album-disc-count']
del self.tags['album-disc-count']
album = self.track.album.copy(num_discs=None)
self.check(self.track.copy(album=album))
def test_multiple_album_num_discs(self):
self.data['tags']['album-disc-count'].append(9)
self.tags['album-disc-count'].append(9)
self.check(self.track)
def test_missing_album_artist_name(self):
del self.data['tags']['album-artist']
del self.tags['album-artist']
album = self.track.album.copy(artists=[])
self.check(self.track.copy(album=album))
def test_multiple_album_artist_name(self):
self.data['tags']['album-artist'] = ['name1', 'name2']
self.tags['album-artist'] = ['name1', 'name2']
artists = [Artist(name='name1'), Artist(name='name2')]
album = self.track.album.copy(artists=artists)
self.check(self.track.copy(album=album))
def test_missing_album_artist_musicbrainz_id(self):
del self.data['tags']['musicbrainz-albumartistid']
del self.tags['musicbrainz-albumartistid']
albumartist = list(self.track.album.artists)[0]
albumartist = albumartist.copy(musicbrainz_id=None)
album = self.track.album.copy(artists=[albumartist])
self.check(self.track.copy(album=album))
def test_multiple_album_artist_musicbrainz_id(self):
self.data['tags']['musicbrainz-albumartistid'].append('id')
self.tags['musicbrainz-albumartistid'].append('id')
self.check(self.track)
def test_stream_organization_track_name(self):
del self.data['tags']['title']
self.data['tags']['organization'] = ['organization']
del self.tags['title']
self.tags['organization'] = ['organization']
self.check(self.track.copy(name='organization'))
def test_multiple_organization_track_name(self):
del self.data['tags']['title']
self.data['tags']['organization'] = ['organization1', 'organization2']
del self.tags['title']
self.tags['organization'] = ['organization1', 'organization2']
self.check(self.track.copy(name='organization1; organization2'))
# TODO: combine all comment types?
def test_stream_location_track_comment(self):
del self.data['tags']['comment']
self.data['tags']['location'] = ['location']
del self.tags['comment']
self.tags['location'] = ['location']
self.check(self.track.copy(comment='location'))
def test_multiple_location_track_comment(self):
del self.data['tags']['comment']
self.data['tags']['location'] = ['location1', 'location2']
del self.tags['comment']
self.tags['location'] = ['location1', 'location2']
self.check(self.track.copy(comment='location1; location2'))
def test_stream_copyright_track_comment(self):
del self.data['tags']['comment']
self.data['tags']['copyright'] = ['copyright']
del self.tags['comment']
self.tags['copyright'] = ['copyright']
self.check(self.track.copy(comment='copyright'))
def test_multiple_copyright_track_comment(self):
del self.data['tags']['comment']
self.data['tags']['copyright'] = ['copyright1', 'copyright2']
del self.tags['comment']
self.tags['copyright'] = ['copyright1', 'copyright2']
self.check(self.track.copy(comment='copyright1; copyright2'))
class ScannerTest(unittest.TestCase):
def setUp(self): # noqa
self.errors = {}
self.data = {}
self.tags = {}
self.durations = {}
def find(self, path):
media_dir = path_to_data_dir(path)
@ -285,40 +272,31 @@ class ScannerTest(unittest.TestCase):
key = uri[len('file://'):]
try:
tags, duration = scanner.scan(uri)
self.data[key] = {
'uri': uri, 'tags': tags, 'duration': duration}
self.tags[key] = tags
self.durations[key] = duration
except exceptions.ScannerError as error:
self.errors[key] = error
def check(self, name, key, value):
name = path_to_data_dir(name)
self.assertEqual(self.data[name][key], value)
def check_tag(self, name, key, value):
name = path_to_data_dir(name)
self.assertEqual(self.data[name]['tags'][key], value)
self.assertEqual(self.tags[name][key], value)
def test_data_is_set(self):
def check_duration(self, name, value):
name = path_to_data_dir(name)
self.assertEqual(self.durations[name], value)
def test_tags_is_set(self):
self.scan(self.find('scanner/simple'))
self.assert_(self.data)
self.assert_(self.tags)
def test_errors_is_not_set(self):
self.scan(self.find('scanner/simple'))
self.assert_(not self.errors)
def test_uri_is_set(self):
self.scan(self.find('scanner/simple'))
self.check(
'scanner/simple/song1.mp3', 'uri',
'file://%s' % path_to_data_dir('scanner/simple/song1.mp3'))
self.check(
'scanner/simple/song1.ogg', 'uri',
'file://%s' % path_to_data_dir('scanner/simple/song1.ogg'))
def test_duration_is_set(self):
self.scan(self.find('scanner/simple'))
self.check('scanner/simple/song1.mp3', 'duration', 4680)
self.check('scanner/simple/song1.ogg', 'duration', 4680)
self.check_duration('scanner/simple/song1.mp3', 4680)
self.check_duration('scanner/simple/song1.ogg', 4680)
def test_artist_is_set(self):
self.scan(self.find('scanner/simple'))