<s> from __future__ import ( <EOL> absolute_import , division , print_function , unicode_literals ) <EOL> import collections <EOL> import time <EOL> from mopidy import exceptions <EOL> from mopidy . audio import tags as tags_lib , utils <EOL> from mopidy . internal import encoding <EOL> from mopidy . internal . gi import Gst , GstPbutils <EOL> _DECODER = <NUM_LIT:1> << <NUM_LIT:0> <EOL> _AUDIO = <NUM_LIT:1> << <NUM_LIT:50> <EOL> _DEMUXER = <NUM_LIT:1> << <NUM_LIT:5> <EOL> _DEPAYLOADER = <NUM_LIT:1> << <NUM_LIT:8> <EOL> _PARSER = <NUM_LIT:1> << <NUM_LIT:6> <EOL> _SELECT_TRY = <NUM_LIT:0> <EOL> _SELECT_EXPOSE = <NUM_LIT:1> <EOL> _Result = collections . namedtuple ( <EOL> '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> class Scanner ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , timeout = <NUM_LIT:1000> , proxy_config = None ) : <EOL> self . _timeout_ms = int ( timeout ) <EOL> self . _proxy_config = proxy_config or { } <EOL> def scan ( self , uri , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> timeout = int ( timeout or self . _timeout_ms ) <EOL> tags , duration , seekable , mime = None , None , None , None <EOL> pipeline , signals = _setup_pipeline ( uri , self . _proxy_config ) <EOL> try : <EOL> _start_pipeline ( pipeline ) <EOL> tags , mime , have_audio , duration = _process ( pipeline , timeout ) <EOL> seekable = _query_seekable ( pipeline ) <EOL> finally : <EOL> signals . clear ( ) <EOL> pipeline . set_state ( Gst . State . NULL ) <EOL> del pipeline <EOL> return _Result ( uri , tags , duration , seekable , mime , have_audio ) <EOL> def _setup_pipeline ( uri , proxy_config = None ) : <EOL> src = Gst . Element . make_from_uri ( Gst . URIType . SRC , uri ) <EOL> if not src : <EOL> raise exceptions . ScannerError ( '<STR_LIT>' % uri ) <EOL> typefind = Gst . ElementFactory . make ( '<STR_LIT>' ) <EOL> decodebin = Gst . ElementFactory . make ( '<STR_LIT>' ) <EOL> pipeline = Gst . ElementFactory . make ( '<STR_LIT>' ) <EOL> for e in ( src , typefind , decodebin ) : <EOL> pipeline . add ( e ) <EOL> src . link ( typefind ) <EOL> typefind . link ( decodebin ) <EOL> if proxy_config : <EOL> utils . setup_proxy ( src , proxy_config ) <EOL> signals = utils . Signals ( ) <EOL> signals . connect ( typefind , '<STR_LIT>' , _have_type , decodebin ) <EOL> signals . connect ( decodebin , '<STR_LIT>' , _pad_added , pipeline ) <EOL> signals . connect ( decodebin , '<STR_LIT>' , _autoplug_select ) <EOL> return pipeline , signals <EOL> def _have_type ( element , probability , caps , decodebin ) : <EOL> decodebin . set_property ( '<STR_LIT>' , caps ) <EOL> struct = Gst . Structure . new_empty ( '<STR_LIT>' ) <EOL> struct . set_value ( '<STR_LIT>' , caps . get_structure ( <NUM_LIT:0> ) ) <EOL> element . get_bus ( ) . post ( Gst . Message . new_application ( element , struct ) ) <EOL> def _pad_added ( element , pad , pipeline ) : <EOL> sink = Gst . ElementFactory . make ( '<STR_LIT>' ) <EOL> sink . set_property ( '<STR_LIT>' , False ) <EOL> pipeline . add ( sink ) <EOL> sink . sync_state_with_parent ( ) <EOL> pad . link ( sink . get_static_pad ( '<STR_LIT>' ) ) <EOL> if pad . query_caps ( ) . is_subset ( Gst . Caps . from_string ( '<STR_LIT>' ) ) : <EOL> struct = Gst . Structure . new_empty ( '<STR_LIT>' ) <EOL> element . get_bus ( ) . post ( Gst . Message . new_application ( element , struct ) ) <EOL> def _autoplug_select ( element , pad , caps , factory ) : <EOL> if factory . list_is_type ( _DECODER | _AUDIO ) : <EOL> struct = Gst . Structure . new_empty ( '<STR_LIT>' ) <EOL> element . get_bus ( ) . post ( Gst . Message . new_application ( element , struct ) ) <EOL> if not factory . list_is_type ( _DEMUXER | _DEPAYLOADER | _PARSER ) : <EOL> return _SELECT_EXPOSE <EOL> return _SELECT_TRY <EOL> def _start_pipeline ( pipeline ) : <EOL> result = pipeline . set_state ( Gst . State . PAUSED ) <EOL> if result == Gst . StateChangeReturn . NO_PREROLL : <EOL> pipeline . set_state ( Gst . State . PLAYING ) <EOL> def _query_seekable ( pipeline ) : <EOL> query = Gst . Query . new_seeking ( Gst . Format . TIME ) <EOL> pipeline . query ( query ) <EOL> return query . parse_seeking ( ) [ <NUM_LIT:1> ] <EOL> def _process ( pipeline , timeout_ms ) : <EOL> bus = pipeline . get_bus ( ) <EOL> tags = { } <EOL> mime = None <EOL> have_audio = False <EOL> missing_message = None <EOL> duration = None <EOL> types = ( <EOL> Gst . MessageType . ELEMENT | <EOL> Gst . MessageType . APPLICATION | <EOL> Gst . MessageType . ERROR | <EOL> Gst . MessageType . EOS | <EOL> Gst . MessageType . ASYNC_DONE | <EOL> Gst . MessageType . DURATION_CHANGED | <EOL> Gst . MessageType . TAG <EOL> ) <EOL> timeout = timeout_ms <EOL> start = int ( time . time ( ) * <NUM_LIT:1000> ) <EOL> while timeout > <NUM_LIT:0> : <EOL> message = bus . timed_pop_filtered ( timeout * Gst . MSECOND , types ) <EOL> if message is None : <EOL> break <EOL> elif message . type == Gst . MessageType . ELEMENT : <EOL> if GstPbutils . is_missing_plugin_message ( message ) : <EOL> missing_message = message <EOL> elif message . type == Gst . MessageType . APPLICATION : <EOL> if message . get_structure ( ) . get_name ( ) == '<STR_LIT>' : <EOL> mime = message . get_structure ( ) . get_value ( '<STR_LIT>' ) . get_name ( ) <EOL> if mime and ( <EOL> mime . startswith ( '<STR_LIT>' ) or mime == '<STR_LIT>' ) : <EOL> return tags , mime , have_audio , duration <EOL> elif message . get_structure ( ) . get_name ( ) == '<STR_LIT>' : <EOL> have_audio = True <EOL> elif message . type == Gst . MessageType . ERROR : <EOL> error = encoding . locale_decode ( message . parse_error ( ) [ <NUM_LIT:0> ] ) <EOL> if missing_message and not mime : <EOL> caps = missing_message . get_structure ( ) . get_value ( '<STR_LIT>' ) <EOL> mime = caps . get_structure ( <NUM_LIT:0> ) . get_name ( ) <EOL> return tags , mime , have_audio , duration <EOL> raise exceptions . ScannerError ( error ) <EOL> elif message . type == Gst . MessageType . EOS : <EOL> return tags , mime , have_audio , duration <EOL> elif message . type == Gst . MessageType . ASYNC_DONE : <EOL> success , duration = pipeline . query_duration ( Gst . Format . TIME ) <EOL> if success : <EOL> duration = duration // Gst . MSECOND <EOL> else : <EOL> duration = None <EOL> if tags and duration is not None : <EOL> return tags , mime , have_audio , duration <EOL> result = pipeline . set_state ( Gst . State . PLAYING ) <EOL> if result == Gst . StateChangeReturn . FAILURE : <EOL> return tags , mime , have_audio , duration <EOL> elif message . type == Gst . MessageType . DURATION_CHANGED : <EOL> duration = <NUM_LIT:0> <EOL> elif message . type == Gst . MessageType . TAG : <EOL> taglist = message . parse_tag ( ) <EOL> tags . update ( tags_lib . convert_taglist ( taglist ) ) <EOL> timeout = timeout_ms - ( int ( time . time ( ) * <NUM_LIT:1000> ) - start ) <EOL> if tags and duration is not None : <EOL> pipeline . set_state ( Gst . State . PAUSED ) <EOL> raise exceptions . ScannerError ( '<STR_LIT>' % timeout_ms ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import os <EOL> import sys <EOL> from mopidy . internal import path <EOL> scanner = Scanner ( <NUM_LIT> ) <EOL> for uri in sys . argv [ <NUM_LIT:1> : ] : <EOL> if not Gst . uri_is_valid ( uri ) : <EOL> uri = path . path_to_uri ( os . path . abspath ( uri ) ) <EOL> try : <EOL> result = scanner . scan ( uri ) <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> print ( '<STR_LIT>' % ( key , getattr ( result , key ) ) ) <EOL> print ( '<STR_LIT>' ) <EOL> for tag , value in result . tags . items ( ) : <EOL> print ( '<STR_LIT>' % ( tag , value ) ) <EOL> except exceptions . ScannerError as error : <EOL> print ( '<STR_LIT>' % ( uri , error ) ) </s>
