From 445573dcdbedfdb1bd4048c439eec962d5af0eaf Mon Sep 17 00:00:00 2001 From: Martin Konecny Date: Fri, 15 Mar 2013 15:07:55 -0400 Subject: [PATCH] CC-1469: Crossfading support (non-equal power) -webstreams scheduled in the future are now working... --- python_apps/pypo/eventtypes.py | 6 + python_apps/pypo/pypofetch.py | 2 +- python_apps/pypo/pypoliqqueue.py | 65 ++- python_apps/pypo/pypopush.py | 612 +-------------------------- python_apps/pypo/telnetliquidsoap.py | 90 ++++ 5 files changed, 177 insertions(+), 598 deletions(-) create mode 100644 python_apps/pypo/eventtypes.py diff --git a/python_apps/pypo/eventtypes.py b/python_apps/pypo/eventtypes.py new file mode 100644 index 000000000..5f9c871db --- /dev/null +++ b/python_apps/pypo/eventtypes.py @@ -0,0 +1,6 @@ +FILE = "file" +EVENT = "event" +STREAM_BUFFER_START = "stream_buffer_start" +STREAM_OUTPUT_START = "stream_output_start" +STREAM_BUFFER_END = "stream_buffer_end" +STREAM_OUTPUT_END = "stream_output_end" diff --git a/python_apps/pypo/pypofetch.py b/python_apps/pypo/pypofetch.py index 3804b3458..66243a15a 100644 --- a/python_apps/pypo/pypofetch.py +++ b/python_apps/pypo/pypofetch.py @@ -540,7 +540,7 @@ class PypoFetch(Thread): #check if this file is opened (sometimes Liquidsoap is still #playing the file due to our knowledge of the track length #being incorrect!) - if not self.is_file_opened(): + if not self.is_file_opened(path): os.remove(path) except Exception, e: self.logger.error(e) diff --git a/python_apps/pypo/pypoliqqueue.py b/python_apps/pypo/pypoliqqueue.py index 87458d4df..54d7c5d93 100644 --- a/python_apps/pypo/pypoliqqueue.py +++ b/python_apps/pypo/pypoliqqueue.py @@ -1,9 +1,13 @@ from threading import Thread from collections import deque from datetime import datetime +from pypofetch import PypoFetch + +import eventtypes import traceback import sys +import time from Queue import Empty @@ -52,6 +56,8 @@ class PypoLiqQueue(Thread): else: time_until_next_play = None else: + self.logger.info("New schedule received: %s", media_schedule) + #new schedule received. Replace old one with this. schedule_deque.clear() @@ -62,6 +68,8 @@ class PypoLiqQueue(Thread): if len(keys): time_until_next_play = self.date_interval_to_seconds(\ keys[0] - datetime.utcnow()) + else: + time_until_next_play = None def is_media_item_finished(self, media_item): if media_item is None: @@ -88,15 +96,53 @@ class PypoLiqQueue(Thread): show name of every media_item as well, just to keep Liquidsoap up-to-date about which show is playing. """ - - 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 + 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(media_item) + elif media_item['type'] == eventtypes.STREAM_OUTPUT_END: + self.telnet_liquidsoap.stop_web_stream_output(media_item) + else: raise UnknownMediaItemType(str(media_item)) + + 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 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 date_interval_to_seconds(self, interval): """ @@ -117,3 +163,6 @@ class PypoLiqQueue(Thread): class NoQueueAvailableException(Exception): pass + +class UnknownMediaItemType(Exception): + pass diff --git a/python_apps/pypo/pypopush.py b/python_apps/pypo/pypopush.py index 0b4371042..6ae3e27d1 100644 --- a/python_apps/pypo/pypopush.py +++ b/python_apps/pypo/pypopush.py @@ -16,7 +16,6 @@ from pypofetch import PypoFetch from telnetliquidsoap import TelnetLiquidsoap from pypoliqqueue import PypoLiqQueue - from Queue import Empty, Queue from threading import Thread @@ -41,7 +40,6 @@ try: LS_HOST = config['ls_host'] LS_PORT = config['ls_port'] PUSH_INTERVAL = 2 - MAX_LIQUIDSOAP_QUEUE_LENGTH = 2 except Exception, e: logger.error('Error loading config file %s', e) sys.exit() @@ -90,37 +88,14 @@ class PypoPush(Thread): loops = 0 heartbeat_period = math.floor(30 / PUSH_INTERVAL) - next_media_item_chain = None media_schedule = None - time_until_next_play = None - chains = None while True: try: - if time_until_next_play is None: - media_schedule = self.queue.get(block=True) - else: - media_schedule = self.queue.get(block=True, timeout=time_until_next_play) - except Empty, e: - #We only get here when a new chain of tracks are ready to be played. - #"timeout" has parameter has been reached. - self.push_to_liquidsoap(next_media_item_chain) - - next_media_item_chain = self.get_next_schedule_chain(chains, datetime.utcnow()) - if next_media_item_chain is not None: - try: - chains.remove(next_media_item_chain) - except ValueError, e: - self.logger.error(str(e)) - - chain_start = next_media_item_chain[0]['start'] - time_until_next_play = self.date_interval_to_seconds(chain_start - datetime.utcnow()) - self.logger.debug("Blocking %s seconds until show start", time_until_next_play) - else: - self.logger.debug("Blocking indefinitely since no show scheduled next") - time_until_next_play = None + media_schedule = self.queue.get(block=True) except Exception, e: self.logger.error(str(e)) + raise else: #separate media_schedule list into currently_playing and #scheduled_for_future lists @@ -128,60 +103,8 @@ class PypoPush(Thread): self.separate_present_future(media_schedule) self.verify_correct_present_media(currently_playing) - self.future_scheduled_queue.put(scheduled_for_future) - """ - #queue.get timeout never had a chance to expire. Instead a new - #schedule was received. Let's parse this schedule and generate - #a new timeout. - try: - chains = self.get_all_chains(media_schedule) - - #We get to the following lines only if a schedule was received. - liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap() - liquidsoap_stream_id = self.get_current_stream_id_from_liquidsoap() - - tnow = datetime.utcnow() - current_event_chain, original_chain = \ - self.get_current_chain(chains, tnow) - - if len(current_event_chain) > 0: - try: - chains.remove(original_chain) - except ValueError, e: - self.logger.error(str(e)) - - #At this point we know that Liquidsoap is playing something, and that something - #is scheduled. We need to verify whether the schedule we just received matches - #what Liquidsoap is playing, and if not, correct it. - self.handle_new_schedule(media_schedule, \ - liquidsoap_queue_approx, \ - liquidsoap_stream_id, \ - current_event_chain) - - #At this point everything in the present has been taken care of and Liquidsoap - #is playing whatever is scheduled. - #Now we need to prepare ourselves for future scheduled events. - next_media_item_chain = self.get_next_schedule_chain(chains, tnow) - - self.logger.debug("Next schedule chain: %s", next_media_item_chain) - if next_media_item_chain is not None: - try: - chains.remove(next_media_item_chain) - except ValueError, e: - self.logger.error(str(e)) - - chain_start = datetime.strptime(next_media_item_chain[0]['start'], "%Y-%m-%d-%H-%M-%S") - time_until_next_play = self.date_interval_to_seconds(chain_start - datetime.utcnow()) - self.logger.debug("Blocking %s seconds until show start", time_until_next_play) - else: - self.logger.debug("Blocking indefinitely since no show scheduled") - time_until_next_play = None - except Exception, e: - self.logger.error(str(e)) - """ - if loops % heartbeat_period == 0: self.logger.info("heartbeat") loops = 0 @@ -209,7 +132,7 @@ class PypoPush(Thread): return present, future def verify_correct_present_media(self, scheduled_now): - #verify whether Liquidsoap is currently playing the correct items. + #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 @@ -219,6 +142,9 @@ class PypoPush(Thread): #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 + schedule_ids = set() for i in scheduled_now: schedule_ids.add(i["row_id"]) @@ -273,198 +199,24 @@ class PypoPush(Thread): return response - def get_queue_items_from_liquidsoap(self): - """ - This function connects to Liquidsoap to find what media items are in its queue. - """ - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) + #def is_correct_current_item(self, media_item, liquidsoap_queue_approx, liquidsoap_stream_id): + #correct = False + #if media_item is None: + #correct = (len(liquidsoap_queue_approx) == 0 and liquidsoap_stream_id == "-1") + #else: + #if is_file(media_item): + #if len(liquidsoap_queue_approx) == 0: + #correct = False + #else: + #correct = liquidsoap_queue_approx[0]['start'] == media_item['start'] and \ + #liquidsoap_queue_approx[0]['row_id'] == media_item['row_id'] and \ + #liquidsoap_queue_approx[0]['end'] == media_item['end'] and \ + #liquidsoap_queue_approx[0]['replay_gain'] == media_item['replay_gain'] + #elif is_stream(media_item): + #correct = liquidsoap_stream_id == str(media_item['row_id']) - msg = 's0.queue\n' - tn.write(msg) - response = tn.read_until("\r\n").strip(" \r\n") - tn.write('exit\n') - tn.read_all() - except Exception, e: - self.logger.error("Error connecting to Liquidsoap: %s", e) - response = [] - finally: - self.telnet_lock.release() - - liquidsoap_queue_approx = [] - - if len(response) > 0: - items_in_queue = response.split(" ") - - self.logger.debug("items_in_queue: %s", items_in_queue) - - for item in items_in_queue: - if item in self.pushed_objects: - liquidsoap_queue_approx.append(self.pushed_objects[item]) - else: - """ - We should only reach here if Pypo crashed and restarted (because self.pushed_objects was reset). In this case - let's clear the entire Liquidsoap queue. - """ - self.logger.error("ID exists in liquidsoap queue that does not exist in our pushed_objects queue: " + item) - self.clear_liquidsoap_queue() - liquidsoap_queue_approx = [] - break - - return liquidsoap_queue_approx - - def is_correct_current_item(self, media_item, liquidsoap_queue_approx, liquidsoap_stream_id): - correct = False - if media_item is None: - correct = (len(liquidsoap_queue_approx) == 0 and liquidsoap_stream_id == "-1") - else: - if is_file(media_item): - if len(liquidsoap_queue_approx) == 0: - correct = False - else: - correct = liquidsoap_queue_approx[0]['start'] == media_item['start'] and \ - liquidsoap_queue_approx[0]['row_id'] == media_item['row_id'] and \ - liquidsoap_queue_approx[0]['end'] == media_item['end'] and \ - liquidsoap_queue_approx[0]['replay_gain'] == media_item['replay_gain'] - elif is_stream(media_item): - correct = liquidsoap_stream_id == str(media_item['row_id']) - - self.logger.debug("Is current item correct?: %s", str(correct)) - return correct - - - #clear all webstreams and files from Liquidsoap - def clear_all_liquidsoap_items(self): - self.remove_from_liquidsoap_queue(0, None) - self.stop_web_stream_all() - - def handle_new_schedule(self, media_schedule, liquidsoap_queue_approx, liquidsoap_stream_id, current_event_chain): - """ - This function's purpose is to gracefully handle situations where - Liquidsoap already has a track in its queue, but the schedule - has changed. If the schedule has changed, this function's job is to - call other functions that will connect to Liquidsoap and alter its - queue. - """ - file_chain = filter(lambda item: (item["type"] == "file"), current_event_chain) - stream_chain = filter(lambda item: (item["type"] == "stream_output_start"), current_event_chain) - - self.logger.debug(current_event_chain) - - #Take care of the case where the current playing may be incorrect - if len(current_event_chain) > 0: - - current_item = current_event_chain[0] - if not self.is_correct_current_item(current_item, liquidsoap_queue_approx, liquidsoap_stream_id): - self.clear_all_liquidsoap_items() - if is_stream(current_item): - if current_item['row_id'] != self.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.start_web_stream_buffer(current_item) - self.start_web_stream(current_item) - if is_file(current_item): - file_chain = self.modify_first_link_cue_point(file_chain) - self.push_to_liquidsoap(file_chain) - #we've changed the queue, so let's refetch it - liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap() - - elif not self.is_correct_current_item(None, liquidsoap_queue_approx, liquidsoap_stream_id): - #Liquidsoap is playing something even though it shouldn't be - self.clear_all_liquidsoap_items() - - - #If the current item scheduled is a file, then files come in chains, and - #therefore we need to make sure the entire chain is correct. - if len(current_event_chain) > 0 and is_file(current_event_chain[0]): - problem_at_iteration = self.find_removed_items(media_schedule, liquidsoap_queue_approx) - - if problem_at_iteration is not None: - #Items that are in Liquidsoap's queue aren't scheduled anymore. We need to connect - #and remove these items. - self.logger.debug("Change in link %s of current chain", problem_at_iteration) - self.remove_from_liquidsoap_queue(problem_at_iteration, liquidsoap_queue_approx[problem_at_iteration:]) - - if problem_at_iteration is None and len(file_chain) > len(liquidsoap_queue_approx): - self.logger.debug("New schedule has longer current chain.") - problem_at_iteration = len(liquidsoap_queue_approx) - - if problem_at_iteration is not None: - self.logger.debug("Change in chain at link %s", problem_at_iteration) - - chain_to_push = file_chain[problem_at_iteration:] - if len(chain_to_push) > 0: - chain_to_push = self.modify_first_link_cue_point(chain_to_push) - self.push_to_liquidsoap(chain_to_push) - - - """ - Compare whats in the liquidsoap_queue to the new schedule we just - received in media_schedule. This function only iterates over liquidsoap_queue_approx - and finds if every item in that list is still scheduled in "media_schedule". It doesn't - take care of the case where media_schedule has more items than liquidsoap_queue_approx - """ - def find_removed_items(self, media_schedule, liquidsoap_queue_approx): - #iterate through the items we got from the liquidsoap queue and - #see if they are the same as the newly received schedule - iteration = 0 - problem_at_iteration = None - for queue_item in liquidsoap_queue_approx: - if queue_item['start'] in media_schedule.keys(): - media_item = media_schedule[queue_item['start']] - if queue_item['row_id'] == media_item['row_id']: - if queue_item['end'] == media_item['end']: - #Everything OK for this iteration. - pass - else: - problem_at_iteration = iteration - break - else: - #A different item has been scheduled at the same time! Need to remove - #all tracks from the Liquidsoap queue starting at this point, and re-add - #them. - problem_at_iteration = iteration - break - else: - #There are no more items scheduled for this time! The user has shortened - #the playlist, so we simply need to remove tracks from the queue. - problem_at_iteration = iteration - break - iteration += 1 - return problem_at_iteration - - - - def get_all_chains(self, media_schedule): - chains = [] - - current_chain = [] - - sorted_keys = sorted(media_schedule.keys()) - - for mkey in sorted_keys: - media_item = media_schedule[mkey] - if media_item['independent_event']: - if len(current_chain) > 0: - chains.append(current_chain) - - chains.append([media_item]) - current_chain = [] - elif len(current_chain) == 0: - current_chain.append(media_item) - elif media_item['start'] == current_chain[-1]['end']: - current_chain.append(media_item) - else: - #current item is not a continuation of the chain. - #Start a new one instead - chains.append(current_chain) - current_chain = [media_item] - - if len(current_chain) > 0: - chains.append(current_chain) - - return chains + #self.logger.debug("Is current item correct?: %s", str(correct)) + #return correct def modify_cue_point(self, link): tnow = datetime.utcnow() @@ -479,75 +231,6 @@ class PypoPush(Thread): 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 modify_first_link_cue_point(self, chain): - if not len(chain): - return [] - - first_link = chain[0] - - self.modify_cue_point(first_link) - if float(first_link['cue_in']) >= float(first_link['cue_out']): - chain = chain [1:] - - return chain - - """ - Returns two chains, original chain and current_chain. current_chain is a subset of - original_chain but can also be equal to original chain. - - We return original chain because the user of this function may want to clean - up the input 'chains' list - - chain, original = get_current_chain(chains) - - and - chains.remove(chain) can throw a ValueError exception - - but - chains.remove(original) won't - """ - def get_current_chain(self, chains, tnow): - current_chain = [] - original_chain = None - - for chain in chains: - iteration = 0 - for link in chain: - link_start = link['start'] - link_end = link['end'] - - self.logger.debug("tnow %s, chain_start %s", tnow, link_start) - if link_start <= tnow and tnow < link_end: - current_chain = chain[iteration:] - original_chain = chain - break - iteration += 1 - - return current_chain, original_chain - - """ - The purpose of this function is to take a look at the last received schedule from - pypo-fetch and return the next chain of media_items. A chain is defined as a sequence - of media_items where the end time of media_item 'n' is the start time of media_item - 'n+1' - """ - def get_next_schedule_chain(self, chains, tnow): - #all media_items are now divided into chains. Let's find the one that - #starts closest in the future. - closest_start = None - closest_chain = None - for chain in chains: - chain_start = chain[0]['start'] - chain_end = chain[-1]['end'] - self.logger.debug("tnow %s, chain_start %s", tnow, chain_start) - if (closest_start == None or chain_start < closest_start) and \ - (chain_start > tnow or \ - (chain_start < tnow and chain_end > tnow)): - closest_start = chain_start - closest_chain = chain - - return closest_chain - def date_interval_to_seconds(self, interval): """ @@ -559,94 +242,6 @@ class PypoPush(Thread): return seconds - def push_to_liquidsoap(self, event_chain): - - try: - for media_item in event_chain: - if media_item['type'] == "file": - - """ - 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']: - self.telnet_to_liquidsoap(media_item) - else: - self.logger.warn("File %s did not become ready in less than 5 seconds. Skipping...", media_item['dst']) - elif media_item['type'] == "event": - 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") - elif media_item['type'] == 'stream_buffer_start': - self.start_web_stream_buffer(media_item) - elif media_item['type'] == "stream_output_start": - if media_item['row_id'] != self.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.start_web_stream_buffer(media_item) - self.start_web_stream(media_item) - elif media_item['type'] == "stream_buffer_end": - self.stop_web_stream_buffer(media_item) - elif media_item['type'] == "stream_output_end": - self.stop_web_stream_output(media_item) - except Exception, e: - self.logger.error('Pypo Push Exception: %s', e) - finally: - self.queue_id = (self.queue_id + 1) % 8 - - - def start_web_stream_buffer(self, media_item): - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - - msg = 'dynamic_source.id %s\n' % media_item['row_id'] - self.logger.debug(msg) - tn.write(msg) - - #msg = 'dynamic_source.read_start %s\n' % media_item['uri'].encode('latin-1') - msg = 'http.restart %s\n' % media_item['uri'].encode('latin-1') - self.logger.debug(msg) - tn.write(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - - self.current_prebuffering_stream_id = media_item['row_id'] - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - - def start_web_stream(self, media_item): - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - - #TODO: DO we need this? - msg = 'streams.scheduled_play_start\n' - tn.write(msg) - - msg = 'dynamic_source.output_start\n' - self.logger.debug(msg) - tn.write(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - - self.current_prebuffering_stream_id = None - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - def stop_web_stream_all(self): try: self.telnet_lock.acquire() @@ -673,167 +268,6 @@ class PypoPush(Thread): finally: self.telnet_lock.release() - def stop_web_stream_buffer(self, media_item): - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - #dynamic_source.stop http://87.230.101.24:80/top100station.mp3 - - #msg = 'dynamic_source.read_stop %s\n' % media_item['row_id'] - msg = 'http.stop\n' - self.logger.debug(msg) - tn.write(msg) - - msg = 'dynamic_source.id -1\n' - self.logger.debug(msg) - tn.write(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - def stop_web_stream_output(self, media_item): - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - #dynamic_source.stop http://87.230.101.24:80/top100station.mp3 - - msg = 'dynamic_source.output_stop\n' - self.logger.debug(msg) - tn.write(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - def clear_liquidsoap_queue(self): - self.logger.debug("Clearing Liquidsoap queue") - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - msg = "source.skip\n" - tn.write(msg) - tn.write("exit\n") - tn.read_all() - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - def remove_from_liquidsoap_queue(self, problem_at_iteration, liquidsoap_queue_approx): - - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - - if problem_at_iteration == 0: - msg = "source.skip\n" - self.logger.debug(msg) - tn.write(msg) - else: - # Remove things in reverse order. - queue_copy = liquidsoap_queue_approx[::-1] - - for queue_item in queue_copy: - msg = "queue.remove %s\n" % queue_item['queue_id'] - self.logger.debug(msg) - tn.write(msg) - response = tn.read_until("\r\n").strip("\r\n") - - if "No such request in my queue" in response: - """ - Cannot remove because Liquidsoap started playing the item. Need - to use source.skip instead - """ - msg = "source.skip\n" - self.logger.debug(msg) - tn.write(msg) - - msg = "s0.queue\n" - self.logger.debug(msg) - tn.write(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - def sleep_until_start(self, media_item): - """ - The purpose of this function is to look at the difference between - "now" and when the media_item starts, and sleep for that period of time. - After waking from sleep, this function returns. - """ - - mi_start = media_item['start'][0:19] - - #strptime returns struct_time in local time - epoch_start = calendar.timegm(time.strptime(mi_start, '%Y-%m-%d-%H-%M-%S')) - - #Return the time as a floating point number expressed in seconds since the epoch, in UTC. - epoch_now = time.time() - - self.logger.debug("Epoch start: %s" % epoch_start) - self.logger.debug("Epoch now: %s" % epoch_now) - - sleep_time = epoch_start - epoch_now - - if sleep_time < 0: - sleep_time = 0 - - self.logger.debug('sleeping for %s s' % (sleep_time)) - time.sleep(sleep_time) - - def telnet_to_liquidsoap(self, media_item): - """ - telnets to liquidsoap and pushes the media_item to its queue. Push the - show name of every media_item as well, just to keep Liquidsoap up-to-date - about which show is playing. - """ - try: - self.telnet_lock.acquire() - tn = telnetlib.Telnet(LS_HOST, LS_PORT) - - annotation = self.create_liquidsoap_annotation(media_item) - msg = 's%s.push %s\n' % (self.queue_id, annotation.encode('utf-8')) - self.logger.debug(msg) - tn.write(msg) - queue_id = tn.read_until("\r\n").strip("\r\n") - - #remember the media_item's queue id which we may use - #later if we need to remove it from the queue. - media_item['queue_id'] = queue_id - - #add media_item to the end of our queue - self.pushed_objects[queue_id] = media_item - - show_name = media_item['show_name'] - msg = 'vars.show_name %s\n' % show_name.encode('utf-8') - tn.write(msg) - self.logger.debug(msg) - - tn.write("exit\n") - self.logger.debug(tn.read_all()) - except Exception, e: - self.logger.error(str(e)) - finally: - self.telnet_lock.release() - - def create_liquidsoap_annotation(self, media): - # We need liq_start_next value in the annotate. That is the value that controls overlap duration of crossfade. - return 'annotate:media_id="%s",liq_start_next="0",liq_fade_in="%s",liq_fade_out="%s",liq_cue_in="%s",liq_cue_out="%s",schedule_table_id="%s",replay_gain="%s dB":%s' \ - % (media['id'], float(media['fade_in']) / 1000, float(media['fade_out']) / 1000, float(media['cue_in']), float(media['cue_out']), media['row_id'], media['replay_gain'], media['dst']) - def run(self): try: self.main() except Exception, e: diff --git a/python_apps/pypo/telnetliquidsoap.py b/python_apps/pypo/telnetliquidsoap.py index 7bc7e362d..019a62b29 100644 --- a/python_apps/pypo/telnetliquidsoap.py +++ b/python_apps/pypo/telnetliquidsoap.py @@ -12,6 +12,7 @@ class TelnetLiquidsoap: self.ls_host = ls_host self.ls_port = ls_port self.logger = logger + self.current_prebuffering_stream_id = None def __connect(self): return telnetlib.Telnet(self.ls_host, self.ls_port) @@ -61,6 +62,95 @@ class TelnetLiquidsoap: finally: self.telnet_lock.release() + + def stop_web_stream_buffer(self, media_item): + try: + self.telnet_lock.acquire() + tn = telnetlib.Telnet(self.ls_host, self.ls_port) + #dynamic_source.stop http://87.230.101.24:80/top100station.mp3 + + #msg = 'dynamic_source.read_stop %s\n' % media_item['row_id'] + msg = 'http.stop\n' + self.logger.debug(msg) + tn.write(msg) + + msg = 'dynamic_source.id -1\n' + self.logger.debug(msg) + tn.write(msg) + + tn.write("exit\n") + self.logger.debug(tn.read_all()) + + except Exception, e: + self.logger.error(str(e)) + finally: + self.telnet_lock.release() + + def stop_web_stream_output(self, media_item): + try: + self.telnet_lock.acquire() + tn = telnetlib.Telnet(self.ls_host, self.ls_port) + #dynamic_source.stop http://87.230.101.24:80/top100station.mp3 + + msg = 'dynamic_source.output_stop\n' + self.logger.debug(msg) + tn.write(msg) + + tn.write("exit\n") + self.logger.debug(tn.read_all()) + + except Exception, e: + self.logger.error(str(e)) + finally: + self.telnet_lock.release() + + def start_web_stream(self, media_item): + try: + self.telnet_lock.acquire() + tn = telnetlib.Telnet(self.ls_host, self.ls_port) + + #TODO: DO we need this? + msg = 'streams.scheduled_play_start\n' + tn.write(msg) + + msg = 'dynamic_source.output_start\n' + self.logger.debug(msg) + tn.write(msg) + + tn.write("exit\n") + self.logger.debug(tn.read_all()) + + self.current_prebuffering_stream_id = None + except Exception, e: + self.logger.error(str(e)) + finally: + self.telnet_lock.release() + + def start_web_stream_buffer(self, media_item): + try: + self.telnet_lock.acquire() + tn = telnetlib.Telnet(self.ls_host, self.ls_port) + + msg = 'dynamic_source.id %s\n' % media_item['row_id'] + self.logger.debug(msg) + tn.write(msg) + + #msg = 'dynamic_source.read_start %s\n' % media_item['uri'].encode('latin-1') + msg = 'http.restart %s\n' % media_item['uri'].encode('latin-1') + self.logger.debug(msg) + tn.write(msg) + + tn.write("exit\n") + self.logger.debug(tn.read_all()) + + #TODO.. + self.current_prebuffering_stream_id = media_item['row_id'] + except Exception, e: + self.logger.error(str(e)) + finally: + self.telnet_lock.release() + + class DummyTelnetLiquidsoap: def __init__(self, telnet_lock, logger):