diff --git a/docs/source/tracks.rst b/docs/source/tracks.rst index 0413900..1aeb54e 100644 --- a/docs/source/tracks.rst +++ b/docs/source/tracks.rst @@ -11,7 +11,7 @@ RhythmTrack TimbreTrack ======================================== -.. autoclass:: comsar.tracks.TimbreTrack +.. autoclass:: rtd.xxx.MySuperClass :members: FormTrack diff --git a/src/rtd/__init__.py b/src/rtd/__init__.py index c6750dd..e69de29 100644 --- a/src/rtd/__init__.py +++ b/src/rtd/__init__.py @@ -1,3 +0,0 @@ -import pkg_resources as _pkg - -__version__ = _pkg.get_distribution('comsar').version diff --git a/src/rtd/tracks/__init__.py b/src/rtd/tracks/__init__.py deleted file mode 100644 index a9453e5..0000000 --- a/src/rtd/tracks/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from . _timbre import TimbreTrack diff --git a/src/rtd/tracks/_form.py b/src/rtd/tracks/_form.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/rtd/tracks/_pitch.py b/src/rtd/tracks/_pitch.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/rtd/tracks/_rhythm.py b/src/rtd/tracks/_rhythm.py deleted file mode 100644 index 68115df..0000000 --- a/src/rtd/tracks/_rhythm.py +++ /dev/null @@ -1,43 +0,0 @@ -# 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 diff --git a/src/rtd/tracks/_timbre.py b/src/rtd/tracks/_timbre.py deleted file mode 100644 index 7c05b5e..0000000 --- a/src/rtd/tracks/_timbre.py +++ /dev/null @@ -1,205 +0,0 @@ -"""comsar/tracks/timbre.py -- TimbreTack implementation -License: BSD-3-Clasuse -Copyright (C) 2020, Michael Blaß, michael.blass@uni-hamburg.de -""" -from datetime import datetime -from timeit import default_timer as timer -from typing import Optional - -import numpy as np -import pandas as pd - -from apollon.audio import AudioFile -from apollon.segment import Segmentation -from apollon.signal import container, features -from apollon.signal.spectral import StftSegments - -import comsar -from . utilities import (TrackMeta, TrackResult, SourceMeta, - TimbreTrackParams, TimbreTrackCorrGramParams) - -STFT_DEFAULT = container.StftParams(fps=44100, window='hamming', n_fft=None, - n_perseg=2**15, n_overlap=2**14, - extend=True, pad=True) - -CORR_DIM_DEFAULT = container.CorrDimParams(delay=14, m_dim=80, n_bins=1000, - scaling_size=10) - -CORR_GRAM_DEFAULT = container.CorrGramParams(wlen=2**10, n_delay=2**8, - total=True) - - -class TimbreTrack: - """High-level interface for timbre feature extraction.""" - def __init__(self, - stft_params: Optional[container.StftParams] = None, - corr_dim_params: Optional[container.CorrDimParams] = None, - ) -> None: - """ - Args: - stft_params: Parameter for STFT. - corr_dim_params: Parameter set for correlation dimension. - """ - self.params = TimbreTrackParams(stft_params or STFT_DEFAULT, - corr_dim_params or CORR_DIM_DEFAULT) - - self.cutter = Segmentation(self.params.stft.n_perseg, - self.params.stft.n_overlap, - self.params.stft.extend, - self.params.stft.pad) - - self.stft = StftSegments(self.params.stft.fps, self.params.stft.window, - self.params.stft.n_fft) - - self.feature_names = ('SpectralCentroid', 'SpectralSpread', - 'SpectralFlux', 'Roughness', 'Sharpness', - 'SPL', 'CorrelationDimension') - - self.funcs = [features.spectral_centroid, - features.spectral_spread, - features.spectral_flux, - features.roughness_helmholtz, - features.sharpness, - features.spl, - features.cdim] - - self.pace = np.zeros(self.n_features) - self.verbose = False - - @property - def n_features(self) -> int: - """Returns: - Number of features on ``TimbreTrack``. - """ - return len(self.feature_names) - - def extract(self, path) -> pd.DataFrame: - """Run TimbreTrack on audio file. - - Args: - path: Path to audio file. - - Returns: - Extracted features. - """ - snd = AudioFile(path) - if snd.fps != self.params.stft.fps: - snd.close() - raise ValueError('Sample rate of {snd!str} differs from init.') - - segs = self.cutter.transform(snd.data.squeeze()) - sxx = self.stft.transform(segs) - - args = [(sxx.frqs, sxx.power), - (sxx.frqs, sxx.power), - (sxx.abs,), - (sxx.d_frq, sxx.abs, 15000), - (sxx.frqs, sxx.abs), - (segs.data,), - (segs.data,)] - - kwargs = [{}, {}, {}, {}, {}, {}, self.params.corr_dim.to_dict()] - - out = np.zeros((segs.n_segs, self.n_features)) - for i, (fun, arg, kwarg) in enumerate(zip(self.funcs, args, kwargs)): - out[:, i] = self._worker(i, fun, arg, kwarg) - - file_meta = SourceMeta(*snd.file_name.split('.'), snd.hash) - track_meta = TrackMeta(comsar.__version__, datetime.utcnow(), - file_meta) - out = pd.DataFrame(data=out, columns=self.feature_names) - snd.close() - return TrackResult(track_meta, self.params, out) - - def _worker(self, idx, func, args, kwargs) -> np.ndarray: - print(self.feature_names[idx], end=' ... ') - pace = timer() - res = func(*args, **kwargs) - pace = timer() - pace - self.pace[idx] = pace - print(f'{pace:.4} s.') - return res - - -class TimbreTrackCorrGram: - """Compute timbre track of an audio file. - """ - def __init__(self, - stft_params: Optional[container.StftParams] = None, - corr_dim_params: Optional[container.CorrDimParams] = None, - corr_gram_params: Optional[container.CorrGramParams] = None) -> None: - """ - Args: - """ - self.params = TimbreTrackCorrGramParams(stft_params or STFT_DEFAULT, - corr_dim_params or CORR_DIM_DEFAULT, - corr_gram_params or CORR_GRAM_DEFAULT) - - self.cutter = Segmentation(self.params.stft.n_perseg, - self.params.stft.n_overlap, - self.params.stft.extend, - self.params.stft.pad) - self.stft = StftSegments(self.params.stft.fps, self.params.stft.window, - self.params.stft.n_fft) - - self.feature_names = ('SpectralCentroid', 'SpectralSpread', - 'SpectralFlux', 'Roughness', 'Sharpness', - 'SPL', 'CorrelationDimension', 'Correlogram') - - self.funcs = [features.spectral_centroid, - features.spectral_spread, - features.spectral_flux, - features.roughness_helmholtz, - features.sharpness, - features.spl, - features.cdim, - features.correlogram] - - self.pace = np.zeros(self.n_features) - self.verbose = False - - @property - def n_features(self) -> int: - """Number of features on track""" - return len(self.feature_names) - - def extract(self, path) -> pd.DataFrame: - """Perform extraction. - """ - snd = AudioFile(path) - if snd.fps != self.params.stft.fps: - snd.close() - raise ValueError('Sample rate of {snd!str} differs from init.') - - segs = self.cutter.transform(snd.data.squeeze()) - sxx = self.stft.transform(segs) - - args = [(sxx.frqs, sxx.power), - (sxx.frqs, sxx.power), - (sxx.abs,), - (sxx.d_frq, sxx.abs, 15000), - (sxx.frqs, sxx.abs), - (segs.data,), - (segs.data,), - (segs.data,)] - - kwargs = [{}, {}, {}, {}, {}, {}, self.params.corr_dim.to_dict(), - self.params.corr_gram.to_dict()] - - out = np.zeros((segs.n_segs, self.n_features)) - for i, (fun, arg, kwarg) in enumerate(zip(self.funcs, args, kwargs)): - out[:, i] = self._worker(i, fun, arg, kwarg) - snd.close() - - meta = TrackMeta(comsar.__version__, datetime.utcnow(), snd.file_name) - out = pd.DataFrame(data=out, columns=self.feature_names) - return TrackResult(meta, self.params, out) - - def _worker(self, idx, func, args, kwargs) -> np.ndarray: - print(self.feature_names[idx], end=' ... ') - pace = timer() - res = func(*args, **kwargs) - pace = timer() - pace - self.pace[idx] = pace - print(f'{pace:.4} s.') - return res diff --git a/src/rtd/tracks/utilities.py b/src/rtd/tracks/utilities.py deleted file mode 100644 index c07c68b..0000000 --- a/src/rtd/tracks/utilities.py +++ /dev/null @@ -1,135 +0,0 @@ -"""comsar/tracks/untilities.py -- Utilities -License: BSD-3-Clasuse -Copyright (C) 2020, Michael Blaß, michael.blass@uni-hamburg.de -""" -import datetime -import pathlib -import pickle -from typing import ClassVar, Type, TypeVar, Union - -from dataclasses import dataclass -import numpy as np -import pandas as pd - -from apollon import io -from apollon import container -from apollon.signal import container as asc -from apollon.tools import standardize -from apollon import types - - -T = TypeVar('T') - - -@dataclass -class SourceMeta(container.Params): - """Source file meta data.""" - _schema: ClassVar[types.Schema] = None - name: str - extension: str - hash_: str - - -@dataclass -class TrackMeta(container.Params): - """Track meta data.""" - _schema: ClassVar[types.Schema] = None - version: str - extraction_date: datetime.datetime - source: SourceMeta - - -@dataclass -class TrackParams(container.Params): - """Track parameter base class.""" - _schema: ClassVar[types.Schema] = None - - -@dataclass -class TimbreTrackParams(TrackParams): - """Parameter set for TimbreTrack""" - stft: asc.StftParams - corr_dim: asc.CorrDimParams - - -@dataclass -class TimbreTrackCorrGramParams(TrackParams): - """Parameter set for TimbreTrack""" - stft: asc.StftParams - corr_dim: asc.CorrDimParams - - -class TrackResult: - """Provide track results.""" - def __init__(self, meta: TrackMeta, params: TrackParams, - data: pd.DataFrame) -> None: - self._meta = meta - self._params = params - self._data = data - - @property - def data(self) -> np.ndarray: - """Return the raw data array.""" - return self._data.to_numpy() - - @property - def features(self) -> pd.DataFrame: - """Extracted feautures.""" - return self._data - - @property - def features_names(self) -> list: - """Name of each feature.""" - return self._data.columns.to_list() - - @property - def z_score(self) -> pd.DataFrame: - """Z-score of extracted features.""" - return standardize(self.features) - - def to_csv(self, path: Union[str, pathlib.Path]) -> None: - """Serialize features to csv file. - - This does not save parameters, and meta data. - - Args: - path: Destination path. - """ - self._data.to_csv(path) - - def to_dict(self) -> dict: - """Serialize TrackResults to dictionary.""" - return {'meta': self._meta.to_dict(), - 'params': self._params.to_dict(), - 'data': self._data.to_dict(orient='list')} - - def to_json(self, path: Union[str, pathlib.Path]) -> None: - """Serialize TrackResults to JSON.""" - io.json.dump(self.to_dict(), path) - - - def to_mongo(self, db_con) -> None: - """Write TrackResults to open MongoDB connection:""" - pass - - def to_pickle(self, path: Union[str, pathlib.Path]) -> None: - """Serialize Track Results to pickle.""" - path = pathlib.Path(path) - with path.open('wb') as fobj: - pickle.dump(self, fobj) - - @classmethod - def read_json(cls: Type[T], path: Union[str, pathlib.Path]) -> T: - """Read TrackResults form json.""" - raw = io.json.load(path) - meta = TrackMeta.from_dict(raw['meta']) - params = TimbreTrackParams.from_dict(raw['params']) - data = pd.DataFrame(raw['data']) - return cls(meta, params, data) - - @classmethod - def read_pickle(cls: Type[T], path: Union[str, pathlib.Path]) -> T: - """Read pickled TrackResults.""" - path = pathlib.Path(path) - with path.open('rb') as fobj: - return pickle.load(fobj) diff --git a/src/rtd/xxx.py b/src/rtd/xxx.py new file mode 100644 index 0000000..1d0b23e --- /dev/null +++ b/src/rtd/xxx.py @@ -0,0 +1,6 @@ +from apollon.audio import AudioFile + + +class MySuperClass: + def __init__(self, herb): + self.herb = herb