CC-1469: Crossfading support (non-equal power)

-refactor

python_apps/pypo/pypoliquidsoap.py

@@ -0,0 +1,214 @@
+from pypofetch import PypoFetch
+from telnetliquidsoap import TelnetLiquidsoap
+
+from datetime import datetime
+from datetime import timedelta
+
+import eventtypes
+import time
+
+class PypoLiquidsoap():
+    def __init__(self, logger, telnet_lock, host, port):
+        self.logger = logger
+        self.liq_queue_tracker = {
+                "s0": None,
+                "s1": None,
+                "s2": None,
+                "s3": None,
+                }
+
+        self.telnet_liquidsoap = TelnetLiquidsoap(telnet_lock, \
+                logger,\
+                host,\
+                port)
+
+
+    def push_item(self, media_item):
+        if media_item["type"] == eventtypes.FILE:
+            self.handle_file_type(media_item)
+        elif media_item["type"] == eventtypes.EVENT:
+            self.handle_event_type(media_item)
+        elif media_item["type"] == eventtypes.STREAM_BUFFER_START:
+            self.telnet_liquidsoap.start_web_stream_buffer(media_item)
+        elif media_item["type"] == eventtypes.STREAM_OUTPUT_START:
+            if media_item['row_id'] != self.telnet_liquidsoap.current_prebuffering_stream_id:
+                #this is called if the stream wasn't scheduled sufficiently ahead of time
+                #so that the prebuffering stage could take effect. Let's do the prebuffering now.
+                self.telnet_liquidsoap.start_web_stream_buffer(media_item)
+            self.telnet_liquidsoap.start_web_stream(media_item)
+        elif media_item['type'] == eventtypes.STREAM_BUFFER_END:
+            self.telnet_liquidsoap.stop_web_stream_buffer()
+        elif media_item['type'] == eventtypes.STREAM_OUTPUT_END:
+            self.telnet_liquidsoap.stop_web_stream_output()
+        else: raise UnknownMediaItemType(str(media_item))
+
+    def handle_file_type(self, media_item):
+        """
+        Wait maximum 5 seconds (50 iterations) for file to become ready, 
+        otherwise give up on it.
+        """
+        iter_num = 0
+        while not media_item['file_ready'] and iter_num < 50:
+            time.sleep(0.1)
+            iter_num += 1
+
+        if media_item['file_ready']:
+            available_queue = self.find_available_queue()
+
+            try:
+                self.telnet_liquidsoap.queue_push(available_queue, media_item)
+                self.liq_queue_tracker[available_queue] = media_item
+            except Exception as e:
+                self.logger.error(e)
+                raise
+        else:
+            self.logger.warn("File %s did not become ready in less than 5 seconds. Skipping...", media_item['dst'])
+
+    def handle_event_type(self, media_item):
+        if media_item['event_type'] == "kick_out":
+            PypoFetch.disconnect_source(self.logger, self.telnet_lock, "live_dj")
+        elif media_item['event_type'] == "switch_off":
+            PypoFetch.switch_source(self.logger, self.telnet_lock, "live_dj", "off")
+
+
+    def is_media_item_finished(self, media_item):
+        if media_item is None:
+            return True
+        else:
+            return datetime.utcnow() > media_item['end']
+
+    def find_available_queue(self):
+        available_queue = None
+        for i in self.liq_queue_tracker:
+            mi = self.liq_queue_tracker[i]
+            if mi == None or self.is_media_item_finished(mi):
+                #queue "i" is available. Push to this queue
+                available_queue = i
+
+        if available_queue == None:
+            raise NoQueueAvailableException()
+
+        return available_queue
+
+    def get_queues():
+        return self.liq_queue_tracker
+
+
+    def verify_correct_present_media(self, scheduled_now):
+        #verify whether Liquidsoap is currently playing the correct files.
+        #if we find an item that Liquidsoap is not playing, then push it
+        #into one of Liquidsoap's queues. If Liquidsoap is already playing
+        #it do nothing. If Liquidsoap is playing a track that isn't in
+        #currently_playing then stop it.
+
+        #Check for Liquidsoap media we should source.skip
+        #get liquidsoap items for each queue. Since each queue can only have one
+        #item, we should have a max of 8 items.
+
+        #TODO: Verify start, end, replay_gain is the same
+        #TODO: Verify this is a file or webstream and also handle webstreams
+        #TODO: Do webstreams span a time or do they have an instantaneous point?
+
+        #2013-03-21-22-56-00_0: {
+        #id: 1,
+        #type: "stream_output_start",
+        #row_id: 41,
+        #uri: "http://stream2.radioblackout.org:80/blackout.ogg",
+        #start: "2013-03-21-22-56-00",
+        #end: "2013-03-21-23-26-00",
+        #show_name: "Untitled Show",
+        #independent_event: true
+        #},
+
+
+        scheduled_now_files = \
+                filter(lambda x: x["type"] == eventtypes.FILE, scheduled_now)
+
+        scheduled_now_webstream = \
+                filter(lambda x: x["type"] == eventtypes.STREAM_OUTPUT_START, \
+                        scheduled_now)
+
+        schedule_ids = set(map(lambda x: i["row_id"], scheduled_now_files))
+
+
+        liq_queue_ids = set()
+        for i in self.liq_queue_tracker:
+            mi = self.liq_queue_tracker[i]
+            if not self.is_media_item_finished(mi):
+                liq_queue_ids.add(mi["row_id"])
+
+        to_be_removed = liq_queue_ids - schedule_ids
+        to_be_added = schedule_ids - liq_queue_ids
+
+        if len(to_be_removed):
+            self.logger.info("Need to remove items from Liquidsoap: %s" % \
+                    to_be_removed)
+
+            #remove files from Liquidsoap's queue
+            for i in self.liq_queue_tracker:
+                mi = self.liq_queue_tracker[i]
+                if mi is not None and mi["row_id"] in to_be_removed:
+                    self.stop(i)
+
+            #stop any webstreams
+            if self.telnet_liquidsoap.get_current_stream_id() in to_be_removed:
+                self.telnet_liquidsoap.stop_web_stream_buffer()
+                self.telnet_liquidsoap.stop_web_stream_output()
+
+        if len(to_be_added):
+            self.logger.info("Need to add items to Liquidsoap *now*: %s" % \
+                    to_be_added)
+
+            for i in scheduled_now:
+                if i["row_id"] in to_be_added:
+                    self.modify_cue_point(i)
+                    self.play(i)
+
+        current_stream_id = self.telnet_liquidsoap.get_current_stream_id()
+        #if len(scheduled_now_webstream):
+            #if current_stream_id != scheduled_now_webstream[0]:
+                #self.play(scheduled_now_webstream[0])
+
+    def play(self, media_item):
+           self.telnet_liquidsoap.push_item(media_item)
+
+    def stop(self, queue):
+        self.telnet_liquidsoap.queue_remove(queue)
+        self.liq_queue_tracker[queue] = None
+
+    def is_file(self, media_item):
+        return media_item["type"] == eventtypes.FILE
+
+    def modify_cue_point(self, link):
+        if not self.is_file(link):
+            return
+
+        tnow = datetime.utcnow()
+
+        link_start = link['start']
+
+        diff_td = tnow - link_start
+        diff_sec = self.date_interval_to_seconds(diff_td)
+
+        if diff_sec > 0:
+            self.logger.debug("media item was supposed to start %s ago. Preparing to start..", diff_sec)
+            original_cue_in_td = timedelta(seconds=float(link['cue_in']))
+            link['cue_in'] = self.date_interval_to_seconds(original_cue_in_td) + diff_sec
+
+    def date_interval_to_seconds(self, interval):
+        """
+        Convert timedelta object into int representing the number of seconds. If
+        number of seconds is less than 0, then return 0.
+        """
+        seconds = (interval.microseconds + \
+                   (interval.seconds + interval.days * 24 * 3600) * 10 ** 6) / float(10 ** 6)
+        if seconds < 0: seconds = 0
+
+        return seconds
+
+
+class UnknownMediaItemType(Exception):
+    pass
+
+class NoQueueAvailableException(Exception):
+    pass