Skip to content

TAAT API Reference

QueryResult

TAAT QueryResult class. Instances of this class are not intended to be created directly, instead they are created and returned by the TAAT query function (see below).

Methods

export(output_filepath) Exports query result data to a file in JSON format.

plot() Creates and displays a plot of the query result scores.

pprint() Pretty prints the query result.

write(outdir, format='wav') Writes query result data to disk as audio files.

Source code in src/taat.py 
 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
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class QueryResult:
    """
    TAAT QueryResult class. Instances of this class are not intended to be created directly, instead they are created and returned by the TAAT _query_ function (see below).

    Methods
    -------

    export(output_filepath)
        Exports query result data to a file in JSON format.

    plot()
        Creates and displays a plot of the query result scores.

    pprint()
        Pretty prints the query result.

    write(outdir, format='wav')
        Writes query result data to disk as audio files.
    """

    def __init__(self, query_filepath, result, info):
        self.query_filepath = query_filepath
        self.info = info
        self.result = result

    def plot(self):
        """
        Creates and displays a plot of the query result scores.
        """
        result = self.result
        scores = {}
        ref_files = []
        for (_, val) in result.items():
            score = val["score"]
            ref_file = val["reference_file"]
            ref_files.append(ref_file)
            scores[ref_file] = score
        filename = os.path.basename(self.query_filepath)
        fig, ax = plt.subplots(nrows=1, ncols=1, figsize=[16,8])
        ax.plot(list(scores.values()), label="Path Cosine Scores")
        ax.legend()
        ax.set_xticks(range(0 ,len(ref_files)), labels=ref_files,
                      rotation=45, ha="right", rotation_mode="anchor")
        ax.set_ylabel("Scores")
        ax.set_title(f"Scores for query file '{filename}'")
        plt.tight_layout()
        plt.show()

    def pprint(self):
        """
        Pretty prints the query result.
        """
        pprint.pp(self.result)

    def export(self, output_filepath):
        """
        Export TAAT query results to a JSON file.

        Parameters
        ----------

        **_output_filepath_ (str)**: Path to which to export the data. Should have the '.json' file extension.
        """
        with open(output_filepath, "w") as f:
            print(f"Writing TAAT analysis data for {self.query_filepath} to {output_filepath}")
            json.dump(self.result, f, indent=3)

    def write(self, outdir, format="wav"):
        """
        Write matches to disk as audio files.

        Parameters
        ----------

        **_outdir_ (str)**: Path to folder into which to write the audio data. The folder will be created if it doesn't exist.

        **_format_ (str), optional**: Audio file format.
        """
        query_filepath = self.query_filepath
        matches = self.result

        source_dir = self.info["source_dir"]
        features = self.info["features"]
        sr = self.info["sr"]
        n_fft = self.info["n_fft"]
        hop_length = self.info["hop_length"]
        k = self.info["k"]
        metric = self.info["metric"]
        num_paths = self.info["num_paths"]

        if not os.path.exists(outdir):
            os.makedirs(outdir)
        info = dict(self.info)
        info["query_filepath"] = query_filepath
        with open(os.path.join(outdir, "info.json"), "w") as f:
            json.dump(info, f, indent=3)

        for match in matches.values():
            ref_filepath = os.path.join(source_dir, match["reference_file"])
            matches_dir = make_dir_for_file_path(outdir, ref_filepath)
            ref_segs = match["reference_segments"]
            query_segs = match["query_segments"]
            for (i, ref_seg) in enumerate(ref_segs):
                ref_start, ref_end = ref_seg
                query_start, query_end = query_segs[i]
                write_path_file(matches_dir, ref_filepath, f"ref_{i+1}", ref_start/1000, ref_end/1000, sr)
                write_path_file(matches_dir, query_filepath, f"query_{i+1}", query_start/1000, query_end/1000, sr)

export(output_filepath)

Export TAAT query results to a JSON file.

Parameters

output_filepath (str): Path to which to export the data. Should have the '.json' file extension.

Source code in src/taat.py 
66
67
68
69
70
71
72
73
74
75
76
77
def export(self, output_filepath):
    """
    Export TAAT query results to a JSON file.

    Parameters
    ----------

    **_output_filepath_ (str)**: Path to which to export the data. Should have the '.json' file extension.
    """
    with open(output_filepath, "w") as f:
        print(f"Writing TAAT analysis data for {self.query_filepath} to {output_filepath}")
        json.dump(self.result, f, indent=3)

plot()

Creates and displays a plot of the query result scores.

Source code in src/taat.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def plot(self):
    """
    Creates and displays a plot of the query result scores.
    """
    result = self.result
    scores = {}
    ref_files = []
    for (_, val) in result.items():
        score = val["score"]
        ref_file = val["reference_file"]
        ref_files.append(ref_file)
        scores[ref_file] = score
    filename = os.path.basename(self.query_filepath)
    fig, ax = plt.subplots(nrows=1, ncols=1, figsize=[16,8])
    ax.plot(list(scores.values()), label="Path Cosine Scores")
    ax.legend()
    ax.set_xticks(range(0 ,len(ref_files)), labels=ref_files,
                  rotation=45, ha="right", rotation_mode="anchor")
    ax.set_ylabel("Scores")
    ax.set_title(f"Scores for query file '{filename}'")
    plt.tight_layout()
    plt.show()

pprint()

Pretty prints the query result.

Source code in src/taat.py 
60
61
62
63
64
def pprint(self):
    """
    Pretty prints the query result.
    """
    pprint.pp(self.result)

write(outdir, format='wav')

Write matches to disk as audio files.

Parameters

outdir (str): Path to folder into which to write the audio data. The folder will be created if it doesn't exist.

format (str), optional: Audio file format.

Source code in src/taat.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def write(self, outdir, format="wav"):
    """
    Write matches to disk as audio files.

    Parameters
    ----------

    **_outdir_ (str)**: Path to folder into which to write the audio data. The folder will be created if it doesn't exist.

    **_format_ (str), optional**: Audio file format.
    """
    query_filepath = self.query_filepath
    matches = self.result

    source_dir = self.info["source_dir"]
    features = self.info["features"]
    sr = self.info["sr"]
    n_fft = self.info["n_fft"]
    hop_length = self.info["hop_length"]
    k = self.info["k"]
    metric = self.info["metric"]
    num_paths = self.info["num_paths"]

    if not os.path.exists(outdir):
        os.makedirs(outdir)
    info = dict(self.info)
    info["query_filepath"] = query_filepath
    with open(os.path.join(outdir, "info.json"), "w") as f:
        json.dump(info, f, indent=3)

    for match in matches.values():
        ref_filepath = os.path.join(source_dir, match["reference_file"])
        matches_dir = make_dir_for_file_path(outdir, ref_filepath)
        ref_segs = match["reference_segments"]
        query_segs = match["query_segments"]
        for (i, ref_seg) in enumerate(ref_segs):
            ref_start, ref_end = ref_seg
            query_start, query_end = query_segs[i]
            write_path_file(matches_dir, ref_filepath, f"ref_{i+1}", ref_start/1000, ref_end/1000, sr)
            write_path_file(matches_dir, query_filepath, f"query_{i+1}", query_start/1000, query_end/1000, sr)

query(source_dir, query_filepath, backend='cross_similarity', features=['melspectrogram'], sr=16000, n_fft=2048, hop_length=1024, k=5, metric='cosine', num_paths=5, no_identity_match=True, verbose=False)

Extracts feature data from the audio file supplied in query_filepath and attempts to match it using cross-similarity scores with the audio files supplied in source_dir.

Parameters

source_dir (str): Path to the folder of files whose feature data will be extracted.

query_filepath (str): Path to the file to be queried against source_dir.

backend (str): Set to one of 'cross_similarity' or 'dtw' (dynamic time warping).

features (list[str]), optional: List of features to extract in the analysis. Available features are: stft, melspectrogram, chroma_cqt, mfcc, rms, tempogram, spectral_centroid, spectral_flatness, spectral_bandwidth and spectral_contrast.

sr (int), optional: Sample rate for the audio loaded for the analysis.

n_fft (int), optional: FFT analysis frame size.

hop_length (int), optional: FFT analysis hop length.

k (int), optional: Number of nearest-neighbours to compute for each analysis sample.

metric (str), optional: Distance metric to use for the cross-similarity analysis.

num_paths (int), optional: Number of RQA paths to compute.

no_identity_match (bool), optional: Whether or not to include the queried file in the result, if it is itself already present in the source folder.

Returns

query_result (QueryResult class instance)

Source code in src/taat.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def query(source_dir, query_filepath, backend="cross_similarity", features=["melspectrogram"], sr=16000, n_fft=2048, hop_length=1024, k=5, metric="cosine", num_paths=5, no_identity_match=True, verbose=False):
    """
    Extracts feature data from the audio file supplied in _query_filepath_ and attempts to match it using cross-similarity scores with the audio files supplied in _source_dir_.

    Parameters
    ----------

    **_source_dir_ (str)**: Path to the folder of files whose feature data will be extracted.

    **_query_filepath_ (str)**: Path to the file to be queried against _source_dir_.

    **_backend_ (str)**: Set to one of 'cross_similarity' or 'dtw' (dynamic time warping).

    **_features_ (list[str]), optional**: List of features to extract in the analysis. Available features are: stft, melspectrogram, chroma_cqt, mfcc, rms, tempogram, spectral_centroid, spectral_flatness, spectral_bandwidth and spectral_contrast.

    **_sr_ (int), optional**: Sample rate for the audio loaded for the analysis.

    **_n_fft_ (int), optional**: FFT analysis frame size.

    **_hop_length_ (int), optional**: FFT analysis hop length.

    **_k_ (int), optional**: Number of nearest-neighbours to compute for each analysis sample.

    **_metric_ (str), optional**: Distance metric to use for the cross-similarity analysis.

    **_num_paths_ (int), optional**: Number of RQA paths to compute.

    **_no_identity_match_ (bool), optional**: Whether or not to include the queried file in the result, if it is itself already present in the source folder.

    Returns
    -------

    **_query_result_ (QueryResult class instance)**
    """
    for dirpath, dirnames, filenames in os.walk(source_dir):
        matches = {}
        for filename in filenames:
            if filename.endswith(".wav"):
                if (no_identity_match and filename != os.path.basename(query_filepath)) or (not no_identity_match):
                    ref_filepath = os.path.join(dirpath, filename)
                    ref_xsim, ref_rqa, ref_paths = run_backend(ref_filepath, ref_filepath, backend=backend, features=features, sr=sr, n_fft=n_fft, hop_length=hop_length, k=k, metric=metric, num_paths=num_paths, enhance=True)
                    query_xsim, query_rqa, query_paths = run_backend(ref_filepath, query_filepath, backend=backend, features=features, sr=sr, n_fft=n_fft, hop_length=hop_length, k=k, metric=metric, num_paths=num_paths, enhance=True)
                    paths, _ = get_time_formatted_paths(query_paths, n_fft=n_fft, hop_length=hop_length)
                    for (i, (ref_start, ref_stop, query_start, query_stop)) in enumerate(paths):
                        match = {
                            "score": get_path_score(ref_rqa, query_rqa, ref_paths[i], query_paths[i]),
                            "queryStart": query_start,
                            "queryStop": query_stop,
                            "referenceStart": ref_start,
                            "referenceStop": ref_stop,
                        }
                        if ref_filepath not in matches:
                            matches[ref_filepath] = [match]
                        else:
                            matches[ref_filepath].append(match)
        info = {
            "source_dir": source_dir,
            "features": features,
            "sr": sr,
            "n_fft": n_fft,
            "hop_length": hop_length,
            "k": k,
            "metric": metric,
            "num_paths": num_paths
        }
        if verbose:
            return QueryResult(query_filepath=query_filepath,
                               result=matches,
                               info=info)
        else:
            parsed_result = parse_query_output(query_filepath, matches)
            return QueryResult(query_filepath=query_filepath,
                               result=parsed_result,
                               info=info)