from datetime import datetime from datetime import timedelta import sys import time import logging import logging.config import telnetlib import calendar import json import math from threading import Thread from api_clients import api_client from configobj import ConfigObj # configure logging logging.config.fileConfig("logging.cfg") # loading config file try: config = ConfigObj('/etc/airtime/pypo.cfg') LS_HOST = config['ls_host'] LS_PORT = config['ls_port'] PUSH_INTERVAL = 2 MAX_LIQUIDSOAP_QUEUE_LENGTH = 2 except Exception, e: logger = logging.getLogger() logger.error('Error loading config file %s', e) sys.exit() class PypoPush(Thread): def __init__(self, q, telnet_lock): Thread.__init__(self) self.api_client = api_client.api_client_factory(config) self.queue = q self.media = dict() self.telnet_lock = telnet_lock self.push_ahead = 10 self.last_end_time = 0 self.pushed_objects = {} self.logger = logging.getLogger('push') def push(self): """ The Push Loop - the push loop periodically checks if there is a playlist that should be scheduled at the current time. If yes, the current liquidsoap playlist gets replaced with the corresponding one, then liquidsoap is asked (via telnet) to reload and immediately play it. """ liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap() # get a new schedule from pypo-fetch if not self.queue.empty(): # make sure we get the latest schedule while not self.queue.empty(): self.media = self.queue.get() self.logger.debug("Received data from pypo-fetch") self.logger.debug('media %s' % json.dumps(self.media)) self.handle_new_media(self.media, liquidsoap_queue_approx) media = self.media if len(liquidsoap_queue_approx) < MAX_LIQUIDSOAP_QUEUE_LENGTH: if media: tnow = datetime.utcnow() tcoming = tnow + timedelta(seconds=self.push_ahead) for key in media.keys(): media_item = media[key] item_start = datetime.strptime(media_item['start'][0:19], "%Y-%m-%d-%H-%M-%S") item_end = datetime.strptime(media_item['end'][0:19], "%Y-%m-%d-%H-%M-%S") if len(liquidsoap_queue_approx) == 0 and item_start <= tnow and tnow < item_end: """ Something is scheduled now, but Liquidsoap is not playing anything! Let's play the current media_item """ self.logger.debug("Found media_item that should be playing! Starting...") adjusted_cue_in = tnow - item_start adjusted_cue_in_seconds = self.date_interval_to_seconds(adjusted_cue_in) self.logger.debug("Found media_item that should be playing! Adjust cue point by %ss" % adjusted_cue_in_seconds) self.push_to_liquidsoap(media_item, adjusted_cue_in_seconds) elif tnow <= item_start and item_start < tcoming: """ If the media item starts in the next 10 seconds, push it to the queue. """ self.logger.debug('Preparing to push media item scheduled at: %s', key) if self.push_to_liquidsoap(media_item, None): self.logger.debug("Pushed to liquidsoap, updating 'played' status.") """ Temporary solution to make sure we don't push the same track multiple times. Not a full solution because if we get a new schedule, the key becomes available again. """ #TODO del media[key] def date_interval_to_seconds(self, interval): return (interval.microseconds + (interval.seconds + interval.days * 24 * 3600) * 10**6) / 10**6 def push_to_liquidsoap(self, media_item, adjusted_cue_in=None): """ This function looks at the media item, and either pushes it to the Liquidsoap queue immediately, or if the queue is empty - waits until the start time of the media item before pushing it. """ if adjusted_cue_in is not None: media_item["cue_in"] = adjusted_cue_in + float(media_item["cue_in"]) try: if media_item["start"] == self.last_end_time: """ this media item is attached to the end of the last track, so let's push it now so that Liquidsoap can start playing it immediately after (and prepare crossfades if need be). """ self.logger.debug("Push track immediately.") self.telnet_to_liquidsoap(media_item) self.last_end_time = media_item["end"] else: """ this media item does not start right after a current playing track. We need to sleep, and then wake up when this track starts. """ self.logger.debug("sleep until track start.") self.sleep_until_start(media_item) self.telnet_to_liquidsoap(media_item) self.last_end_time = media_item["end"] except Exception, e: self.logger.error('Pypo Push Exception: %s', e) return False return True 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) msg = 'queue.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(str(e)) 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 handle_new_media(self, media, liquidsoap_queue_approx): """ 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. """ if len(liquidsoap_queue_approx) == 0: """ liquidsoap doesn't have anything in its queue, so we have nothing to worry about. Life is good. """ pass elif len(liquidsoap_queue_approx) == 1: queue_item_0_start = liquidsoap_queue_approx[0]['start'] try: if liquidsoap_queue_approx[0]['id'] != media[queue_item_0_start]['id']: """ liquidsoap's queue does not match the schedule we just received from the Airtime server. The queue is only of length 1 which means the item in the queue is playing. Need to do source.skip. Since only one item, we don't have to worry about the current item ending and us calling source.skip unintentionally on the next item (there is no next item). """ self.logger.debug("%s from ls does not exist in queue new schedule. Removing" % liquidsoap_queue_approx[0]['id'], media) self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[0]) except KeyError, k: self.logger.debug("%s from ls does not exist in queue schedule: %s Removing" % (queue_item_0_start, media)) self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[0]) elif len(liquidsoap_queue_approx) == 2: queue_item_0_start = liquidsoap_queue_approx[0]['start'] queue_item_1_start = liquidsoap_queue_approx[1]['start'] if queue_item_1_start in media.keys(): if liquidsoap_queue_approx[1]['id'] != media[queue_item_1_start]['id']: self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[1]) else: self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[1]) if queue_item_0_start in media.keys(): if liquidsoap_queue_approx[0]['id'] != media[queue_item_0_start]['id']: self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[0]) else: self.remove_from_liquidsoap_queue(liquidsoap_queue_approx[0]) 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, media_item, do_only_source_skip=False): if 'queue_id' in media_item: queue_id = media_item['queue_id'] try: self.telnet_lock.acquire() tn = telnetlib.Telnet(LS_HOST, LS_PORT) msg = "queue.remove %s\n" % 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) tn.write("exit\n") tn.read_all() except Exception, e: self.logger.error(str(e)) finally: self.telnet_lock.release() else: self.logger.error("'queue_id' key doesn't exist in media_item dict()") 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) #tn.write(("vars.pypo_data %s\n"%liquidsoap_data["schedule_id"]).encode('utf-8')) annotation = self.create_liquidsoap_annotation(media_item) msg = 'queue.push %s\n' % 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): return 'annotate:media_id="%s",liq_cue_in="%s",liq_cue_out="%s",schedule_table_id="%s":%s' \ % (media['id'], float(media['cue_in']), float(media['cue_out']), media['row_id'], media['dst']) def run(self): loops = 0 heartbeat_period = math.floor(30/PUSH_INTERVAL) while True: if loops % heartbeat_period == 0: self.logger.info("heartbeat") loops = 0 try: self.push() except Exception, e: self.logger.error('Pypo Push Exception: %s', e) time.sleep(PUSH_INTERVAL) loops += 1