<s> from __future__ import absolute_import , unicode_literals <EOL> from mopidy import httpclient <EOL> from mopidy . internal . gi import Gst <EOL> def calculate_duration ( num_samples , sample_rate ) : <EOL> """<STR_LIT>""" <EOL> return Gst . util_uint64_scale ( num_samples , Gst . SECOND , sample_rate ) <EOL> def create_buffer ( data , timestamp = None , duration = None ) : <EOL> """<STR_LIT>""" <EOL> if not data : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> buffer_ = Gst . Buffer . new_wrapped ( data ) <EOL> if timestamp is not None : <EOL> buffer_ . pts = timestamp <EOL> if duration is not None : <EOL> buffer_ . duration = duration <EOL> return buffer_ <EOL> def millisecond_to_clocktime ( value ) : <EOL> """<STR_LIT>""" <EOL> return value * Gst . MSECOND <EOL> def clocktime_to_millisecond ( value ) : <EOL> """<STR_LIT>""" <EOL> return value // Gst . MSECOND <EOL> def supported_uri_schemes ( uri_schemes ) : <EOL> """<STR_LIT>""" <EOL> supported_schemes = set ( ) <EOL> registry = Gst . Registry . get ( ) <EOL> for factory in registry . get_feature_list ( Gst . ElementFactory ) : <EOL> for uri in factory . get_uri_protocols ( ) : <EOL> if uri in uri_schemes : <EOL> supported_schemes . add ( uri ) <EOL> return supported_schemes <EOL> def setup_proxy ( element , config ) : <EOL> """<STR_LIT>""" <EOL> if not hasattr ( element . props , '<STR_LIT>' ) or not config . get ( '<STR_LIT>' ) : <EOL> return <EOL> element . set_property ( '<STR_LIT>' , httpclient . format_proxy ( config , auth = False ) ) <EOL> element . set_property ( '<STR_LIT>' , config . get ( '<STR_LIT:username>' ) ) <EOL> element . set_property ( '<STR_LIT>' , config . get ( '<STR_LIT:password>' ) ) <EOL> class Signals ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _ids = { } <EOL> def connect ( self , element , event , func , * args ) : <EOL> """<STR_LIT>""" <EOL> assert ( element , event ) not in self . _ids <EOL> self . _ids [ ( element , event ) ] = element . connect ( event , func , * args ) <EOL> def disconnect ( self , element , event ) : <EOL> """<STR_LIT>""" <EOL> signal_id = self . _ids . pop ( ( element , event ) , None ) <EOL> if signal_id is not None : <EOL> element . disconnect ( signal_id ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> for element , event in self . _ids . keys ( ) : <EOL> element . disconnect ( self . _ids . pop ( ( element , event ) ) ) </s>
