<commit_before># -*- Mode: Python -*-
# vi:si:et:sw=4:sts=4:ts=4

# Flumotion - a streaming media server
# Copyright (C) 2004,2005,2006,2007,2008,2009 Fluendo, S.L.
# Copyright (C) 2010,2011 Flumotion Services, S.A.
# All rights reserved.
#
# This file may be distributed and/or modified under the terms of
# the GNU Lesser General Public License version 2.1 as published by
# the Free Software Foundation.
# This file is distributed without any warranty; without even the implied
# warranty of merchantability or fitness for a particular purpose.
# See "LICENSE.LGPL" in the source distribution for more information.
#
# Headers in this file shall remain intact.

import gst
from twisted.internet import defer
from flumotion.component import feedcomponent
from flumotion.twisted.defer import RetryingDeferred
from flumotion.common import errors

__version__ = "$Rev$"


class Icecast(feedcomponent.ParseLaunchComponent):

    configured = False

    def get_pipeline_string(self, properties):
        return "souphttpsrc name=src ! typefind name=tf"

    def _typefind_have_caps_cb(self, tf, prob, caps):
        # Basing on the cappabilities plug additional gst compoponents:
        # 1. If we have pure audio (http src doesn't support ICY) plug parser
        # 2. If we have application/x-icy plug the icydemuxer and than parser
        capsname = caps[0].get_name()
        tf_src_pad = tf.get_pad('src')
        gdp_sink_pad = tf_src_pad.get_peer()
        # unlink the typefind from the gdp pad so that we can put another
        # component in it's place
        tf_src_pad.unlink(gdp_sink_pad)

        if capsname == 'application/x-icy':
            demuxer = gst.element_factory_make("icydemux")
            demuxer.set_state(gst.STATE_PLAYING)
            self._demuxer_name = demuxer.get_name()
            self.pipeline.add(demuxer)
            tf.link(demuxer)
            # demuxer src pad is dynamic, we need to register a callback
            demuxer.connect('pad-added', self._link_parser, gdp_sink_pad)
        else:
            self._demuxer_name = None
            self._link_parser(tf, tf_src_pad, gdp_sink_pad)

    def _link_parser(self, element, pad, gdp_sink_pad):
        # Append the audio parser to the end of the pipeline
        caps = pad.get_caps()
        capsname = caps.get_structure(0).get_name()
        self._parser_name = None
        parser = None
        if self.passthrough:
            self.info("Acting in passthrough mode, not parsing the audio")
            pad.link(gdp_sink_pad)
            return
        if capsname == 'application/ogg':
            parser = gst.element_factory_make('oggparse')
        elif capsname == 'audio/mpeg':
            mpegversion = caps[0]['mpegversion']
            if mpegversion == 1:
                self.info("Detecting MP3 stream. Adding 'mp3parse'")
                parser = gst.element_factory_make('mp3parse')
            elif mpegversion in [2, 4]:
                self.info("Detecting AAC stream. Adding 'aacparse'")
                parser = gst.element_factory_make('aacparse')
        if parser:
            self._parser_name = parser.get_name()
            parser.set_state(gst.STATE_PLAYING)
            self.pipeline.add(parser)
            element.link(parser)
            parser.get_pad('src').link(gdp_sink_pad)
        else:
            # in case we good sth else than mp3 or ogg just connect the
            # gdb back
            self.warning("Couldn't find the correct parser for caps: %s",\
                capsname)
            pad.link(gdp_sink_pad)

    def configure_pipeline(self, pipeline, properties):
        # Later, when the typefind element has successfully found the type
        # of the data, we'll rebuild the pipeline.
        self.src = pipeline.get_by_name('src')
        self.url = properties['url']
        self.passthrough = properties.get('passthrough', False)
        self.src.set_property('location', self.url)
        self.src.set_property('iradio-mode', True)

        typefind = pipeline.get_by_name('tf')
        self.signal_id = typefind.connect('have-type',\
                self._typefind_have_caps_cb)

        if not self.configured:
            self.attachPadMonitorToElement('src',
                                           self._src_connected,
                                           self._src_disconnected)
        self.reconnecting = False
        self.reconnector = RetryingDeferred(self.connect)
        self.reconnector.initialDelay = 1.0
        self.attemptD = None

        def _drop_eos(pad, event):
            self.debug('Swallowing event %r', event)
            if event.type == gst.EVENT_EOS:
                return False
            return True
        self.configured = True
        self.src.get_pad('src').add_event_probe(_drop_eos)

    def bus_message_received_cb(self, bus, message):
        if message.type == gst.MESSAGE_ERROR and message.src == self.src:
            gerror, debug = message.parse_error()
            self.warning('element %s error %s %s',
                    message.src.get_path_string(), gerror, debug)
            if self.reconnecting:
                self._retry()
            return True
        feedcomponent.ParseLaunchComponent.bus_message_received_cb(
                self, bus, message)

    def connect(self):
        self.info('Connecting to icecast server on %s', self.url)
        self.src.set_state(gst.STATE_READY)
        # can't just self.src.set_state(gst.STATE_PLAYING),
        # because the pipeline might NOT be in PLAYING,
        # if we never connected to Icecast and never went to PLAYING
        self.try_start_pipeline(force=True)
        self.attemptD = defer.Deferred()
        return self.attemptD

    def _src_connected(self, name):
        self.info('Connected to icecast server on %s', self.url)
        if self.reconnecting:
            assert self.attemptD
            self.attemptD.callback(None)
            self.reconnecting = False

    def _reset(self, pad):
        # remove all the elements downstream souphttpsrc.
        tf = self.get_element('tf')<commit_msg>Do not reset the pipeline when there is no parser/demuxer preset<commit_after># -*- Mode: Python -*-
# vi:si:et:sw=4:sts=4:ts=4

# Flumotion - a streaming media server
# Copyright (C) 2004,2005,2006,2007,2008,2009 Fluendo, S.L.
# Copyright (C) 2010,2011 Flumotion Services, S.A.
# All rights reserved.
#
# This file may be distributed and/or modified under the terms of
# the GNU Lesser General Public License version 2.1 as published by
# the Free Software Foundation.
# This file is distributed without any warranty; without even the implied
# warranty of merchantability or fitness for a particular purpose.
# See "LICENSE.LGPL" in the source distribution for more information.
#
# Headers in this file shall remain intact.

import gst
from twisted.internet import defer
from flumotion.component import feedcomponent
from flumotion.twisted.defer import RetryingDeferred
from flumotion.common import errors

__version__ = "$Rev$"


class Icecast(feedcomponent.ParseLaunchComponent):

    configured = False

    def get_pipeline_string(self, properties):
        return "souphttpsrc name=src ! typefind name=tf"

    def _typefind_have_caps_cb(self, tf, prob, caps):
        # Basing on the cappabilities plug additional gst compoponents:
        # 1. If we have pure audio (http src doesn't support ICY) plug parser
        # 2. If we have application/x-icy plug the icydemuxer and than parser
        capsname = caps[0].get_name()
        tf_src_pad = tf.get_pad('src')
        gdp_sink_pad = tf_src_pad.get_peer()
        # unlink the typefind from the gdp pad so that we can put another
        # component in it's place
        tf_src_pad.unlink(gdp_sink_pad)

        if capsname == 'application/x-icy':
            demuxer = gst.element_factory_make("icydemux")
            demuxer.set_state(gst.STATE_PLAYING)
            self._demuxer_name = demuxer.get_name()
            self.pipeline.add(demuxer)
            tf.link(demuxer)
            # demuxer src pad is dynamic, we need to register a callback
            demuxer.connect('pad-added', self._link_parser, gdp_sink_pad)
        else:
            self._demuxer_name = None
            self._link_parser(tf, tf_src_pad, gdp_sink_pad)

    def _link_parser(self, element, pad, gdp_sink_pad):
        # Append the audio parser to the end of the pipeline
        caps = pad.get_caps()
        capsname = caps.get_structure(0).get_name()
        self._parser_name = None
        parser = None
        if self.passthrough:
            self.info("Acting in passthrough mode, not parsing the audio")
            pad.link(gdp_sink_pad)
            return
        if capsname == 'application/ogg':
            parser = gst.element_factory_make('oggparse')
        elif capsname == 'audio/mpeg':
            mpegversion = caps[0]['mpegversion']
            if mpegversion == 1:
                self.info("Detecting MP3 stream. Adding 'mp3parse'")
                parser = gst.element_factory_make('mp3parse')
            elif mpegversion in [2, 4]:
                self.info("Detecting AAC stream. Adding 'aacparse'")
                parser = gst.element_factory_make('aacparse')
        if parser:
            self._parser_name = parser.get_name()
            parser.set_state(gst.STATE_PLAYING)
            self.pipeline.add(parser)
            element.link(parser)
            parser.get_pad('src').link(gdp_sink_pad)
        else:
            # in case we good sth else than mp3 or ogg just connect the
            # gdb back
            self.warning("Couldn't find the correct parser for caps: %s",\
                capsname)
            pad.link(gdp_sink_pad)

    def configure_pipeline(self, pipeline, properties):
        # Later, when the typefind element has successfully found the type
        # of the data, we'll rebuild the pipeline.
        self.src = pipeline.get_by_name('src')
        self.url = properties['url']
        self.passthrough = properties.get('passthrough', False)
        self.src.set_property('location', self.url)
        self.src.set_property('iradio-mode', True)

        typefind = pipeline.get_by_name('tf')
        self.signal_id = typefind.connect('have-type',\
                self._typefind_have_caps_cb)

        if not self.configured:
            self.attachPadMonitorToElement('src',
                                           self._src_connected,
                                           self._src_disconnected)
        self.reconnecting = False
        self.reconnector = RetryingDeferred(self.connect)
        self.reconnector.initialDelay = 1.0
        self.attemptD = None

        def _drop_eos(pad, event):
            self.debug('Swallowing event %r', event)
            if event.type == gst.EVENT_EOS:
                return False
            return True
        self.configured = True
        self.src.get_pad('src').add_event_probe(_drop_eos)

    def bus_message_received_cb(self, bus, message):
        if message.type == gst.MESSAGE_ERROR and message.src == self.src:
            gerror, debug = message.parse_error()
            self.warning('element %s error %s %s',
                    message.src.get_path_string(), gerror, debug)
            if self.reconnecting:
                self._retry()
            return True
        feedcomponent.ParseLaunchComponent.bus_message_received_cb(
                self, bus, message)

    def connect(self):
        self.info('Connecting to icecast server on %s', self.url)
        self.src.set_state(gst.STATE_READY)
        # can't just self.src.set_state(gst.STATE_PLAYING),
        # because the pipeline might NOT be in PLAYING,
        # if we never connected to Icecast and never went to PLAYING
        self.try_start_pipeline(force=True)
        self.attemptD = defer.Deferred()
        return self.attemptD

    def _src_connected(self, name):
        self.info('Connected to icecast server on %s', self.url)
        if self.reconnecting:
            assert self.attemptD
            self.attemptD.callback(None)
            self.reconnecting = False

    def _reset(self, pad):
        # remove all the elements downstream souphttpsrc.
        if not self._parser_name:
            self.reconnecting = True
            self.reconnector.start()
            return

        tf = self.get_element('tf')
{'fraction_non_alphanumeric': 0.08743832403138396, 'fraction_numerical': 0.008250424654210144, 'mean_word_length': 2.289172652301144, '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': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5993498', 'n_tokens_mistral': 3696, 'n_tokens_neox': 3550, 'n_words': 960}