Skip to content
Snippets Groups Projects
Commit e676882a authored by Blaß, Michael's avatar Blaß, Michael :speech_balloon:
Browse files

Init

parents
No related branches found
No related tags found
No related merge requests found
# Mac
.DS_Store
*.py[cod]
__pychache__/
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
*.so
.mypy*/
*.swp
*.ipynb
.ipynb_checkpoints
.mypy_cache
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
.hypothesis/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# dotenv
.env
# virtualenv
.venv
venv/
ENV/
# Spyder project settings
.spyderproject
# Rope project settings
.ropeproject
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# mblass@posteo.net
from . import apollon_features
from . import apollon_onsets
from . import apollon_hmm
from . import apollon_position
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# mblass@posteo.net
import argparse
import itertools
import json
import sys
import typing
import soundfile as sf
import logging
import time
from .. import analyses
from .. types import PathType
from .. import io
from .. audio import load_audio
class ShortPiece(Exception):
pass
class BadSampleRate(Exception):
pass
def _check_audio(path):
snd_info = sf.info(path)
if snd_info.duration < 30:
logging.error('Piece to short: {}'.format(path))
# raise ShortPiece('Duration of {} is less than {} s.'.format(path, 30))
return 10
if snd_info.samplerate != 44100:
logging.error('Bad sample rate: {}'.format(path))
# raise BadSampleRate('Sample rate of {} Hz cannot be processed'.format(snd_info.samplerate))
return 10
return 0
def main(argv: dict = None) -> int:
logging.basicConfig(filename='fe.log', filemode='w', level=logging.DEBUG)
if argv is None:
argv = sys.argv
timer_total = time.time()
for path in argv.files:
logging.info('Loading {}'.format(path))
print('Processing: {}'.format(path), end=' ... ', flush=True)
timer_start = time.time()
if _check_audio(path) != 0:
return 10
snd = load_audio(path)
snd.cut(snd.fps*2, snd.size-(snd.fps*5))
track_data = {}
if argv.rhythm:
track_data['rhythm'] = analyses.rhythm_track(snd)
if argv.timbre:
track_data['timbre'] = analyses.timbre_track(snd)
if argv.pitch:
track_data['pitch'] = analyses.pitch_track(snd)
out_path = io.generate_outpath(path, argv.outpath, 'feat')
io.dump_json(track_data, out_path)
timer_stop = time.time()
print('Done in {:.5} s.'.format(timer_stop-timer_start), flush=True)
logging.info('--- JOB DONE ---')
print('Job done. Total time: {:.5} s.'.format(time.time()-timer_total))
return 0
if __name__ == '__main__':
sys.exit(main())
#!/usr/bin/env python3
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# mblass@posteo.net
import argparse
import json
import pathlib
import sys
import typing
import numpy as np
from .. import io
from .. hmm import PoissonHmm
from .. types import Array as _Array
from .. import tools
def _load_track_file(track_file: str) -> dict:
track_file = pathlib.Path(track_file)
with track_file.open('r') as fobj:
track_data = json.load(fobj, object_hook=io.decode_array)
return track_data
def _parse_feature_file(track_file: str, feature_path: str) -> _Array:
feature = _load_track_file(track_file)
for key in feature_path.split('.'):
try:
feature = feature[key]
except KeyError:
print('Error. Node "{}" not in "{}".'.format(key, track_file))
exit(10)
return _scaling(feature, key)
def _scaling(data: _Array, feature: str) -> _Array:
features1000 = ['skewness', 'kurtosis', 'loudness',
'roughness', 'sharpness']
if feature in features1000:
out = tools.scale(data, 1, 1000)
else:
out = data
return out.round().astype(int)
def _train_n_hmm(data: _Array, m_states: int, n_trails: int):
"""Trains ``n_trails`` HMMs each initialized with a random tpm.
Args:
data: Possibly unporcessed input data set.
m_states: Number of states.
n_trails: Number of trails.
Returns:
Best model regarding to log-likelihood.
"""
trails = []
for i in range(n_trails):
hmm = PoissonHmm(data, m_states, init_lambda='hist', init_gamma='softmax')
hmm.fit(data)
if hmm.success and not np.isnan(hmm.quality.nll):
trails.append(hmm)
for i in range(n_trails):
hmm = PoissonHmm(data, m_states, init_lambda='linear', init_gamma='softmax')
hmm.fit(data)
if hmm.success and not np.isnan(hmm.quality.nll):
trails.append(hmm)
for i in range(n_trails):
hmm = PoissonHmm(data, m_states, init_lambda='quantile', init_gamma='softmax')
hmm.fit(data)
if hmm.success and not np.isnan(hmm.quality.nll):
trails.append(hmm)
if len(trails) == 0:
return None
return min(trails, key=lambda hmm: abs(hmm.quality.nll))
def main(argv=None) -> int:
if argv is None:
argv = sys.argv
for trf in argv.track_files:
feature = _parse_feature_file(trf, argv.feature_path)
hmm = _train_n_hmm(feature, argv.mstates, 5)
if hmm is None:
print('Error. Could not train HMM on {}'.format(trf))
continue
out_path = io.generate_outpath(trf, argv.outpath, 'hmm')
io.dump_json(hmm.to_dict(), out_path)
return 0
if __name__ == '__main__':
sys.exit(main())
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# mblass@posteo.net
import argparse
import multiprocessing as mp
import sys
from .. import onsets
def _parse_cml(argv):
parser = argparse.ArgumentParser(description='Apollon onset detection engine')
parser.add_argument('--amplitude', action='store_true',
help='Detect onsets based on local extrema in the time domain signal.')
parser.add_argument('--entropy', action='store_true',
help='Detect onsets based on time domain entropy maxima.')
parser.add_argument('--flux', action='store_true',
help='Detect onsets based on spectral flux.')
parser.add_argument('-o', '--outpath', action='store',
help='Output file path.')
parser.add_argument('filepath', type=str, nargs=1)
return parser.parse_args(argv)
def _amp(a):
print('Amplitude')
return a
def _entropy(a):
print('Entropy')
return a
def _flux(a):
print('Flux')
return a
def main(argv=None):
if argv is None:
argv = sys.argv
args = _parse_cml(argv)
args = _parse_cml(argv)
detectors = {'amplitude': _amp,
'entropy': _entropy,
'flux': _flux}
methods = [func for name, func in detectors.items() if getattr(args, name)]
if len(methods) == 0:
print('At least one detection method required. Aborting.')
return 1
with mp.Pool(processes=3) as pool:
results = [pool.apply_async(meth, (i,)) for i, meth in enumerate(methods)]
out = [res.get() for res in results]
return out
if __name__ == '__main__':
sys.exit(main())
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# mblass@posteo.net
import argparse
import pathlib
import pickle
import sys
import numpy as np
import pandas as pd
from .. import io
from .. som.utilities import get_winner
def main(argv=None) -> int:
if argv is None:
argv = sys.argv
weights = io.load(argv.som_file)
if argv.rt:
for hmm in argv.objective_files:
hmm = io.load_json(hmm)
gamma_ = hmm.params.gamma_.astype('float64').flatten()
flat_idx = get_winner(weights, gamma_)
print(np.unravel_index(flat_idx, (30, 30)))
elif argv.tt:
fs_path = pathlib.Path(argv.som_file).parent.joinpath('fstats.pkl')
with fs_path.open('rb') as fobj:
fstats = pickle.load(fobj)
feature_mean, feature_std = fstats
for file_name in argv.objective_files:
data = io.load_json(file_name)
feat = data.timbre.features
timbre_vector = np.array([
feat.spectral.centroid.mean(), feat.spectral.centroid.std(),
feat.spectral.spread.mean(), feat.spectral.spread.std(),
feat.spectral.skewness.mean(), feat.spectral.skewness.std(),
feat.spectral.kurtosis.mean(), feat.spectral.kurtosis.std(),
feat.temporal.flux.mean(), feat.temporal.flux.std(),
feat.perceptual.roughness.mean(), feat.perceptual.roughness.std(),
feat.perceptual.sharpness.mean(), feat.perceptual.sharpness.std(),
feat.perceptual.loudness.mean(), feat.perceptual.loudness.std()])
timbre_vector = (timbre_vector - feature_mean) / feature_std
flat_idx = get_winner(weights, timbre_vector)
print(np.unravel_index(flat_idx, (40, 40)))
else:
print('Error. Need track info.')
return -10
return 0
if __name__ == '__main__':
sys.exit(main())
import logging
class TrackBase:
def __init__(self, source)
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# michael.blass@uni-hamburg.de
import logging
from apollon.signal import spectral
from apollon import tools
from apollon import onsets
from apollon import segment
def rhythm_track(snd: AudioFile) -> dict:
"""Perform rhythm track analysis of given audio file.
Args:
snd: Sound data.
fps: Sample rate.
Returns:
Rhythm track parameters and data.
"""
logging.info('Starting rhythm track for {!s}'.format(snd.file))
onsets = FluxOnsetDetector(snd.data, snd.fps)
segs = segment.by_onsets(snd.data, 2**11, onsets.index())
spctr = Spectrum(segs, snd.fps, window='hamming', n_fft=2**15)
onsets_features = {
'peaks': onsets.peaks,
'index': onsets.index(),
'times': onsets.times(snd.fps)
}
track_data = {
'meta': {'source': str(snd.file.absolute()),
'time_stamp': time_stamp()},
'params': {'onsets': onsets.params(), 'spectrum': spctr.params()},
'features': {'onsets': onsets_features,
'spectrum':
spctr.extract(cf_low=100, cf_high=9000).as_dict()}
}
logging.info(f'Done with rhythm track for {snd.file!s}.')
return track_data
#!/usr/bin/env python3
# Licensed under the terms of the BSD-3-Clause license.
# Copyright (C) 2019 Michael Blaß
# michael.blass@uni-hamburg.de
"""Launch the comsar main app."""
import sys
import argparse
import apollon
from apollon import commands
_valid_subcommand = ('features', 'onsets', 'hmm', 'som', 'export', 'position')
def _parse_cml(argv):
parser = argparse.ArgumentParser('Computational Music and Sound Archiving')
parser.add_argument(
'--version', action='version', version=apollon.__version__,
help='Display apollon version.')
subparsers = parser.add_subparsers()
sp_features = _create_subparser_features(subparsers)
sp_hmm = _create_subparser_hmm(subparsers)
sp_position = _create_subparser_position(subparsers)
return parser.parse_args(argv[1:])
def _create_subparser_features(subparsers):
sp_features = subparsers.add_parser('features', help='')
sp_features.add_argument(
'--rhythm', action='store_true',
help='Extract features for rhythm track.')
sp_features.add_argument(
'--timbre', action='store_true',
help='Extract features for timbre track.')
sp_features.add_argument(
'--pitch', action='store_true',
help='Extract pitch track features')
sp_features.add_argument(
'--export', type=str, action='store')
sp_features.add_argument(
'-o', '--outpath', action='store', help='Output file path.')
sp_features.add_argument(
'files', type=str, nargs='+',
help='Auio files.')
sp_features.set_defaults(func=commands.apollon_features.main)
return sp_features
def _create_subparser_hmm(subparsers):
sp_hmm = subparsers.add_parser('hmm', help='Train HMMs')
sp_hmm.add_argument(
'track_files', type=str, action='store', nargs='+',
help='Path to track file or path to file of paths')
sp_hmm.add_argument(
'feature_path', type=str, action='store',
help='Specifies feature within track file.')
sp_hmm.add_argument(
'-m', '--mstates', type=int, action='store', default=4,
help='Number of HMM states.')
sp_hmm.add_argument(
'-o', '--outpath', type=str, action='store',
help='Output file path.')
sp_hmm.set_defaults(func=commands.apollon_hmm.main)
return sp_hmm
def _create_subparser_position(subparsers):
sp_position = subparsers.add_parser('position',
help='Map coordinate of input.')
sp_position.add_argument(
'som_file', type=str, action='store',
help='Path to SOM file.')
sp_position.add_argument(
'objective_files', type=str, action='store', nargs='+',
help='Path to objective files.')
sp_position.add_argument(
'--rt', action='store_true', default=False,
help='Compute SOM position for rhythm track.')
sp_position.add_argument(
'--tt', action='store_true', default=False,
help='Compute SOM position for timbr track.')
sp_position.set_defaults(func=commands.apollon_position.main)
return sp_position
def main(argv=None):
if argv is None:
argv = sys.argv
args = _parse_cml(argv)
return args.func(args)
if __name__ == '__main__':
sys.exit(main())
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""get_som_winner.py
(c) Michael Blaß, 2016
Return index of winning neuron given a SOM and a HMM.
"""
import sys
from optparse import OptionParser
import pathlib
from numpy import atleast_2d
from numpy import unravel_index
from apollon import io
def main():
usage = 'usage: %prog [OPTIONS] som_file hmm_file'
parser = OptionParser(usage=usage)
parser.add_option('-f', '--flat', action='store_true',
help='return flat index')
(opts, args) = parser.parse_args()
if len(args) != 2:
print('Specify exactly two arguments (som_file, hmm_file).')
sys.exit(1)
som_file = pathlib.Path(args[0])
hmm_file = pathlib.Path(args[1])
if som_file.exists() and str(som_file).endswith('.som'):
if hmm_file.exists() and str(hmm_file).endswith('.hmm'):
som = io.load(str(som_file))
hmm = io.load(str(hmm_file))
else:
raise FileNotFoundError('File {} not found or is no valid HMM.'
.format(args[1]))
else:
raise FileNotFoundError('File {} not found or is no valid SOM.'
.format(args[0]))
foo = som.get_winners(hmm.reshape(16))[0]
if opts.flat:
print(foo)
else:
x, y = unravel_index(foo, som.shape[:2])
print("{},{}".format(x, y))
if __name__ == "__main__":
sys.exit(main())
#!/usr/bin/env python3
"""make_hmm.py
(c) Michael Blaß, 2016
Train a PoissonHmm
"""
import pathlib
from optparse import OptionParser
import sys
from apollon import segment
from apollon.audio import load_audio
from apollon.onsets import FluxOnsetDetector
from apollon.signal.spectral import Spectrum
from apollon.hmm import PoissonHmm
from apollon.io import save
def main():
def verbose_msg(s):
if opts.verbose:
print(s)
usage = 'usage: %prog [OPTIONS] path_to_wav'
parser = OptionParser(usage=usage)
parser.add_option('-v', '--verbose', action='store_true',
help='enable verbose mode')
(opts, args) = parser.parse_args()
if len(args) == 0:
print('Path to .wav-file not specified.')
sys.exit(1)
snd = load_audio(args[0])
onsets = FluxOnsetDetector(snd.data, snd.fps)
segs = segment.by_onsets(snd.data, 2**11, onsets.index())
spctr = Spectrum(segs, snd.fps, window='hamming')
feat = spctr.centroid().round().astype(int)
mod = PoissonHmm(feat, 4, verbose=False)
mod.fit(feat)
# save model
out_fname = snd.file.stem + '.hmm'
save(mod.params.gamma_, out_fname)
if __name__ == "__main__":
sys.exit(main())
[metadata]
name = apollon
version = 0.1.2.1
description = Computational Music and Sound Archiving
long_description = file: README.md
licence = BSD-3-Clause
author = Michael Blaß
author_email = michael.blass@uni-hamburg.de
keywords = hmm, som, apollon, comsar, music, analysis
classifiers =
Programming Language :: Python :: 3
Programming Language :: Python :: 3.6
Programming Language :: Python :: 3.7
Programming Language :: Python :: 3 :: Only
Programming Language :: Python :: Implementation :: CPython
Topic :: Scientific/Engineering
Topic :: Database
Topic :: Multimedia :: Sound/Audio :: Analysis
Intended Audience :: Science/Research
Intended Audience :: Information Technology
License :: OSI Approved :: BSD License
[options]
zip_safe = True
include_package_data = True
packages = find:
scripts = scripts/comsar
setup_requires = numpy
install_requires =
numpy
scipy >= "0.19.0"
matplotlib >= "2"
pandas >= "0.20"
setuptools >= "40.0.0"
sklearn >= "0.20"
#!/usr/bin/env python3
from setuptools import setup
from setuptools.config import read_configuration
config = read_configuration('./setup.cfg')
setup()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment