Source code for mopidy.audio.scan

from __future__ import (
    absolute_import, division, print_function, unicode_literals)

import collections
import time

from mopidy import exceptions
from mopidy.audio import tags as tags_lib, utils
from mopidy.internal import encoding
from mopidy.internal.gi import Gst, GstPbutils

# GST_ELEMENT_FACTORY_LIST:
_DECODER = 1 << 0
_AUDIO = 1 << 50
_DEMUXER = 1 << 5
_DEPAYLOADER = 1 << 8
_PARSER = 1 << 6

# GST_TYPE_AUTOPLUG_SELECT_RESULT:
_SELECT_TRY = 0
_SELECT_EXPOSE = 1

_Result = collections.namedtuple(
    'Result', ('uri', 'tags', 'duration', 'seekable', 'mime', 'playable'))


# TODO: replace with a scan(uri, timeout=1000, proxy_config=None)?
[docs]class Scanner(object): """ Helper to get tags and other relevant info from URIs. :param timeout: timeout for scanning a URI in ms :param proxy_config: dictionary containing proxy config strings. :type event: int """ def __init__(self, timeout=1000, proxy_config=None): self._timeout_ms = int(timeout) self._proxy_config = proxy_config or {}
[docs] def scan(self, uri, timeout=None): """ Scan the given uri collecting relevant metadata. :param uri: URI of the resource to scan. :type uri: string :param timeout: timeout for scanning a URI in ms. Defaults to the ``timeout`` value used when creating the scanner. :type timeout: int :return: A named tuple containing ``(uri, tags, duration, seekable, mime)``. ``tags`` is a dictionary of lists for all the tags we found. ``duration`` is the length of the URI in milliseconds, or :class:`None` if the URI has no duration. ``seekable`` is boolean. indicating if a seek would succeed. """ timeout = int(timeout or self._timeout_ms) tags, duration, seekable, mime = None, None, None, None pipeline, signals = _setup_pipeline(uri, self._proxy_config) try: _start_pipeline(pipeline) tags, mime, have_audio = _process(pipeline, timeout) duration = _query_duration(pipeline) seekable = _query_seekable(pipeline) finally: signals.clear() pipeline.set_state(Gst.State.NULL) del pipeline return _Result(uri, tags, duration, seekable, mime, have_audio)
# Turns out it's _much_ faster to just create a new pipeline for every as # decodebins and other elements don't seem to take well to being reused. def _setup_pipeline(uri, proxy_config=None): src = Gst.Element.make_from_uri(Gst.URIType.SRC, uri) if not src: raise exceptions.ScannerError('GStreamer can not open: %s' % uri) typefind = Gst.ElementFactory.make('typefind') decodebin = Gst.ElementFactory.make('decodebin') pipeline = Gst.ElementFactory.make('pipeline') for e in (src, typefind, decodebin): pipeline.add(e) src.link(typefind) typefind.link(decodebin) if proxy_config: utils.setup_proxy(src, proxy_config) signals = utils.Signals() signals.connect(typefind, 'have-type', _have_type, decodebin) signals.connect(decodebin, 'pad-added', _pad_added, pipeline) signals.connect(decodebin, 'autoplug-select', _autoplug_select) return pipeline, signals def _have_type(element, probability, caps, decodebin): decodebin.set_property('sink-caps', caps) struct = Gst.Structure.new_empty('have-type') struct.set_value('caps', caps.get_structure(0)) element.get_bus().post(Gst.Message.new_application(element, struct)) def _pad_added(element, pad, pipeline): sink = Gst.ElementFactory.make('fakesink') sink.set_property('sync', False) pipeline.add(sink) sink.sync_state_with_parent() pad.link(sink.get_static_pad('sink')) if pad.query_caps().is_subset(Gst.Caps.from_string('audio/x-raw')): # Probably won't happen due to autoplug-select fix, but lets play it # safe until we've tested more. struct = Gst.Structure.new_empty('have-audio') element.get_bus().post(Gst.Message.new_application(element, struct)) def _autoplug_select(element, pad, caps, factory): if factory.list_is_type(_DECODER | _AUDIO): struct = Gst.Structure.new_empty('have-audio') element.get_bus().post(Gst.Message.new_application(element, struct)) if not factory.list_is_type(_DEMUXER | _DEPAYLOADER | _PARSER): return _SELECT_EXPOSE return _SELECT_TRY def _start_pipeline(pipeline): result = pipeline.set_state(Gst.State.PAUSED) if result == Gst.StateChangeReturn.NO_PREROLL: pipeline.set_state(Gst.State.PLAYING) def _query_duration(pipeline, timeout=100): # 1. Try and get a duration, return if success. # 2. Some formats need to play some buffers before duration is found. # 3. Wait for a duration change event. # 4. Try and get a duration again. success, duration = pipeline.query_duration(Gst.Format.TIME) if success and duration >= 0: return duration // Gst.MSECOND result = pipeline.set_state(Gst.State.PLAYING) if result == Gst.StateChangeReturn.FAILURE: return None gst_timeout = timeout * Gst.MSECOND bus = pipeline.get_bus() bus.timed_pop_filtered(gst_timeout, Gst.MessageType.DURATION_CHANGED) success, duration = pipeline.query_duration(Gst.Format.TIME) if success and duration >= 0: return duration // Gst.MSECOND return None def _query_seekable(pipeline): query = Gst.Query.new_seeking(Gst.Format.TIME) pipeline.query(query) return query.parse_seeking()[1] def _process(pipeline, timeout_ms): bus = pipeline.get_bus() tags = {} mime = None have_audio = False missing_message = None types = ( Gst.MessageType.ELEMENT | Gst.MessageType.APPLICATION | Gst.MessageType.ERROR | Gst.MessageType.EOS | Gst.MessageType.ASYNC_DONE | Gst.MessageType.TAG ) timeout = timeout_ms previous = int(time.time() * 1000) while timeout > 0: message = bus.timed_pop_filtered(timeout * Gst.MSECOND, types) if message is None: break elif message.type == Gst.MessageType.ELEMENT: if GstPbutils.is_missing_plugin_message(message): missing_message = message elif message.type == Gst.MessageType.APPLICATION: if message.get_structure().get_name() == 'have-type': mime = message.get_structure().get_value('caps').get_name() if mime and ( mime.startswith('text/') or mime == 'application/xml'): return tags, mime, have_audio elif message.get_structure().get_name() == 'have-audio': have_audio = True elif message.type == Gst.MessageType.ERROR: error = encoding.locale_decode(message.parse_error()[0]) if missing_message and not mime: caps = missing_message.get_structure().get_value('detail') mime = caps.get_structure(0).get_name() return tags, mime, have_audio raise exceptions.ScannerError(error) elif message.type == Gst.MessageType.EOS: return tags, mime, have_audio elif message.type == Gst.MessageType.ASYNC_DONE: if message.src == pipeline: return tags, mime, have_audio elif message.type == Gst.MessageType.TAG: taglist = message.parse_tag() # Note that this will only keep the last tag. tags.update(tags_lib.convert_taglist(taglist)) now = int(time.time() * 1000) timeout -= now - previous previous = now raise exceptions.ScannerError('Timeout after %dms' % timeout_ms) if __name__ == '__main__': import os import sys from mopidy.internal import path scanner = Scanner(5000) for uri in sys.argv[1:]: if not Gst.uri_is_valid(uri): uri = path.path_to_uri(os.path.abspath(uri)) try: result = scanner.scan(uri) for key in ('uri', 'mime', 'duration', 'playable', 'seekable'): print('%-20s %s' % (key, getattr(result, key))) print('tags') for tag, value in result.tags.items(): print('%-20s %s' % (tag, value)) except exceptions.ScannerError as error: print('%s: %s' % (uri, error))