Skip to content

Instantly share code, notes, and snippets.

@gluneau
Forked from virtuald/bpmdetect.py
Last active October 5, 2022 10:53
Show Gist options
  • Save gluneau/a5b53448f9a9f1202c374b7fc01fe5ec to your computer and use it in GitHub Desktop.
Save gluneau/a5b53448f9a9f1202c374b7fc01fe5ec to your computer and use it in GitHub Desktop.
GStreamer BPM Detector frontend
#!/usr/bin/env python
'''
Simple program that uses the 'bpmdetect' GStreamer plugin to detect
the BPM of a song, and outputs that to console.
Requires GStreamer 1.x, PyGObject 1.x, and gst-plugins-bad
Copyright (C) 2015 Dustin Spicuzza
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License version 2 as
published by the Free Software Foundation.
'''
from __future__ import print_function
import inspect
import sys
import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GObject, Gio, GLib
def detect_bpm(uri, on_complete):
'''
Detects the BPM of a song using GStreamer's bpmdetect plugin
.. note:: The plugin emits the BPM at various times during
song processing, but the bpm detector accumulates
the results so this will only return the last
result.
'''
bpm = [None]
def _on_message(bus, msg):
if msg.type == Gst.MessageType.TAG:
tags = msg.parse_tag()
# Discard tags already set on the file
if tags.n_tags() > 1:
return
v = tags.get_value_index('beats-per-minute', 0)
try:
v = float(v)
except:
return
if v > 0:
bpm[0] = v
elif msg.type == Gst.MessageType.ERROR:
playbin.set_state(Gst.State.NULL)
gerror, debug_info = msg.parse_error()
if gerror:
on_complete(None, gerror.message.rstrip("."))
else:
on_complete(None, debug_info)
elif msg.type == Gst.MessageType.EOS:
playbin.set_state(Gst.State.NULL)
on_complete(bpm[0], None)
audio_sink = Gst.Bin.new('audiosink')
# bpmdetect doesn't work properly with more than one channel,
# see https://bugzilla.gnome.org/show_bug.cgi?id=751457
cf = Gst.ElementFactory.make('capsfilter')
cf.props.caps = Gst.Caps.from_string('audio/x-raw,channels=1')
fakesink = Gst.ElementFactory.make('fakesink')
fakesink.props.sync = False
fakesink.props.signal_handoffs = False
bpmdetect = Gst.ElementFactory.make('bpmdetect')
if bpmdetect is None:
on_complete(None, "GStreamer BPM detection plugin not found")
return
audio_sink.add(cf)
audio_sink.add(bpmdetect)
audio_sink.add(fakesink)
cf.link(bpmdetect)
bpmdetect.link(fakesink)
audio_sink.add_pad(Gst.GhostPad.new('sink', cf.get_static_pad('sink')))
playbin = Gst.ElementFactory.make('playbin')
playbin.props.audio_sink = audio_sink
bus = playbin.get_bus()
bus.add_signal_watch()
bus.connect('message', _on_message)
playbin.props.uri = uri
playbin.set_state(Gst.State.PLAYING)
if __name__ == '__main__':
Gst.init(None)
if len(sys.argv) != 2:
print("Usage: %s filename\n" % sys.argv[0])
print(inspect.cleandoc(__doc__))
exit(1)
uri = Gio.File.new_for_commandline_arg(sys.argv[1]).get_uri()
print("Processing file: ", uri)
def on_complete(bpm, error):
if error is not None:
print("ERROR:", error)
else:
print("BPM:", bpm)
ml.quit()
detect_bpm(uri, on_complete)
ml = GLib.MainLoop()
ml.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment