<commit_before>import json
from scipy.io import savemat
from six.moves import cStringIO
import tempfile
import numpy as np
from copy import deepcopy
import timeit

from alignedpitchfilter.AlignedPitchFilter import AlignedPitchFilter
from alignednotemodel.AlignedNoteModel import AlignedNoteModel

from tomato.MCRCaller import MCRCaller
from tomato.IO import IO
from tomato.Analyzer import Analyzer
from tomato.Plotter import Plotter

import warnings
import logging
logging.basicConfig(level=logging.INFO)

# instantiate a mcr_caller
_mcr_caller = MCRCaller()


class JointAnalyzer(Analyzer):
    def __init__(self, verbose=False):
        super(JointAnalyzer, self).__init__(verbose=verbose)

        # extractors
        self._tonicTempoExtractor = _mcr_caller.get_binary_path(
            'extractTonicTempoTuning')
        self._audioScoreAligner = _mcr_caller.get_binary_path(
            'alignAudioScore')
        self._alignedPitchFilter = AlignedPitchFilter()
        self._alignedNoteModel = AlignedNoteModel()

    def analyze(self, score_filename='', score_data=None,
                audio_filename='', audio_pitch=None, audio_tonic=None,
                audio_tempo = None):
        # joint score-informed tonic identification and tempo estimation
        try:  # if both are given in advance don't recompute
            if audio_tempo is not None or audio_tempo is not None:
                tonic, tempo = self.extract_tonic_tempo(
                    score_filename, score_data, audio_filename, audio_pitch)
                audio_tonic = audio_tonic if audio_tonic is not None else tonic
                audio_tempo = audio_tempo if audio_tempo is not None else tempo
        except RuntimeError as e:
            warnings.warn(e.message, RuntimeWarning)
            joint_features = None
            audio_features = None
            return joint_features, audio_features

        # section linking and note-level alignment
        try:
            aligned_sections, notes, section_links, section_candidates = \
                self.align_audio_score(score_filename, score_data,
                                       audio_filename, audio_pitch, audio_tonic,
                                       audio_tempo)
        except RuntimeError as e:
            warnings.warn(e.message, RuntimeWarning)
            joint_features = None
            audio_features = {'tonic': audio_tonic, 'tempo': audio_tempo}
            return joint_features, audio_features

        # aligned pitch filter
        aligned_pitch, aligned_notes = self.filter_pitch(audio_pitch, notes)

        # aligned note model
        note_models, pitch_distribution, aligned_tonic = self.get_note_models(
            aligned_pitch, notes, audio_tonic['symbol'])

        joint_features = {'sections': aligned_sections, 'notes': aligned_notes,
                          'note_models': note_models}
        audio_features = {'pitch': aligned_pitch, 'tonic': aligned_tonic,
                          'tempo': audio_tempo}

        return joint_features, audio_features

    @staticmethod
    def summarize(audio_features=None, score_features=None,
                  joint_features=None, score_informed_audio_features=None):
        # initialize
        sdict = {'audio': {}, 'score': score_features, 'joint': {}}

        sdict['audio']['metadata'] = audio_features['metadata']
        if score_informed_audio_features is not None:
            sdict['audio']['pitch'] = score_informed_audio_features['pitch']
            sdict['audio']['tonic'] = score_informed_audio_features['tonic']
            sdict['audio']['tempo'] = score_informed_audio_features['tempo']
            sdict['audio']['melodic_progression'] = \
                score_informed_audio_features['melodic_progression']
            sdict['audio']['transposition'] = score_informed_audio_features[
                'transposition']
            sdict['audio']['pitch_distribution'] = \
                score_informed_audio_features['pitch_distribution']
            sdict['audio']['pitch_class_distribution'] = \
                score_informed_audio_features['pitch_class_distribution']
        else:
            sdict['audio']['pitch'] = audio_features['pitch_filtered']
            sdict['audio']['tonic'] = audio_features['tonic']
            sdict['audio']['melodic_progression'] = audio_features[
                'melodic_progression']
            sdict['audio']['transposition'] = audio_features['transposition']
            sdict['audio']['pitch_distribution'] = \
                audio_features['pitch_distribution']
            sdict['audio']['pitch_class_distribution'] = \
                audio_features['pitch_class_distribution']
            # only add stable_notes if the joint features, hence the note
            # models are not given
            sdict['audio']['stable_notes'] = audio_features['stable_notes']

        if score_features is not None:
            sdict['audio']['makam'] = score_features['makam']['symbtr_slug']
        else:
            sdict['audio']['makam'] = audio_features['makam']

        # accumulate joint dict
        if joint_features is not None:
            sdict['joint']['sections'] = joint_features['sections']
            sdict['joint']['notes'] = joint_features['notes']
            sdict['joint']['note_models'] = joint_features['note_models']

        return sdict

    def extract_tonic_tempo(self, score_filename='', score_data=None,
                            audio_filename='', audio_pitch=None):
        tic = timeit.default_timer()
        self.vprint(u"- Extracting score-informed tonic and tempo of {0:s}"
                    .format(audio_filename))

        # create the temporary input and output files wanted by the binary
        temp_score_data_file = IO.create_temp_file(
            '.json', json.dumps(score_data))

        # matlab
        matout = cStringIO()
        savemat(matout, audio_pitch)

        temp_pitch_file = IO.create_temp_file(
            '.mat', matout.getvalue())

        temp_out_folder = tempfile.mkdtemp()
<commit_msg>Made tonic and tempo extraction in analysis modular<commit_after>import json
from scipy.io import savemat
from six.moves import cStringIO
import tempfile
import numpy as np
from copy import deepcopy
import timeit

from alignedpitchfilter.AlignedPitchFilter import AlignedPitchFilter
from alignednotemodel.AlignedNoteModel import AlignedNoteModel

from tomato.MCRCaller import MCRCaller
from tomato.IO import IO
from tomato.Analyzer import Analyzer
from tomato.Plotter import Plotter

import warnings
import logging
logging.basicConfig(level=logging.INFO)

# instantiate a mcr_caller
_mcr_caller = MCRCaller()


class JointAnalyzer(Analyzer):
    def __init__(self, verbose=False):
        super(JointAnalyzer, self).__init__(verbose=verbose)

        # extractors
        self._tonicTempoExtractor = _mcr_caller.get_binary_path(
            'extractTonicTempoTuning')
        self._audioScoreAligner = _mcr_caller.get_binary_path(
            'alignAudioScore')
        self._alignedPitchFilter = AlignedPitchFilter()
        self._alignedNoteModel = AlignedNoteModel()

    def analyze(self, score_filename='', score_data=None,
                audio_filename='', audio_pitch=None, audio_tonic=None,
                audio_tempo=None):
        # joint score-informed tonic identification and tempo estimation
        try:  # if both are given in advance don't recompute
            audio_tonic, audio_tempo = self._parse_and_extract_tonic_tempo(
                score_filename, score_data, audio_filename, audio_pitch,
                audio_tonic, audio_tempo)
        except RuntimeError as e:
            warnings.warn(e.message, RuntimeWarning)
            joint_features = None
            audio_features = None
            return joint_features, audio_features

        # section linking and note-level alignment
        try:
            aligned_sections, notes, section_links, section_candidates = \
                self.align_audio_score(score_filename, score_data,
                                       audio_filename, audio_pitch,
                                       audio_tonic, audio_tempo)
        except RuntimeError as e:
            warnings.warn(e.message, RuntimeWarning)
            joint_features = None
            audio_features = {'tonic': audio_tonic, 'tempo': audio_tempo}
            return joint_features, audio_features

        # aligned pitch filter
        aligned_pitch, aligned_notes = self.filter_pitch(audio_pitch, notes)

        # aligned note model
        note_models, pitch_distribution, aligned_tonic = self.get_note_models(
            aligned_pitch, notes, audio_tonic['symbol'])

        joint_features = {'sections': aligned_sections, 'notes': aligned_notes,
                          'note_models': note_models}
        audio_features = {'pitch': aligned_pitch, 'tonic': aligned_tonic,
                          'tempo': audio_tempo}

        return joint_features, audio_features

    @staticmethod
    def summarize(audio_features=None, score_features=None,
                  joint_features=None, score_informed_audio_features=None):
        # initialize
        sdict = {'audio': {}, 'score': score_features, 'joint': {}}

        sdict['audio']['metadata'] = audio_features['metadata']
        if score_informed_audio_features is not None:
            sdict['audio']['pitch'] = score_informed_audio_features['pitch']
            sdict['audio']['tonic'] = score_informed_audio_features['tonic']
            sdict['audio']['tempo'] = score_informed_audio_features['tempo']
            sdict['audio']['melodic_progression'] = \
                score_informed_audio_features['melodic_progression']
            sdict['audio']['transposition'] = score_informed_audio_features[
                'transposition']
            sdict['audio']['pitch_distribution'] = \
                score_informed_audio_features['pitch_distribution']
            sdict['audio']['pitch_class_distribution'] = \
                score_informed_audio_features['pitch_class_distribution']
        else:
            sdict['audio']['pitch'] = audio_features['pitch_filtered']
            sdict['audio']['tonic'] = audio_features['tonic']
            sdict['audio']['melodic_progression'] = audio_features[
                'melodic_progression']
            sdict['audio']['transposition'] = audio_features['transposition']
            sdict['audio']['pitch_distribution'] = \
                audio_features['pitch_distribution']
            sdict['audio']['pitch_class_distribution'] = \
                audio_features['pitch_class_distribution']
            # only add stable_notes if the joint features, hence the note
            # models are not given
            sdict['audio']['stable_notes'] = audio_features['stable_notes']

        if score_features is not None:
            sdict['audio']['makam'] = score_features['makam']['symbtr_slug']
        else:
            sdict['audio']['makam'] = audio_features['makam']

        # accumulate joint dict
        if joint_features is not None:
            sdict['joint']['sections'] = joint_features['sections']
            sdict['joint']['notes'] = joint_features['notes']
            sdict['joint']['note_models'] = joint_features['note_models']

        return sdict

    def _parse_and_extract_tonic_tempo(self, score_filename, score_data,
                                       audio_filename, audio_pitch,
                                       audio_tonic=None, audio_tempo=None):
        """
        Parses the tonic and tempo inputs in the JointAnalyzer.analyze and
        computes whichever is not supplied.
        :param score_filename:
        :param score_data:
        :param audio_filename:
        :param audio_pitch:
        :param audio_tempo:
        :param audio_tonic:
        :return:
        """
        if audio_tonic is None or audio_tempo is None:
            # the tonic or the tempo is not provided, call the extractor
            tonic, tempo = self.extract_tonic_tempo(
                score_filename, score_data, audio_filename, audio_pitch)
            audio_tonic = audio_tonic if audio_tonic is not None else tonic
            audio_tempo = audio_tempo if audio_tempo is not None else tempo

        return audio_tonic, audio_tempo

    def extract_tonic_tempo(self, score_filename='', score_data=None,
                            audio_filename='', audio_pitch=None):
        tic = timeit.default_timer()
        self.vprint(u"- Extracting score-informed tonic and tempo of {0:s}"
                    .format(audio_filename))

        # create the temporary input and output files wanted by the binary
        temp_score_data_file = IO.create_temp_file(
            '.json', json.dumps(score_data))

        # matlab
        matout = cStringIO()
        savemat(matout, audio_pitch)

        temp_pitch_file = IO.create_temp_file(
            '.mat', matout.getvalue())

        temp_out_folder = tempfile.mkdtemp()

{'fraction_non_alphanumeric': 0.08963499152933929, 'fraction_numerical': 0.00015401201293700908, 'mean_word_length': 2.6014975041597337, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '494786', 'n_tokens_mistral': 3629, 'n_tokens_neox': 3400, 'n_words': 755}