56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
from json import JSONEncoder
|
|
import numpy
|
|
import os
|
|
import os.path
|
|
import logging
|
|
import requests
|
|
from pyAudioAnalysis import audioBasicIO
|
|
from pyAudioAnalysis import ShortTermFeatures
|
|
|
|
|
|
class NumpyArrayEncoder(JSONEncoder):
|
|
def default(self, obj):
|
|
if isinstance(obj, numpy.ndarray):
|
|
return obj.tolist()
|
|
return JSONEncoder.default(self, obj)
|
|
|
|
|
|
def do_extraction(file_path: str):
|
|
logging.info("Running extraction...")
|
|
|
|
[Fs, x] = audioBasicIO.read_audio_file(file_path)
|
|
F, f_names = ShortTermFeatures.feature_extraction(x, Fs, 0.050 * Fs, 0.025 * Fs)
|
|
|
|
return {"F": F, "f_names": f_names}
|
|
|
|
|
|
def run_everything(parameters: dict):
|
|
tag = parameters['tag']
|
|
logging.info(f"Downloading sample: {tag}")
|
|
|
|
file_path = os.path.join("/tmp/extractor-service/", f"{tag}.wav")
|
|
r = requests.get(f"http://storage-service/object/{tag}")
|
|
with open(file_path, 'wb') as f:
|
|
f.write(r.content)
|
|
|
|
# download done. Do extraction magic
|
|
try:
|
|
results = do_extraction(file_path)
|
|
finally:
|
|
os.remove(file_path)
|
|
|
|
logging.info(f"Pushing results to AI service...")
|
|
|
|
response = {
|
|
"tag": tag,
|
|
"results": results
|
|
}
|
|
|
|
logging.debug(f"Data being pushed: {str(response)}")
|
|
|
|
# r = requests.post('http://ai-service/asd', data=json.dumps(results, cls=NumpyArrayEncoder), headers={'Content-Type': 'application/json'})
|
|
|
|
# r.raise_for_status()
|