<commit_before>import gobject
gobject.threads_init()

import pygst
pygst.require('0.10')
import gst

import logging
import threading

from mopidy.process import BaseProcess, unpickle_connection

logger = logging.getLogger('mopidy.outputs.gstreamer')

class GStreamerOutput(object):
    """
    Audio output through GStreamer.

    Starts the :class:`GStreamerProcess`.
    """

    def __init__(self, core_queue, output_queue):
        self.process = GStreamerProcess(core_queue, output_queue)
        self.process.start()

    def destroy(self):
        self.process.terminate()

class GStreamerMessagesThread(threading.Thread):
    def run(self):
        gobject.MainLoop().run()

class GStreamerProcess(BaseProcess):
    """
    A process for all work related to GStreamer.

    The main loop processes events from both Mopidy and GStreamer.

    Make sure this subprocess is started by the MainThread in the top-most
    parent process, and not some other thread. If not, we can get into the
    problems described at
    http://jameswestby.net/weblog/tech/14-caution-python-multiprocessing-and-glib-dont-mix.html.
    """

    def __init__(self, core_queue, output_queue):
        super(GStreamerProcess, self).__init__()
        self.core_queue = core_queue
        self.output_queue = output_queue
        self.gst_pipeline = None

    def run_inside_try(self):
        self.setup()
        while True:
            message = self.output_queue.get()
            self.process_mopidy_message(message)

    def setup(self):
        logger.debug(u'Setting up GStreamer pipeline')

        # Start a helper thread that can run the gobject.MainLoop
        messages_thread = GStreamerMessagesThread()
        messages_thread.daemon = True
        messages_thread.start()

        self.gst_pipeline = gst.parse_launch('playbin')

        # Setup bus and message processor
        gst_bus = self.gst_pipeline.get_bus()
        gst_bus.add_signal_watch()
        gst_bus.connect('message', self.process_gst_message)

    def process_mopidy_message(self, message):
        """Process messages from the rest of Mopidy."""
        if message['command'] == 'play_uri':
            response = self.play_uri(message['uri'])
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        elif message['command'] == 'deliver_data':
            self.deliver_data(message['caps'], message['data'])
        elif message['command'] == 'end_of_data_stream':
            self.end_of_data_stream()
        elif message['command'] == 'set_state':
            response = self.set_state(message['state'])
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        elif message['command'] == 'get_volume':
            volume = self.get_volume()
            connection = unpickle_connection(message['reply_to'])
            connection.send(volume)
        elif message['command'] == 'set_volume':
            self.set_volume(message['volume'])
        elif message['command'] == 'set_position':
            self.set_position(message['position'])
        elif message['command'] == 'get_position':
            response = self.get_position()
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        else:
            logger.warning(u'Cannot handle message: %s', message)

    def process_gst_message(self, bus, message):
        """Process messages from GStreamer."""
        if message.type == gst.MESSAGE_EOS:
            logger.debug(u'GStreamer signalled end-of-stream. '
                'Sending end_of_track to core_queue ...')
            self.core_queue.put({'command': 'end_of_track'})
        elif message.type == gst.MESSAGE_ERROR:
            self.set_state('NULL')
            error, debug = message.parse_error()
            logger.error(u'%s %s', error, debug)
            # FIXME Should we send 'stop_playback' to core here? Can we
            # differentiate on how serious the error is?

    def play_uri(self, uri):
        """Play audio at URI"""
        self.set_state('READY')
        self.gst_pipeline.set_property('uri', uri)
        return self.set_state('PLAYING')

    def deliver_data(self, caps_string, data):
        """Deliver audio data to be played"""
        data_src = self.gst_pipeline.get_by_name('src')
        caps = gst.caps_from_string(caps_string)
        buffer_ = gst.Buffer(buffer(data))
        buffer_.set_caps(caps)
        data_src.set_property('caps', caps)
        data_src.emit('push-buffer', buffer_)

    def end_of_data_stream(self):
        """
        Add end-of-stream token to source.

        We will get a GStreamer message when the stream playback reaches the
        token, and can then do any end-of-stream related tasks.
        """
        self.gst_pipeline.get_by_name('src').emit('end-of-stream')

    def set_state(self, state_name):
        """
        Set the GStreamer state. Returns :class:`True` if successful.<commit_msg>Use uridecodebin with auto connecting to play local music<commit_after>import gobject
gobject.threads_init()

import pygst
pygst.require('0.10')
import gst

import logging
import threading

from mopidy.process import BaseProcess, unpickle_connection

logger = logging.getLogger('mopidy.outputs.gstreamer')

class GStreamerOutput(object):
    """
    Audio output through GStreamer.

    Starts the :class:`GStreamerProcess`.
    """

    def __init__(self, core_queue, output_queue):
        self.process = GStreamerProcess(core_queue, output_queue)
        self.process.start()

    def destroy(self):
        self.process.terminate()

class GStreamerMessagesThread(threading.Thread):
    def run(self):
        gobject.MainLoop().run()

class GStreamerProcess(BaseProcess):
    """
    A process for all work related to GStreamer.

    The main loop processes events from both Mopidy and GStreamer.

    Make sure this subprocess is started by the MainThread in the top-most
    parent process, and not some other thread. If not, we can get into the
    problems described at
    http://jameswestby.net/weblog/tech/14-caution-python-multiprocessing-and-glib-dont-mix.html.
    """

    def __init__(self, core_queue, output_queue):
        super(GStreamerProcess, self).__init__()
        self.core_queue = core_queue
        self.output_queue = output_queue
        self.gst_pipeline = None
        self.gst_decoder = None

    def run_inside_try(self):
        self.setup()
        while True:
            message = self.output_queue.get()
            self.process_mopidy_message(message)

    def setup(self):
        logger.debug(u'Setting up GStreamer pipeline')

        # Start a helper thread that can run the gobject.MainLoop
        messages_thread = GStreamerMessagesThread()
        messages_thread.daemon = True
        messages_thread.start()

        self.gst_pipeline = gst.parse_launch(' ! '.join([
            'volume name=volume',
            'autoaudiosink'
        ]))

        decode_bin = gst.element_factory_make('uridecodebin', 'uri')
        decode_bin.connect('pad-added', self.process_new_pad, 'volume')
        app_src = gst.element_factory_make('appsrc', 'src')

        self.gst_pipeline.add(decode_bin)
        self.gst_pipeline.add(app_src)

        # Setup bus and message processor
        gst_bus = self.gst_pipeline.get_bus()
        gst_bus.add_signal_watch()
        gst_bus.connect('message', self.process_gst_message)

    def process_new_pad(self, source, pad, target):
        pad.link(self.gst_pipeline.get_by_name(target).get_pad('sink'))

    def process_mopidy_message(self, message):
        """Process messages from the rest of Mopidy."""
        if message['command'] == 'play_uri':
            response = self.play_uri(message['uri'])
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        elif message['command'] == 'deliver_data':
            self.deliver_data(message['caps'], message['data'])
        elif message['command'] == 'end_of_data_stream':
            self.end_of_data_stream()
        elif message['command'] == 'set_state':
            response = self.set_state(message['state'])
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        elif message['command'] == 'get_volume':
            volume = self.get_volume()
            connection = unpickle_connection(message['reply_to'])
            connection.send(volume)
        elif message['command'] == 'set_volume':
            self.set_volume(message['volume'])
        elif message['command'] == 'set_position':
            self.set_position(message['position'])
        elif message['command'] == 'get_position':
            response = self.get_position()
            connection = unpickle_connection(message['reply_to'])
            connection.send(response)
        else:
            logger.warning(u'Cannot handle message: %s', message)

    def process_gst_message(self, bus, message):
        """Process messages from GStreamer."""
        if message.type == gst.MESSAGE_EOS:
            logger.debug(u'GStreamer signalled end-of-stream. '
                'Sending end_of_track to core_queue ...')
            self.core_queue.put({'command': 'end_of_track'})
        elif message.type == gst.MESSAGE_ERROR:
            self.set_state('NULL')
            error, debug = message.parse_error()
            logger.error(u'%s %s', error, debug)
            # FIXME Should we send 'stop_playback' to core here? Can we
            # differentiate on how serious the error is?

    def play_uri(self, uri):
        """Play audio at URI"""
        self.set_state('READY')
        self.gst_pipeline.get_by_name('uri').set_property('uri', uri)
        return self.set_state('PLAYING')

    def deliver_data(self, caps_string, data):
        """Deliver audio data to be played"""
        data_src = self.gst_pipeline.get_by_name('src')
        caps = gst.caps_from_string(caps_string)
        buffer_ = gst.Buffer(buffer(data))
        buffer_.set_caps(caps)
        data_src.set_property('caps', caps)
        data_src.emit('push-buffer', buffer_)

    def end_of_data_stream(self):
        """
        Add end-of-stream token to source.

        We will get a GStreamer message when the stream playback reaches the
        token, and can then do any end-of-stream related tasks.
        """
        self.gst_pipeline.get_by_name('src').emit('end-of-stream')

    def set_state(self, state_name):
        """
        Set the GStreamer state. Returns :class:`True` if successful.
{'fraction_non_alphanumeric': 0.10925159834524258, 'fraction_numerical': 0.000940203083866115, 'mean_word_length': 2.729663394109397, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '940021', 'n_tokens_mistral': 3112, 'n_tokens_neox': 3073, 'n_words': 694}