-
-
Notifications
You must be signed in to change notification settings - Fork 20
/
bpm_lookup.py
70 lines (52 loc) · 1.9 KB
/
bpm_lookup.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import re
import sys
import uuid
from urllib.parse import quote
import requests
import ujson
from troi import Element, Artist, Recording, PipelineError
class BPMLookupElement(Element):
'''
Look up musicbrainz earliest release year for a list of recordings, based on artist credit name and recording name.
By default items that are not found in the year lookup are not returned by this element. Pass
skip_not_found=False to init to keep tracks that failed the year lookup.
'''
SERVER_URL = "https://bono.metabrainz.org/bpm-key-lookup/json"
def __init__(self, skip_not_found=True):
Element.__init__(self)
self.skip_not_found = skip_not_found
@staticmethod
def inputs():
return [ Recording ]
@staticmethod
def outputs():
return [ Recording ]
def read(self, inputs):
recordings = inputs[0]
if not recordings:
return []
data = []
for r in recordings:
data.append({ '[recording_mbid]': r.mbid })
r = requests.post(self.SERVER_URL, json=data)
if r.status_code != 200:
raise PipelineError("Cannot fetch recording BPM from datasets: HTTP code %d" % r.status_code)
try:
rows = ujson.loads(r.text)
except ValueError as err:
raise PipelineError("Cannot fetch recording BPM from datasets: " + str(err))
mbid_index = {}
for row in rows:
mbid_index[row['recording_mbid']] = row['bpm']
output = []
for r in recordings:
try:
r.acousticbrainz["bpm"] = mbid_index[r.mbid]
except KeyError:
if self.skip_not_found:
self.debug("recording (%s) not found, skipping." % (r.mbid))
else:
output.append(r)
continue
output.append(r)
return output