andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

audio: Cleanup scanner code and support live sources

Browse Source 
- Make attributes internal with _ naming
- Cleanup handling of min durtion. Min set to None disables the check.
- Check state change result for no preroll which indicates a live source which
  must transistion to playing to get tags etc.
This commit is contained in:
Thomas Adamcik 2013-12-04 23:41:39 +01:00
parent d7fc04e354
commit 10a448f90d
1 changed files with 32 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
mopidy/audio/scan.py
View File

@ -24,24 +24,24 @@ class Scanner(object):
""" """
def __init__(self, timeout=1000, min_duration=100): def __init__(self, timeout=1000, min_duration=100):
self.timeout_ms = timeout self._timeout_ms = timeout
self.min_duration_ms = min_duration self._min_duration_ms = min_duration
sink = gst.element_factory_make('fakesink') sink = gst.element_factory_make('fakesink')
audio_caps = gst.Caps(b'audio/x-raw-int; audio/x-raw-float') audio_caps = gst.Caps(b'audio/x-raw-int; audio/x-raw-float')
pad_added = lambda src, pad: pad.link(sink.get_pad('sink')) pad_added = lambda src, pad: pad.link(sink.get_pad('sink'))
self.uribin = gst.element_factory_make('uridecodebin') self._uribin = gst.element_factory_make('uridecodebin')
self.uribin.set_property('caps', audio_caps) self._uribin.set_property('caps', audio_caps)
self.uribin.connect('pad-added', pad_added) self._uribin.connect('pad-added', pad_added)
self.pipe = gst.element_factory_make('pipeline') self._pipe = gst.element_factory_make('pipeline')
self.pipe.add(self.uribin) self._pipe.add(self._uribin)
self.pipe.add(sink) self._pipe.add(sink)
self.bus = self.pipe.get_bus() self._bus = self._pipe.get_bus()
self.bus.set_flushing(True) self._bus.set_flushing(True)
def scan(self, uri): def scan(self, uri):
""" """
@ -54,34 +54,40 @@ class Scanner(object):
try: try:
self._setup(uri) self._setup(uri)
data = self._collect() data = self._collect()
# Make sure uri and duration does not come from tags. # Make sure uri, mtime and duration does not come from tags.
data[b'uri'] = uri data[b'uri'] = uri
data[b'mtime'] = self._query_mtime(uri) data[b'mtime'] = self._query_mtime(uri)
data[gst.TAG_DURATION] = self._query_duration() data[gst.TAG_DURATION] = self._query_duration()
finally: finally:
self._reset() self._reset()
if data[gst.TAG_DURATION] < self.min_duration_ms * gst.MSECOND: if self._min_duration_ms is None:
raise exceptions.ScannerError('Rejecting file with less than %dms ' return data
'audio data.' % self.min_duration_ms) elif data[gst.TAG_DURATION] >= self._min_duration_ms * gst.MSECOND:
return data return data
raise exceptions.ScannerError('Rejecting file with less than %dms '
'audio data.' % self._min_duration_ms)
def _setup(self, uri): def _setup(self, uri):
"""Primes the pipeline for collection.""" """Primes the pipeline for collection."""
self.pipe.set_state(gst.STATE_READY) self._pipe.set_state(gst.STATE_READY)
self.uribin.set_property(b'uri', uri) self._uribin.set_property(b'uri', uri)
self.bus.set_flushing(False) self._bus.set_flushing(False)
self.pipe.set_state(gst.STATE_PAUSED) result = self._pipe.set_state(gst.STATE_PAUSED)
if result == gst.STATE_CHANGE_NO_PREROLL:
# Live sources don't pre-roll, so set to playing to get data.
self._pipe.set_state(gst.STATE_PLAYING)
def _collect(self): def _collect(self):
"""Polls for messages to collect data.""" """Polls for messages to collect data."""
start = time.time() start = time.time()
timeout_s = self.timeout_ms / float(1000) timeout_s = self._timeout_ms / float(1000)
poll_timeout_ns = 1000 poll_timeout_ns = 1000
data = {} data = {}
while time.time() - start < timeout_s: while time.time() - start < timeout_s:
message = self.bus.poll(gst.MESSAGE_ANY, poll_timeout_ns) message = self._bus.poll(gst.MESSAGE_ANY, poll_timeout_ns)
if message is None: if message is None:
pass # polling the bus timed out. pass # polling the bus timed out.
@ -90,23 +96,23 @@ class Scanner(object):
elif message.type == gst.MESSAGE_EOS: elif message.type == gst.MESSAGE_EOS:
return data return data
elif message.type == gst.MESSAGE_ASYNC_DONE: elif message.type == gst.MESSAGE_ASYNC_DONE:
if message.src == self.pipe: if message.src == self._pipe:
return data return data
elif message.type == gst.MESSAGE_TAG: elif message.type == gst.MESSAGE_TAG:
taglist = message.parse_tag() taglist = message.parse_tag()
for key in taglist.keys(): for key in taglist.keys():
data[key] = taglist[key] data[key] = taglist[key]
raise exceptions.ScannerError('Timeout after %dms' % self.timeout_ms) raise exceptions.ScannerError('Timeout after %dms' % self._timeout_ms)
def _reset(self): def _reset(self):
"""Ensures we cleanup child elements and flush the bus.""" """Ensures we cleanup child elements and flush the bus."""
self.bus.set_flushing(True) self._bus.set_flushing(True)
self.pipe.set_state(gst.STATE_NULL) self._pipe.set_state(gst.STATE_NULL)
def _query_duration(self): def _query_duration(self):
try: try:
return self.pipe.query_duration(gst.FORMAT_TIME, None)[0] return self._pipe.query_duration(gst.FORMAT_TIME, None)[0]
except gst.QueryError: except gst.QueryError:
return None return None