Major reorganization of the pypo directory.

Much better organization of files into packages

python_apps/pypo/schedule/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"

python_apps/pypo/schedule/listenerstat.py

@@ -0,0 +1,160 @@
+from threading import Thread
+import urllib2
+import xml.dom.minidom
+import base64
+from datetime import datetime
+import traceback
+import logging
+import time
+
+from api_clients import api_client
+
+class ListenerStat(Thread):
+    def __init__(self, logger=None):
+        Thread.__init__(self)
+        self.api_client = api_client.AirtimeApiClient()
+        if logger is None:
+            self.logger = logging.getLogger()
+        else:
+            self.logger = logger
+
+    def get_node_text(self, nodelist):
+        rc = []
+        for node in nodelist:
+            if node.nodeType == node.TEXT_NODE:
+                rc.append(node.data)
+        return ''.join(rc)
+
+    def get_stream_parameters(self):
+        #[{"user":"", "password":"", "url":"", "port":""},{},{}]
+        return self.api_client.get_stream_parameters()
+
+
+    def get_stream_server_xml(self, ip, url, is_shoutcast=False):
+        encoded = base64.b64encode("%(admin_user)s:%(admin_pass)s" % ip)
+
+        header = {"Authorization":"Basic %s" % encoded}
+
+        if is_shoutcast:
+            #user agent is required for shoutcast auth, otherwise it returns 404.
+            user_agent = "Mozilla/5.0 (Linux; rv:22.0) Gecko/20130405 Firefox/22.0"
+            header["User-Agent"] = user_agent
+
+        req = urllib2.Request(
+            #assuming that the icecast stats path is /admin/stats.xml
+            #need to fix this
+            url=url,
+            headers=header)
+
+        f = urllib2.urlopen(req)
+        document = f.read()
+        return document
+
+
+    def get_icecast_stats(self, ip):
+        url = 'http://%(host)s:%(port)s/admin/stats.xml' % ip
+        document = self.get_stream_server_xml(ip, url)
+        dom = xml.dom.minidom.parseString(document)
+        sources = dom.getElementsByTagName("source")
+
+        mount_stats = None
+        for s in sources:
+            #drop the leading '/' character
+            mount_name = s.getAttribute("mount")[1:]
+            if mount_name == ip["mount"]:
+                timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+                listeners = s.getElementsByTagName("listeners")
+                num_listeners = 0
+                if len(listeners):
+                    num_listeners = self.get_node_text(listeners[0].childNodes)
+
+                mount_stats = {"timestamp":timestamp, \
+                            "num_listeners": num_listeners, \
+                            "mount_name": mount_name}
+
+        return mount_stats
+
+    def get_shoutcast_stats(self, ip):
+        url = 'http://%(host)s:%(port)s/admin.cgi?sid=1&mode=viewxml' % ip
+        document = self.get_stream_server_xml(ip, url, is_shoutcast=True)
+        dom = xml.dom.minidom.parseString(document)
+        current_listeners = dom.getElementsByTagName("CURRENTLISTENERS")
+
+        timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")
+        num_listeners = 0
+        if len(current_listeners):
+            num_listeners = self.get_node_text(current_listeners[0].childNodes)
+
+        mount_stats = {"timestamp":timestamp, \
+                    "num_listeners": num_listeners, \
+                    "mount_name": "shoutcast"}
+
+        return mount_stats
+
+    def get_stream_stats(self, stream_parameters):
+        stats = []
+
+        #iterate over stream_parameters which is a list of dicts. Each dict
+        #represents one Airtime stream (currently this limit is 3).
+        #Note that there can be optimizations done, since if all three
+        #streams are the same server, we will still initiate 3 separate
+        #connections
+        for k, v in stream_parameters.items():
+            if v["enable"] == 'true':
+                try:
+                    if v["output"] == "icecast":
+                        mount_stats = self.get_icecast_stats(v)
+                        if mount_stats: stats.append(mount_stats)
+                    else:
+                        stats.append(self.get_shoutcast_stats(v))
+                    self.update_listener_stat_error(v["mount"], 'OK')
+                except Exception, e:
+                    try:
+                        self.update_listener_stat_error(v["mount"], str(e))
+                    except Exception, e:
+                        self.logger.error('Exception: %s', e)
+
+        return stats
+
+    def push_stream_stats(self, stats):
+        self.api_client.push_stream_stats(stats)
+
+    def update_listener_stat_error(self, stream_id, error):
+        keyname = '%s_listener_stat_error' % stream_id
+        data = {keyname: error}
+        self.api_client.update_stream_setting_table(data)
+
+    def run(self):
+        #Wake up every 120 seconds and gather icecast statistics. Note that we
+        #are currently querying the server every 2 minutes for list of
+        #mountpoints as well. We could remove this query if we hooked into
+        #rabbitmq events, and listened for these changes instead.
+        while True:
+            try:
+                stream_parameters = self.get_stream_parameters()
+                stats = self.get_stream_stats(stream_parameters["stream_params"])
+
+                if stats:
+                    self.push_stream_stats(stats)
+            except Exception, e:
+                self.logger.error('Exception: %s', e)
+
+            time.sleep(120)
+
+
+if __name__ == "__main__":
+    # create logger
+    logger = logging.getLogger('std_out')
+    logger.setLevel(logging.DEBUG)
+    # create console handler and set level to debug
+    #ch = logging.StreamHandler()
+    #ch.setLevel(logging.DEBUG)
+    # create formatter
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(lineno)s - %(levelname)s - %(message)s')
+    # add formatter to ch
+    #ch.setFormatter(formatter)
+    # add ch to logger
+    #logger.addHandler(ch)
+
+    ls = ListenerStat(logger)
+    ls.run()

python_apps/pypo/schedule/pypofetch.py

@@ -0,0 +1,584 @@
+# -*- coding: utf-8 -*-
+
+import os
+import sys
+import time
+import logging.config
+import json
+import telnetlib
+import copy
+import subprocess
+import signal
+from datetime import datetime
+import traceback
+import pure
+
+from Queue import Empty
+from threading import Thread
+from subprocess import Popen, PIPE
+
+from api_clients import api_client
+from std_err_override import LogWriter
+
+
+# configure logging
+logging_cfg = os.path.join(os.path.dirname(__file__), "../logging.cfg")
+logging.config.fileConfig(logging_cfg)
+logger = logging.getLogger()
+LogWriter.override_std_err(logger)
+
+def keyboardInterruptHandler(signum, frame):
+    logger = logging.getLogger()
+    logger.info('\nKeyboard Interrupt\n')
+    sys.exit(0)
+signal.signal(signal.SIGINT, keyboardInterruptHandler)
+
+#need to wait for Python 2.7 for this..
+#logging.captureWarnings(True)
+
+POLL_INTERVAL = 1800
+
+config_static = None
+
+class PypoFetch(Thread):
+    def __init__(self, pypoFetch_q, pypoPush_q, media_q, telnet_lock, pypo_liquidsoap, config):
+        Thread.__init__(self)
+        global config_static
+
+        self.api_client = api_client.AirtimeApiClient()
+        self.fetch_queue = pypoFetch_q
+        self.push_queue = pypoPush_q
+        self.media_prepare_queue = media_q
+        self.last_update_schedule_timestamp = time.time()
+        self.config = config
+        config_static = config
+        self.listener_timeout = POLL_INTERVAL
+
+        self.telnet_lock = telnet_lock
+
+        self.logger = logging.getLogger()
+
+        self.pypo_liquidsoap = pypo_liquidsoap
+
+        self.cache_dir = os.path.join(config["cache_dir"], "scheduler")
+        self.logger.debug("Cache dir %s", self.cache_dir)
+
+        try:
+            if not os.path.isdir(dir):
+                """
+                We get here if path does not exist, or path does exist but
+                is a file. We are not handling the second case, but don't
+                think we actually care about handling it.
+                """
+                self.logger.debug("Cache dir does not exist. Creating...")
+                os.makedirs(dir)
+        except Exception, e:
+            pass
+
+        self.schedule_data = []
+        self.logger.info("PypoFetch: init complete")
+
+    """
+    Handle a message from RabbitMQ, put it into our yucky global var.
+    Hopefully there is a better way to do this.
+    """
+    def handle_message(self, message):
+        try:
+            self.logger.info("Received event from Pypo Message Handler: %s" % message)
+
+            m = json.loads(message)
+            command = m['event_type']
+            self.logger.info("Handling command: " + command)
+
+            if command == 'update_schedule':
+                self.schedule_data = m['schedule']
+                self.process_schedule(self.schedule_data)
+            elif command == 'reset_liquidsoap_bootstrap':
+                self.set_bootstrap_variables()
+            elif command == 'update_stream_setting':
+                self.logger.info("Updating stream setting...")
+                self.regenerate_liquidsoap_conf(m['setting'])
+            elif command == 'update_stream_format':
+                self.logger.info("Updating stream format...")
+                self.update_liquidsoap_stream_format(m['stream_format'])
+            elif command == 'update_station_name':
+                self.logger.info("Updating station name...")
+                self.update_liquidsoap_station_name(m['station_name'])
+            elif command == 'update_transition_fade':
+                self.logger.info("Updating transition_fade...")
+                self.update_liquidsoap_transition_fade(m['transition_fade'])
+            elif command == 'switch_source':
+                self.logger.info("switch_on_source show command received...")
+                self.pypo_liquidsoap.\
+                        get_telnet_dispatcher().\
+                        switch_source(m['sourcename'], m['status'])
+            elif command == 'disconnect_source':
+                self.logger.info("disconnect_on_source show command received...")
+                self.pypo_liquidsoap.get_telnet_dispatcher().\
+                        disconnect_source(m['sourcename'])
+            else:
+                self.logger.info("Unknown command: %s" % command)
+
+            # update timeout value
+            if command == 'update_schedule':
+                self.listener_timeout = POLL_INTERVAL
+            else:
+                self.listener_timeout = self.last_update_schedule_timestamp - time.time() + POLL_INTERVAL
+                if self.listener_timeout < 0:
+                    self.listener_timeout = 0
+            self.logger.info("New timeout: %s" % self.listener_timeout)
+        except Exception, e:
+            top = traceback.format_exc()
+            self.logger.error('Exception: %s', e)
+            self.logger.error("traceback: %s", top)
+            self.logger.error("Exception in handling Message Handler message: %s", e)
+
+
+    def switch_source_temp(self, sourcename, status):
+        self.logger.debug('Switching source: %s to "%s" status', sourcename, status)
+        command = "streams."
+        if sourcename == "master_dj":
+            command += "master_dj_"
+        elif sourcename == "live_dj":
+            command += "live_dj_"
+        elif sourcename == "scheduled_play":
+            command += "scheduled_play_"
+
+        if status == "on":
+            command += "start\n"
+        else:
+            command += "stop\n"
+
+        return command
+
+    """
+    Initialize Liquidsoap environment
+    """
+    def set_bootstrap_variables(self):
+        self.logger.debug('Getting information needed on bootstrap from Airtime')
+        try:
+            info = self.api_client.get_bootstrap_info()
+        except Exception, e:
+            self.logger.error('Unable to get bootstrap info.. Exiting pypo...')
+            self.logger.error(str(e))
+
+        self.logger.debug('info:%s', info)
+        commands = []
+        for k, v in info['switch_status'].iteritems():
+            commands.append(self.switch_source_temp(k, v))
+
+        stream_format = info['stream_label']
+        station_name = info['station_name']
+        fade = info['transition_fade']
+
+        commands.append(('vars.stream_metadata_type %s\n' % stream_format).encode('utf-8'))
+        commands.append(('vars.station_name %s\n' % station_name).encode('utf-8'))
+        commands.append(('vars.default_dj_fade %s\n' % fade).encode('utf-8'))
+        self.pypo_liquidsoap.get_telnet_dispatcher().telnet_send(commands)
+
+        self.pypo_liquidsoap.clear_queue_tracker()
+
+    def restart_liquidsoap(self):
+        try:
+            self.telnet_lock.acquire()
+            self.logger.info("Restarting Liquidsoap")
+            subprocess.call('/etc/init.d/airtime-liquidsoap restart', shell=True)
+
+            #Wait here and poll Liquidsoap until it has started up
+            self.logger.info("Waiting for Liquidsoap to start")
+            while True:
+                try:
+                    tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+                    tn.write("exit\n")
+                    tn.read_all()
+                    self.logger.info("Liquidsoap is up and running")
+                    break
+                except Exception, e:
+                    #sleep 0.5 seconds and try again
+                    time.sleep(0.5)
+
+        except Exception, e:
+            self.logger.error(e)
+        finally:
+            self.telnet_lock.release()
+
+        try:
+            self.set_bootstrap_variables()
+            #get the most up to date schedule, which will #initiate the process
+            #of making sure Liquidsoap is playing the schedule
+            self.persistent_manual_schedule_fetch(max_attempts=5)
+        except Exception, e:
+            self.logger.error(str(e))
+
+    """
+    TODO: This function needs to be way shorter, and refactored :/ - MK
+    """
+    def regenerate_liquidsoap_conf(self, setting):
+        existing = {}
+
+        setting = sorted(setting.items())
+        try:
+            fh = open('/etc/airtime/liquidsoap.cfg', 'r')
+        except IOError, e:
+            #file does not exist
+            self.restart_liquidsoap()
+            return
+
+        self.logger.info("Reading existing config...")
+        # read existing conf file and build dict
+        while True:
+            line = fh.readline()
+
+            # empty line means EOF
+            if not line:
+                break
+
+            line = line.strip()
+
+            if not len(line) or line[0] == "#":
+                continue
+
+            try:
+                key, value = line.split('=', 1)
+            except ValueError:
+                continue
+            key = key.strip()
+            value = value.strip()
+            value = value.replace('"', '')
+            if value == '' or value == "0":
+                value = ''
+            existing[key] = value
+        fh.close()
+
+        # dict flag for any change in config
+        change = {}
+        # this flag is to detect disable -> disable change
+        # in that case, we don't want to restart even if there are changes.
+        state_change_restart = {}
+        #restart flag
+        restart = False
+
+        self.logger.info("Looking for changes...")
+        # look for changes
+        for k, s in setting:
+            if "output_sound_device" in s[u'keyname'] or "icecast_vorbis_metadata" in s[u'keyname']:
+                dump, stream = s[u'keyname'].split('_', 1)
+                state_change_restart[stream] = False
+                # This is the case where restart is required no matter what
+                if (existing[s[u'keyname']] != str(s[u'value'])):
+                    self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
+                    restart = True;
+            elif "master_live_stream_port" in s[u'keyname'] or "master_live_stream_mp" in s[u'keyname'] or "dj_live_stream_port" in s[u'keyname'] or "dj_live_stream_mp" in s[u'keyname'] or "off_air_meta" in s[u'keyname']:
+                if (existing[s[u'keyname']] != s[u'value']):
+                    self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
+                    restart = True;
+            else:
+                stream, dump = s[u'keyname'].split('_', 1)
+                if "_output" in s[u'keyname']:
+                    if (existing[s[u'keyname']] != s[u'value']):
+                        self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
+                        restart = True;
+                        state_change_restart[stream] = True
+                    elif (s[u'value'] != 'disabled'):
+                        state_change_restart[stream] = True
+                    else:
+                        state_change_restart[stream] = False
+                else:
+                    # setting inital value
+                    if stream not in change:
+                        change[stream] = False
+                    if not (s[u'value'] == existing[s[u'keyname']]):
+                        self.logger.info("Keyname: %s, Current value: %s, New Value: %s", s[u'keyname'], existing[s[u'keyname']], s[u'value'])
+                        change[stream] = True
+
+        # set flag change for sound_device alway True
+        self.logger.info("Change:%s, State_Change:%s...", change, state_change_restart)
+
+        for k, v in state_change_restart.items():
+            if k == "sound_device" and v:
+                restart = True
+            elif v and change[k]:
+                self.logger.info("'Need-to-restart' state detected for %s...", k)
+                restart = True
+        # rewrite
+        if restart:
+            self.restart_liquidsoap()
+        else:
+            self.logger.info("No change detected in setting...")
+            self.update_liquidsoap_connection_status()
+
+    def update_liquidsoap_connection_status(self):
+        """
+        updates the status of Liquidsoap connection to the streaming server
+        This function updates the bootup time variable in Liquidsoap script
+        """
+
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+            # update the boot up time of Liquidsoap. Since Liquidsoap is not restarting,
+            # we are manually adjusting the bootup time variable so the status msg will get
+            # updated.
+            current_time = time.time()
+            boot_up_time_command = "vars.bootup_time " + str(current_time) + "\n"
+            self.logger.info(boot_up_time_command)
+            tn.write(boot_up_time_command)
+
+            connection_status = "streams.connection_status\n"
+            self.logger.info(connection_status)
+            tn.write(connection_status)
+
+            tn.write('exit\n')
+
+            output = tn.read_all()
+        except Exception, e:
+            self.logger.error(str(e))
+        finally:
+            self.telnet_lock.release()
+
+        output_list = output.split("\r\n")
+        stream_info = output_list[2]
+
+        # streamin info is in the form of:
+        # eg. s1:true,2:true,3:false
+        streams = stream_info.split(",")
+        self.logger.info(streams)
+
+        fake_time = current_time + 1
+        for s in streams:
+            info = s.split(':')
+            stream_id = info[0]
+            status = info[1]
+            if(status == "true"):
+                self.api_client.notify_liquidsoap_status("OK", stream_id, str(fake_time))
+
+
+    def update_liquidsoap_stream_format(self, stream_format):
+        # Push stream metadata to liquidsoap
+        # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+            command = ('vars.stream_metadata_type %s\n' % stream_format).encode('utf-8')
+            self.logger.info(command)
+            tn.write(command)
+            tn.write('exit\n')
+            tn.read_all()
+        except Exception, e:
+            self.logger.error("Exception %s", e)
+        finally:
+            self.telnet_lock.release()
+
+    def update_liquidsoap_transition_fade(self, fade):
+        # Push stream metadata to liquidsoap
+        # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+            command = ('vars.default_dj_fade %s\n' % fade).encode('utf-8')
+            self.logger.info(command)
+            tn.write(command)
+            tn.write('exit\n')
+            tn.read_all()
+        except Exception, e:
+            self.logger.error("Exception %s", e)
+        finally:
+            self.telnet_lock.release()
+
+    def update_liquidsoap_station_name(self, station_name):
+        # Push stream metadata to liquidsoap
+        # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
+        try:
+            try:
+                self.telnet_lock.acquire()
+                tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+                command = ('vars.station_name %s\n' % station_name).encode('utf-8')
+                self.logger.info(command)
+                tn.write(command)
+                tn.write('exit\n')
+                tn.read_all()
+            except Exception, e:
+                self.logger.error(str(e))
+            finally:
+                self.telnet_lock.release()
+        except Exception, e:
+            self.logger.error("Exception %s", e)
+
+    """
+    Process the schedule
+     - Reads the scheduled entries of a given range (actual time +/- "prepare_ahead" / "cache_for")
+     - Saves a serialized file of the schedule
+     - playlists are prepared. (brought to liquidsoap format) and, if not mounted via nsf, files are copied
+       to the cache dir (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss)
+     - runs the cleanup routine, to get rid of unused cached files
+    """
+    def process_schedule(self, schedule_data):
+        self.last_update_schedule_timestamp = time.time()
+        self.logger.debug(schedule_data)
+        media = schedule_data["media"]
+        media_filtered = {}
+
+        # Download all the media and put playlists in liquidsoap "annotate" format
+        try:
+
+            """
+            Make sure cache_dir exists
+            """
+            download_dir = self.cache_dir
+            try:
+                os.makedirs(download_dir)
+            except Exception, e:
+                pass
+
+            media_copy = {}
+            for key in media:
+                media_item = media[key]
+                if (media_item['type'] == 'file'):
+                    self.sanity_check_media_item(media_item)
+                    fileExt = os.path.splitext(media_item['uri'])[1]
+                    dst = os.path.join(download_dir, unicode(media_item['id']) + fileExt)
+                    media_item['dst'] = dst
+                    media_item['file_ready'] = False
+                    media_filtered[key] = media_item
+
+                media_item['start'] = datetime.strptime(media_item['start'], "%Y-%m-%d-%H-%M-%S")
+                media_item['end'] = datetime.strptime(media_item['end'], "%Y-%m-%d-%H-%M-%S")
+                media_copy[media_item['start']] = media_item
+
+
+            self.media_prepare_queue.put(copy.copy(media_filtered))
+        except Exception, e: self.logger.error("%s", e)
+
+        # Send the data to pypo-push
+        self.logger.debug("Pushing to pypo-push")
+        self.push_queue.put(media_copy)
+
+
+        # cleanup
+        try: self.cache_cleanup(media)
+        except Exception, e: self.logger.error("%s", e)
+
+    #do basic validation of file parameters. Useful for debugging
+    #purposes
+    def sanity_check_media_item(self, media_item):
+        start = datetime.strptime(media_item['start'], "%Y-%m-%d-%H-%M-%S")
+        end = datetime.strptime(media_item['end'], "%Y-%m-%d-%H-%M-%S")
+
+        length1 = pure.date_interval_to_seconds(end - start)
+        length2 = media_item['cue_out'] - media_item['cue_in']
+
+        if abs(length2 - length1) > 1:
+            self.logger.error("end - start length: %s", length1)
+            self.logger.error("cue_out - cue_in length: %s", length2)
+            self.logger.error("Two lengths are not equal!!!")
+
+    def is_file_opened(self, path):
+        #Capture stderr to avoid polluting py-interpreter.log
+        proc = Popen(["lsof", path], stdout=PIPE, stderr=PIPE)
+        out = proc.communicate()[0].strip()
+        return bool(out)
+
+    def cache_cleanup(self, media):
+        """
+        Get list of all files in the cache dir and remove them if they aren't being used anymore.
+        Input dict() media, lists all files that are scheduled or currently playing. Not being in this
+        dict() means the file is safe to remove.
+        """
+        cached_file_set = set(os.listdir(self.cache_dir))
+        scheduled_file_set = set()
+
+        for mkey in media:
+            media_item = media[mkey]
+            if media_item['type'] == 'file':
+                fileExt = os.path.splitext(media_item['uri'])[1]
+                scheduled_file_set.add(unicode(media_item["id"]) + fileExt)
+
+        expired_files = cached_file_set - scheduled_file_set
+
+        self.logger.debug("Files to remove " + str(expired_files))
+        for f in expired_files:
+            try:
+                path = os.path.join(self.cache_dir, f)
+                self.logger.debug("Removing %s" % path)
+
+                #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(path):
+                    os.remove(path)
+                    self.logger.info("File '%s' removed" % path)
+                else:
+                    self.logger.info("File '%s' not removed. Still busy!" % path)
+            except Exception, e:
+                self.logger.error("Problem removing file '%s'" % f)
+                self.logger.error(traceback.format_exc())
+
+    def manual_schedule_fetch(self):
+        success, self.schedule_data = self.api_client.get_schedule()
+        if success:
+            self.process_schedule(self.schedule_data)
+        return success
+
+    def persistent_manual_schedule_fetch(self, max_attempts=1):
+        success = False
+        num_attempts = 0
+        while not success and num_attempts < max_attempts:
+            success = self.manual_schedule_fetch()
+            num_attempts += 1
+
+        return success
+
+
+    def main(self):
+        #Make sure all Liquidsoap queues are empty. This is important in the
+        #case where we've just restarted the pypo scheduler, but Liquidsoap still
+        #is playing tracks. In this case let's just restart everything from scratch
+        #so that we can repopulate our dictionary that keeps track of what 
+        #Liquidsoap is playing much more easily.
+        self.pypo_liquidsoap.clear_all_queues()
+
+        self.set_bootstrap_variables()
+
+        # Bootstrap: since we are just starting up, we need to grab the
+        # most recent schedule.  After that we can just wait for updates.
+        success = self.persistent_manual_schedule_fetch(max_attempts=5)
+
+        if success:
+            self.logger.info("Bootstrap schedule received: %s", self.schedule_data)
+
+        loops = 1
+        while True:
+            self.logger.info("Loop #%s", loops)
+            try:
+                """
+                our simple_queue.get() requires a timeout, in which case we
+                fetch the Airtime schedule manually. It is important to fetch
+                the schedule periodically because if we didn't, we would only
+                get schedule updates via RabbitMq if the user was constantly
+                using the Airtime interface.
+
+                If the user is not using the interface, RabbitMq messages are not
+                sent, and we will have very stale (or non-existent!) data about the
+                schedule.
+
+                Currently we are checking every POLL_INTERVAL seconds
+                """
+
+
+                message = self.fetch_queue.get(block=True, timeout=self.listener_timeout)
+                self.handle_message(message)
+            except Empty, e:
+                self.logger.info("Queue timeout. Fetching schedule manually")
+                self.persistent_manual_schedule_fetch(max_attempts=5)
+            except Exception, e:
+                top = traceback.format_exc()
+                self.logger.error('Exception: %s', e)
+                self.logger.error("traceback: %s", top)
+
+            loops += 1
+
+    def run(self):
+        """
+        Entry point of the thread
+        """
+        self.main()

python_apps/pypo/schedule/pypofile.py

@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+
+from threading import Thread
+from Queue import Empty
+
+import logging
+import shutil
+import os
+import sys
+import stat
+
+from std_err_override import LogWriter
+
+# configure logging
+logging_cfg = os.path.join(os.path.dirname(__file__), "../logging.cfg")
+logging.config.fileConfig(logging_cfg)
+logger = logging.getLogger()
+LogWriter.override_std_err(logger)
+
+#need to wait for Python 2.7 for this..
+#logging.captureWarnings(True)
+
+
+class PypoFile(Thread):
+
+    def __init__(self, schedule_queue, config):
+        Thread.__init__(self)
+        self.logger = logging.getLogger()
+        self.media_queue = schedule_queue
+        self.media = None
+        self.cache_dir = os.path.join(config["cache_dir"], "scheduler")
+
+    def copy_file(self, media_item):
+        """
+        Copy media_item from local library directory to local cache directory.
+        """
+        src = media_item['uri']
+        dst = media_item['dst']
+
+        try:
+            src_size = os.path.getsize(src)
+        except Exception, e:
+            self.logger.error("Could not get size of source file: %s", src)
+            return
+
+        dst_exists = True
+        try:
+            dst_size = os.path.getsize(dst)
+        except Exception, e:
+            dst_exists = False
+
+        do_copy = False
+        if dst_exists:
+            if src_size != dst_size:
+                do_copy = True
+            else:
+                self.logger.debug("file %s already exists in local cache as %s, skipping copying..." % (src, dst))
+        else:
+            do_copy = True
+
+        media_item['file_ready'] = not do_copy
+
+        if do_copy:
+            self.logger.debug("copying from %s to local cache %s" % (src, dst))
+            try:
+
+                """
+                List file as "ready" before it starts copying because by the time
+                Liquidsoap is ready to play this file, it should have at least started
+                copying (and can continue copying while Liquidsoap reads from the beginning
+                of the file)
+                """
+                media_item['file_ready'] = True
+
+                """
+                copy will overwrite dst if it already exists
+                """
+                shutil.copy(src, dst)
+
+                #make file world readable
+                os.chmod(dst, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
+            except Exception, e:
+                self.logger.error("Could not copy from %s to %s" % (src, dst))
+                self.logger.error(e)
+
+    def get_highest_priority_media_item(self, schedule):
+        """
+        Get highest priority media_item in the queue. Currently the highest
+        priority is decided by how close the start time is to "now".
+        """
+        if schedule is None or len(schedule) == 0:
+            return None
+
+        sorted_keys = sorted(schedule.keys())
+
+        if len(sorted_keys) == 0:
+            return None
+
+        highest_priority = sorted_keys[0]
+        media_item = schedule[highest_priority]
+
+        self.logger.debug("Highest priority item: %s" % highest_priority)
+
+        """
+        Remove this media_item from the dictionary. On the next iteration
+        (from the main function) we won't consider it for prioritization 
+        anymore. If on the next iteration we have received a new schedule,
+        it is very possible we will have to deal with the same media_items 
+        again. In this situation, the worst possible case is that we try to
+        copy the file again and realize we already have it (thus aborting the copy). 
+        """
+        del schedule[highest_priority]
+
+        return media_item
+
+
+    def main(self):
+        while True:
+            try:
+                if self.media is None or len(self.media) == 0:
+                    """
+                    We have no schedule, so we have nothing else to do. Let's
+                    do a blocked wait on the queue
+                    """
+                    self.media = self.media_queue.get(block=True)
+                else:
+                    """
+                    We have a schedule we need to process, but we also want
+                    to check if a newer schedule is available. In this case
+                    do a non-blocking queue.get and in either case (we get something
+                    or we don't), get back to work on preparing getting files.
+                    """
+                    try:
+                        self.media = self.media_queue.get_nowait()
+                    except Empty, e:
+                        pass
+
+
+                media_item = self.get_highest_priority_media_item(self.media)
+                if media_item is not None:
+                    self.copy_file(media_item)
+            except Exception, e:
+                import traceback
+                top = traceback.format_exc()
+                self.logger.error(str(e))
+                self.logger.error(top)
+                raise
+
+    def run(self):
+        """
+        Entry point of the thread
+        """
+        self.main()

python_apps/pypo/schedule/pypoliqqueue.py

@@ -0,0 +1,87 @@
+from threading import Thread
+from collections import deque
+from datetime import datetime
+
+import traceback
+import sys
+import time
+
+
+from Queue import Empty
+
+import signal
+def keyboardInterruptHandler(signum, frame):
+    logger = logging.getLogger()
+    logger.info('\nKeyboard Interrupt\n')
+    sys.exit(0)
+signal.signal(signal.SIGINT, keyboardInterruptHandler)
+
+class PypoLiqQueue(Thread):
+    def __init__(self, q, pypo_liquidsoap, logger):
+        Thread.__init__(self)
+        self.queue = q
+        self.logger = logger
+        self.pypo_liquidsoap = pypo_liquidsoap
+
+    def main(self):
+        time_until_next_play = None
+        schedule_deque = deque()
+        media_schedule = None
+
+        while True:
+            try:
+                if time_until_next_play is None:
+                    self.logger.info("waiting indefinitely for schedule")
+                    media_schedule = self.queue.get(block=True)
+                else:
+                    self.logger.info("waiting %ss until next scheduled item" % \
+                            time_until_next_play)
+                    media_schedule = self.queue.get(block=True, \
+                            timeout=time_until_next_play)
+            except Empty, e:
+                #Time to push a scheduled item.
+                media_item = schedule_deque.popleft()
+                self.pypo_liquidsoap.play(media_item)
+                if len(schedule_deque):
+                    time_until_next_play = \
+                            self.date_interval_to_seconds(
+                                    schedule_deque[0]['start'] - datetime.utcnow())
+                    if time_until_next_play < 0:
+                        time_until_next_play = 0
+                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()
+
+                keys = sorted(media_schedule.keys())
+                for i in keys:
+                    schedule_deque.append(media_schedule[i])
+
+                if len(keys):
+                    time_until_next_play = self.date_interval_to_seconds(\
+                            keys[0] - datetime.utcnow())
+                else:
+                    time_until_next_play = None
+
+
+    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
+
+    def run(self):
+        try: self.main()
+        except Exception, e:
+            self.logger.error('PypoLiqQueue Exception: %s', traceback.format_exc())
+
+
+

python_apps/pypo/schedule/pypoliquidsoap.py

@@ -0,0 +1,237 @@
+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,\
+                self.liq_queue_tracker.keys())
+
+    def get_telnet_dispatcher(self):
+        return self.telnet_liquidsoap
+
+
+    def play(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":
+            self.telnet_liquidsoap.disconnect_source("live_dj")
+        elif media_item['event_type'] == "switch_off":
+            self.telnet_liquidsoap.switch_source("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 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.
+
+        #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: x["row_id"], scheduled_now_files))
+
+        row_id_map = {}
+        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"])
+                row_id_map[mi["row_id"]] = mi
+
+        to_be_removed = set()
+        to_be_added = set()
+
+        #Iterate over the new files, and compare them to currently scheduled
+        #tracks. If already in liquidsoap queue still need to make sure they don't
+        #have different attributes such replay_gain etc.
+        for i in scheduled_now_files:
+            if i["row_id"] in row_id_map:
+                mi = row_id_map[i["row_id"]]
+                correct = mi['start'] == i['start'] and \
+                        mi['end'] == i['end'] and \
+                        mi['row_id'] == i['row_id'] and \
+                        mi['replay_gain'] == i['replay_gain']
+
+                if not correct:
+                    #need to re-add
+                    self.logger.info("Track %s found to have new attr." % i)
+                    to_be_removed.add(i["row_id"])
+                    to_be_added.add(i["row_id"])
+
+
+        to_be_removed.update(liq_queue_ids - schedule_ids)
+        to_be_added.update(schedule_ids - liq_queue_ids)
+
+        if 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)
+
+        if 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)
+
+        #handle webstreams
+        current_stream_id = self.telnet_liquidsoap.get_current_stream_id()
+        if scheduled_now_webstream:
+            if current_stream_id != scheduled_now_webstream[0]:
+                self.play(scheduled_now_webstream[0])
+        elif current_stream_id != "-1":
+            #something is playing and it shouldn't be.
+            self.telnet_liquidsoap.stop_web_stream_buffer()
+            self.telnet_liquidsoap.stop_web_stream_output()
+
+    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 clear_queue_tracker(self):
+        for i in self.liq_queue_tracker.keys():
+            self.liq_queue_tracker[i] = None
+
+    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
+
+    def clear_all_queues(self):
+        self.telnet_liquidsoap.queue_clear_all()
+
+
+class UnknownMediaItemType(Exception):
+    pass
+
+class NoQueueAvailableException(Exception):
+    pass

python_apps/pypo/schedule/pypopush.py

@@ -0,0 +1,196 @@
+# -*- coding: utf-8 -*-
+
+from datetime import datetime
+from datetime import timedelta
+
+import sys
+import time
+import logging.config
+import telnetlib
+import calendar
+import math
+import traceback
+import os
+
+from pypofetch import PypoFetch
+from pypoliqqueue import PypoLiqQueue
+
+from Queue import Empty, Queue
+
+from threading import Thread
+
+from api_clients import api_client
+from std_err_override import LogWriter
+from configobj import ConfigObj
+
+
+# configure logging
+logging_cfg = os.path.join(os.path.dirname(__file__), "../logging.cfg")
+logging.config.fileConfig(logging_cfg)
+logger = logging.getLogger()
+LogWriter.override_std_err(logger)
+
+#need to wait for Python 2.7 for this..
+#logging.captureWarnings(True)
+
+PUSH_INTERVAL = 2
+
+
+def is_stream(media_item):
+    return media_item['type'] == 'stream_output_start'
+
+def is_file(media_item):
+    return media_item['type'] == 'file'
+
+class PypoPush(Thread):
+    def __init__(self, q, telnet_lock, pypo_liquidsoap, config):
+        Thread.__init__(self)
+        self.api_client = api_client.AirtimeApiClient()
+        self.queue = q
+
+        self.telnet_lock = telnet_lock
+        self.config = config
+
+        self.pushed_objects = {}
+        self.logger = logging.getLogger('push')
+        self.current_prebuffering_stream_id = None
+        self.queue_id = 0
+
+        self.future_scheduled_queue = Queue()
+        self.pypo_liquidsoap = pypo_liquidsoap
+
+        self.plq = PypoLiqQueue(self.future_scheduled_queue, \
+                self.pypo_liquidsoap, \
+                self.logger)
+        self.plq.daemon = True
+        self.plq.start()
+
+
+    def main(self):
+        loops = 0
+        heartbeat_period = math.floor(30 / PUSH_INTERVAL)
+
+        media_schedule = None
+
+        while True:
+            try:
+                media_schedule = self.queue.get(block=True)
+            except Exception, e:
+                self.logger.error(str(e))
+                raise
+            else:
+                self.logger.debug(media_schedule)
+                #separate media_schedule list into currently_playing and
+                #scheduled_for_future lists
+                currently_playing, scheduled_for_future = \
+                        self.separate_present_future(media_schedule)
+
+                self.pypo_liquidsoap.verify_correct_present_media(currently_playing)
+                self.future_scheduled_queue.put(scheduled_for_future)
+
+            if loops % heartbeat_period == 0:
+                self.logger.info("heartbeat")
+                loops = 0
+            loops += 1
+
+
+    def separate_present_future(self, media_schedule):
+        tnow = datetime.utcnow()
+
+        present = []
+        future = {}
+
+        sorted_keys = sorted(media_schedule.keys())
+        for mkey in sorted_keys:
+            media_item = media_schedule[mkey]
+
+            diff_td = tnow - media_item['start']
+            diff_sec = self.date_interval_to_seconds(diff_td)
+
+            if diff_sec >= 0:
+                present.append(media_item)
+            else:
+                future[media_item['start']] = media_item
+
+        return present, future
+
+    def get_current_stream_id_from_liquidsoap(self):
+        response = "-1"
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.config['ls_host'], self.config['ls_port'])
+
+            msg = 'dynamic_source.get_id\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()
+
+        return response
+
+    #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
+
+    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)
+
+        return seconds
+
+    def stop_web_stream_all(self):
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.config['LS_HOST'], self.config['LS_PORT'])
+
+            #msg = 'dynamic_source.read_stop_all xxx\n'
+            msg = 'http.stop\n'
+            self.logger.debug(msg)
+            tn.write(msg)
+
+            msg = 'dynamic_source.output_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 run(self):
+        try: self.main()
+        except Exception, e:
+            top = traceback.format_exc()
+            self.logger.error('Pypo Push Exception: %s', top)
+

python_apps/pypo/schedule/telnetliquidsoap.py

@@ -0,0 +1,289 @@
+import telnetlib
+
+def create_liquidsoap_annotation(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'])
+
+class TelnetLiquidsoap:
+
+    def __init__(self, telnet_lock, logger, ls_host, ls_port, queues):
+        self.telnet_lock = telnet_lock
+        self.ls_host = ls_host
+        self.ls_port = ls_port
+        self.logger = logger
+        self.queues = queues
+        self.current_prebuffering_stream_id = None
+
+    def __connect(self):
+        return telnetlib.Telnet(self.ls_host, self.ls_port)
+
+    def __is_empty(self, tn, queue_id):
+        return True
+
+    def queue_clear_all(self):
+        try:
+            self.telnet_lock.acquire()
+            tn = self.__connect()
+
+            for i in self.queues:
+                msg = 'queues.%s_skip\n' % i
+                self.logger.debug(msg)
+                tn.write(msg)
+            
+            tn.write("exit\n")
+            self.logger.debug(tn.read_all())
+        except Exception:
+            raise
+        finally:
+            self.telnet_lock.release()
+
+    def queue_remove(self, queue_id):
+        try:
+            self.telnet_lock.acquire()
+            tn = self.__connect()
+
+            msg = 'queues.%s_skip\n' % queue_id
+            self.logger.debug(msg)
+            tn.write(msg)
+            
+            tn.write("exit\n")
+            self.logger.debug(tn.read_all())
+        except Exception:
+            raise
+        finally:
+            self.telnet_lock.release()
+
+
+    def queue_push(self, queue_id, media_item):
+        try:
+            self.telnet_lock.acquire()
+            tn = self.__connect()
+
+            if not self.__is_empty(tn, queue_id):
+                raise QueueNotEmptyException()
+
+            annotation = create_liquidsoap_annotation(media_item)
+            msg = '%s.push %s\n' % (queue_id, annotation.encode('utf-8'))
+            self.logger.debug(msg)
+            tn.write(msg)
+
+            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:
+            raise
+        finally:
+            self.telnet_lock.release()
+
+
+    def stop_web_stream_buffer(self):
+        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 = '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):
+        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 = '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 get_current_stream_id(self):
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.ls_host, self.ls_port)
+
+            msg = 'dynamic_source.get_id\n'
+            self.logger.debug(msg)
+            tn.write(msg)
+
+            tn.write("exit\n")
+            stream_id = tn.read_all().splitlines()[0]
+            self.logger.debug("stream_id: %s" % stream_id)
+
+            return stream_id
+        except Exception, e:
+            self.logger.error(str(e))
+        finally:
+            self.telnet_lock.release()
+
+    def disconnect_source(self, sourcename):
+        self.logger.debug('Disconnecting source: %s', sourcename)
+        command = ""
+        if(sourcename == "master_dj"):
+            command += "master_harbor.kick\n"
+        elif(sourcename == "live_dj"):
+            command += "live_dj_harbor.kick\n"
+
+        try:
+            self.telnet_lock.acquire()
+            tn = telnetlib.Telnet(self.ls_host, self.ls_port)
+            self.logger.info(command)
+            tn.write(command)
+            tn.write('exit\n')
+            tn.read_all()
+        except Exception, e:
+            self.logger.error(traceback.format_exc())
+        finally:
+            self.telnet_lock.release()
+
+    def telnet_send(self, commands):
+        try:
+            self.telnet_lock.acquire()
+
+            tn = telnetlib.Telnet(self.ls_host, self.ls_port)
+            for i in commands:
+                self.logger.info(i)
+                tn.write(i)
+
+            tn.write('exit\n')
+            tn.read_all()
+        except Exception, e:
+            self.logger.error(str(e))
+        finally:
+            self.telnet_lock.release()
+
+
+    def switch_source(self, sourcename, status):
+        self.logger.debug('Switching source: %s to "%s" status', sourcename, status)
+        command = "streams."
+        if sourcename == "master_dj":
+            command += "master_dj_"
+        elif sourcename == "live_dj":
+            command += "live_dj_"
+        elif sourcename == "scheduled_play":
+            command += "scheduled_play_"
+
+        if status == "on":
+            command += "start\n"
+        else:
+            command += "stop\n"
+
+        self.telnet_send([command])
+
+class DummyTelnetLiquidsoap:
+
+    def __init__(self, telnet_lock, logger):
+        self.telnet_lock = telnet_lock
+        self.liquidsoap_mock_queues = {}
+        self.logger = logger
+
+        for i in range(4):
+            self.liquidsoap_mock_queues["s"+str(i)] = []
+
+    def queue_push(self, queue_id, media_item):
+        try:
+            self.telnet_lock.acquire()
+
+            self.logger.info("Pushing %s to queue %s" % (media_item, queue_id))
+            from datetime import datetime
+            print "Time now: %s" % datetime.utcnow()
+
+            annotation = create_liquidsoap_annotation(media_item)
+            self.liquidsoap_mock_queues[queue_id].append(annotation)
+        except Exception:
+            raise
+        finally:
+            self.telnet_lock.release()
+
+    def queue_remove(self, queue_id):
+        try:
+            self.telnet_lock.acquire()
+
+            self.logger.info("Purging queue %s" % queue_id)
+            from datetime import datetime
+            print "Time now: %s" % datetime.utcnow()
+
+        except Exception:
+            raise
+        finally:
+            self.telnet_lock.release()
+
+class QueueNotEmptyException(Exception):
+    pass