source: python/lib/aubio/cmd.py @ bd72039

feature/autosinkfeature/constantqfeature/pitchshiftfeature/pydocstringsfeature/timestretchsampler
Last change on this file since bd72039 was bd72039, checked in by Paul Brossier <piem@piem.org>, 3 years ago

python/lib/aubio/cmd.py: add tempo subcommand to extract overall bpm

  • Property mode set to 100644
File size: 15.1 KB
Line 
1#! /usr/bin/env python
2# -*- coding: utf-8 -*-
3
4"""aubio command line tool
5
6This file was written by Paul Brossier <piem@aubio.org> and is released under
7the GNU/GPL v3.
8
9Note: this script is mostly about parsing command line arguments. For more
10readable code examples, check out the `python/demos` folder."""
11
12import sys
13import argparse
14import aubio
15
16def aubio_parser():
17    epilog = 'use "%(prog)s <command> --help" for more info about each command'
18    parser = argparse.ArgumentParser(epilog=epilog)
19    parser.add_argument('-V', '--version', help="show version",
20            action="store_true", dest="show_version")
21
22    subparsers = parser.add_subparsers(dest='command',
23            description="", metavar="<command>")
24
25    # onset subcommand
26    subparser = subparsers.add_parser('onset',
27            help='get onset times',
28            formatter_class = argparse.ArgumentDefaultsHelpFormatter)
29    parser_add_input(subparser)
30    parser_add_buf_hop_size(subparser)
31    helpstr = "onset novelty function"
32    helpstr += " <default|energy|hfc|complex|phase|specdiff|kl|mkl|specflux>"
33    parser_add_method(subparser, helpstr=helpstr)
34    parser_add_threshold(subparser)
35    parser_add_silence(subparser)
36    parser_add_minioi(subparser)
37    parser_add_time_format(subparser)
38    parser_add_verbose_help(subparser)
39    subparser.set_defaults(process=process_onset)
40
41    # pitch subcommand
42    subparser = subparsers.add_parser('pitch',
43            help='extract fundamental frequency')
44    parser_add_input(subparser)
45    parser_add_buf_hop_size(subparser, buf_size=2048)
46    helpstr = "pitch detection method <default|yinfft|yin|mcomb|fcomb|schmitt>"
47    parser_add_method(subparser, helpstr=helpstr)
48    parser_add_threshold(subparser)
49    parser_add_silence(subparser)
50    parser_add_time_format(subparser)
51    parser_add_verbose_help(subparser)
52    subparser.set_defaults(process=process_pitch)
53
54    # tempo subcommand
55    subparser = subparsers.add_parser('beat',
56            help='get locations of beats')
57    parser_add_input(subparser)
58    parser_add_buf_hop_size(subparser, buf_size=1024, hop_size=512)
59    parser_add_time_format(subparser)
60    parser_add_verbose_help(subparser)
61    subparser.set_defaults(process=process_beat)
62
63    # tempo subcommand
64    subparser = subparsers.add_parser('tempo',
65            help='get locations of beats')
66    parser_add_input(subparser)
67    parser_add_buf_hop_size(subparser, buf_size=1024, hop_size=512)
68    parser_add_time_format(subparser)
69    parser_add_verbose_help(subparser)
70    subparser.set_defaults(process=process_tempo)
71
72    # notes subcommand
73    subparser = subparsers.add_parser('notes',
74            help='get midi-like notes')
75    parser_add_input(subparser)
76    parser_add_buf_hop_size(subparser)
77    parser_add_time_format(subparser)
78    parser_add_verbose_help(subparser)
79    subparser.set_defaults(process=process_notes)
80
81    # mfcc subcommand
82    subparser = subparsers.add_parser('mfcc',
83            help='extract mel-frequency cepstrum coefficients')
84    parser_add_input(subparser)
85    parser_add_buf_hop_size(subparser)
86    parser_add_time_format(subparser)
87    parser_add_verbose_help(subparser)
88    subparser.set_defaults(process=process_mfcc)
89
90    # melbands subcommand
91    subparser = subparsers.add_parser('melbands',
92            help='extract mel-frequency energies per band')
93    parser_add_input(subparser)
94    parser_add_buf_hop_size(subparser)
95    parser_add_time_format(subparser)
96    parser_add_verbose_help(subparser)
97    subparser.set_defaults(process=process_melbands)
98
99    return parser
100
101def parser_add_input(parser):
102    parser.add_argument("source_uri", default=None, nargs='?',
103            help="input sound file to analyse", metavar = "<source_uri>")
104    parser.add_argument("-i", "--input", dest = "source_uri2",
105            help="input sound file to analyse", metavar = "<source_uri>")
106    parser.add_argument("-r", "--samplerate",
107            metavar = "<freq>", type=int,
108            action="store", dest="samplerate", default=0,
109            help="samplerate at which the file should be represented")
110
111def parser_add_verbose_help(parser):
112    parser.add_argument("-v","--verbose",
113            action="count", dest="verbose", default=1,
114            help="make lots of noise [default]")
115    parser.add_argument("-q","--quiet",
116            action="store_const", dest="verbose", const=0,
117            help="be quiet")
118
119def parser_add_buf_hop_size(parser, buf_size=512, hop_size=256):
120    parser.add_argument("-B","--bufsize",
121            action="store", dest="buf_size", default=buf_size,
122            metavar = "<size>", type=int,
123            help="buffer size [default=%d]" % buf_size)
124    parser.add_argument("-H","--hopsize",
125            metavar = "<size>", type=int,
126            action="store", dest="hop_size", default=hop_size,
127            help="overlap size [default=%d]" % hop_size)
128
129def parser_add_method(parser, method='default', helpstr='method'):
130    parser.add_argument("-m","--method",
131            metavar = "<method>", type=str,
132            action="store", dest="method", default=method,
133            help="%s [default=%s]" % (helpstr, method))
134
135def parser_add_threshold(parser, default=None):
136    parser.add_argument("-t","--threshold",
137            metavar = "<threshold>", type=float,
138            action="store", dest="threshold", default=default,
139            help="threshold [default=%s]" % default)
140
141def parser_add_silence(parser):
142    parser.add_argument("-s", "--silence",
143            metavar = "<value>", type=float,
144            action="store", dest="silence", default=-70,
145            help="silence threshold")
146
147def parser_add_minioi(parser):
148    parser.add_argument("-M", "--minioi",
149            metavar = "<value>", type=str,
150            action="store", dest="minioi", default="12ms",
151            help="minimum Inter-Onset Interval")
152
153def parser_add_time_format(parser):
154    helpstr = "select time values output format (samples, ms, seconds)"
155    helpstr += " [default=seconds]"
156    parser.add_argument("-T", "--time-format",
157             metavar='format',
158             dest="time_format",
159             default=None,
160             help=helpstr)
161
162# some utilities
163
164def parse_options(args, valid_opts):
165    options = {k :v for k,v in vars(args).items() if k in valid_opts}
166    return options
167
168def remap_pvoc_options(options):
169    # remap buf_size to win_s, hop_size to hop_s
170    # FIXME: adjust python/ext/py-phasevoc.c to understand buf_size/hop_size
171    options['win_s'] = options['buf_size']
172    del options['buf_size']
173    options['hop_s'] = options['hop_size']
174    del options['hop_size']
175    return options
176
177def samples2seconds(n_frames, samplerate):
178    return "%f\t" % (n_frames / float(samplerate))
179
180def samples2milliseconds(n_frames, samplerate):
181    return "%f\t" % (1000. * n_frames / float(samplerate))
182
183def samples2samples(n_frames, samplerate):
184    return "%d\t" % n_frames
185
186def timefunc(mode):
187    if mode is None or mode == 'seconds' or mode == 's':
188        return samples2seconds
189    elif mode == 'ms' or mode == 'milliseconds':
190        return samples2milliseconds
191    elif mode == 'samples':
192        return samples2samples
193    else:
194        raise ValueError('invalid time format %s' % mode)
195
196# definition of processing classes
197
198class default_process(object):
199    def __init__(self, args):
200        if 'time_format' in args:
201            self.time2string = timefunc(args.time_format)
202        if args.verbose > 2 and hasattr(self, 'options'):
203            name = type(self).__name__.split('_')[1]
204            optstr = ' '.join(['running', name, 'with options', repr(self.options), '\n'])
205            sys.stderr.write(optstr)
206    def flush(self, n_frames, samplerate):
207        pass
208
209class process_onset(default_process):
210    valid_opts = ['method', 'hop_size', 'buf_size', 'samplerate']
211    def __init__(self, args):
212        self.options = parse_options(args, self.valid_opts)
213        self.onset = aubio.onset(**self.options)
214        if args.threshold is not None:
215            self.onset.set_threshold(args.threshold)
216        if args.minioi:
217            if args.minioi.endswith('ms'):
218                self.onset.set_minioi_ms(float(args.minioi[:-2]))
219            elif args.minioi.endswith('s'):
220                self.onset.set_minioi_s(float(args.minioi[:-1]))
221            else:
222                self.onset.set_minioi(int(args.minioi))
223        if args.silence:
224            self.onset.set_silence(args.silence)
225        super(process_onset, self).__init__(args)
226    def __call__(self, block):
227        return self.onset(block)
228    def repr_res(self, res, frames_read, samplerate):
229        if res[0] != 0:
230            outstr = self.time2string(self.onset.get_last(), samplerate)
231            sys.stdout.write(outstr + '\n')
232
233class process_pitch(default_process):
234    valid_opts = ['method', 'hop_size', 'buf_size', 'samplerate']
235    def __init__(self, args):
236        self.options = parse_options(args, self.valid_opts)
237        self.pitch = aubio.pitch(**self.options)
238        if args.threshold is not None:
239            self.pitch.set_tolerance(args.threshold)
240        if args.silence is not None:
241            self.pitch.set_silence(args.silence)
242        super(process_pitch, self).__init__(args)
243    def __call__(self, block):
244        return self.pitch(block)
245    def repr_res(self, res, frames_read, samplerate):
246        fmt_out = self.time2string(frames_read, samplerate)
247        sys.stdout.write(fmt_out + "%.6f\n" % res[0])
248
249class process_beat(default_process):
250    valid_opts = ['method', 'hop_size', 'buf_size', 'samplerate']
251    def __init__(self, args):
252        self.options = parse_options(args, self.valid_opts)
253        self.tempo = aubio.tempo(**self.options)
254        super(process_beat, self).__init__(args)
255    def __call__(self, block):
256        return self.tempo(block)
257    def repr_res(self, res, frames_read, samplerate):
258        if res[0] != 0:
259            outstr = self.time2string(self.tempo.get_last(), samplerate)
260            sys.stdout.write(outstr + '\n')
261
262class process_tempo(process_beat):
263    def __init__(self, args):
264        super(process_tempo, self).__init__(args)
265        self.beat_locations = []
266    def repr_res(self, res, frames_read, samplerate):
267        if res[0] != 0:
268            self.beat_locations.append(self.tempo.get_last_s())
269    def flush(self, frames_read, samplerate):
270        import numpy as np
271        bpms = 60./ np.diff(self.beat_locations)
272        median_bpm = np.mean(bpms)
273        sys.stdout.write('%.2f bpm' % median_bpm + '\n')
274
275class process_notes(default_process):
276    valid_opts = ['method', 'hop_size', 'buf_size', 'samplerate']
277    def __init__(self, args):
278        self.options = parse_options(args, self.valid_opts)
279        self.notes = aubio.notes(**self.options)
280        super(process_notes, self).__init__(args)
281    def __call__(self, block):
282        return self.notes(block)
283    def repr_res(self, res, frames_read, samplerate):
284        if res[2] != 0: # note off
285            fmt_out = self.time2string(frames_read, samplerate)
286            sys.stdout.write(fmt_out + '\n')
287        if res[0] != 0: # note on
288            lastmidi = res[0]
289            fmt_out = "%f\t" % lastmidi
290            fmt_out += self.time2string(frames_read, samplerate)
291            sys.stdout.write(fmt_out) # + '\t')
292    def flush(self, frames_read, samplerate):
293        eof = self.time2string(frames_read, samplerate)
294        sys.stdout.write(eof + '\n')
295
296class process_mfcc(default_process):
297    def __init__(self, args):
298        valid_opts = ['hop_size', 'buf_size']
299        options = parse_options(args, valid_opts)
300        self.options = remap_pvoc_options(options)
301        self.pv = aubio.pvoc(**options)
302
303        valid_opts = ['buf_size', 'n_filters', 'n_coeffs', 'samplerate']
304        options = parse_options(args, valid_opts)
305        self.mfcc = aubio.mfcc(**options)
306        self.options.update(options)
307
308        super(process_mfcc, self).__init__(args)
309
310    def __call__(self, block):
311        fftgrain = self.pv(block)
312        return self.mfcc(fftgrain)
313    def repr_res(self, res, frames_read, samplerate):
314        fmt_out = self.time2string(frames_read, samplerate)
315        fmt_out += ' '.join(["% 9.7f" % f for f in res.tolist()])
316        sys.stdout.write(fmt_out + '\n')
317
318class process_melbands(default_process):
319    def __init__(self, args):
320        self.args = args
321        valid_opts = ['hop_size', 'buf_size']
322        options = parse_options(args, valid_opts)
323        options = remap_pvoc_options(options)
324        self.pv = aubio.pvoc(**options)
325
326        valid_opts = ['buf_size', 'n_filters']
327        options = {k :v for k,v in vars(args).items() if k in valid_opts}
328        # FIXME
329        options['win_s'] = options['buf_size']
330        del options['buf_size']
331        self.filterbank = aubio.filterbank(**options)
332        self.filterbank.set_mel_coeffs_slaney(args.samplerate)
333
334        super(process_melbands, self).__init__(args)
335    def __call__(self, block):
336        fftgrain = self.pv(block)
337        return self.filterbank(fftgrain)
338    def repr_res(self, res, frames_read, samplerate):
339        fmt_out = self.time2string(frames_read, samplerate)
340        fmt_out += ' '.join(["% 9.7f" % f for f in res.tolist()])
341        sys.stdout.write(fmt_out + '\n')
342
343def main():
344    parser = aubio_parser()
345    args = parser.parse_args()
346    if args.show_version or ('verbose' in args and args.verbose > 3):
347        sys.stdout.write('aubio version ' + aubio.version + '\n')
348    if args.show_version and args.command is None:
349        sys.exit(0)
350    if args.command is None:
351        sys.stderr.write("Error: a command is required\n")
352        parser.print_help()
353        sys.exit(1)
354    elif not args.source_uri and not args.source_uri2:
355        sys.stderr.write("Error: a source is required\n")
356        parser.print_help()
357        sys.exit(1)
358    elif args.source_uri2 is not None:
359        args.source_uri = args.source_uri2
360    try:
361        # open source_uri
362        with aubio.source(args.source_uri, hop_size=args.hop_size,
363                samplerate=args.samplerate) as a_source:
364            args.samplerate = a_source.samplerate
365            # create the processor for this subcommand
366            processor = args.process(args)
367            frames_read = 0
368            while True:
369                # read new block from source
370                block, read = a_source()
371                # execute processor on this block
372                res = processor(block)
373                # print results for this block
374                if args.verbose > 0:
375                    processor.repr_res(res, frames_read, a_source.samplerate)
376                # increment total number of frames read
377                frames_read += read
378                # exit loop at end of file
379                if read < a_source.hop_size: break
380            # flush the processor if needed
381            processor.flush(frames_read, a_source.samplerate)
382            if args.verbose > 1:
383                fmt_string = "read {:.2f}s"
384                fmt_string += " ({:d} samples in {:d} blocks of {:d})"
385                fmt_string += " from {:s} at {:d}Hz\n"
386                sys.stderr.write(fmt_string.format(
387                        frames_read/float(a_source.samplerate),
388                        frames_read,
389                        frames_read // a_source.hop_size + 1,
390                        a_source.hop_size,
391                        a_source.uri,
392                        a_source.samplerate))
393    except KeyboardInterrupt as e:
394        sys.exit(1)
Note: See TracBrowser for help on using the repository browser.