Merge branch '2.1.x' into devel
Conflicts: python_apps/pypo/liquidsoap_scripts/ls_script.liq
This commit is contained in:
denise
commit
335dd1516b
40 changed files with 1084 additions and 870 deletions
|
|
@ -10,6 +10,7 @@
|
|||
### END INIT INFO
|
||||
|
||||
USERID=pypo
|
||||
ROOTUSERID=root
|
||||
GROUPID=pypo
|
||||
NAME=Airtime\ Playout
|
||||
|
||||
|
|
@ -47,7 +48,7 @@ start () {
|
|||
chown pypo:pypo /etc/airtime
|
||||
chown pypo:pypo /etc/airtime/liquidsoap.cfg
|
||||
|
||||
start-stop-daemon --start --background --quiet --chuid $USERID:$GROUPID --make-pidfile --pidfile $PIDFILE0 --startas $DAEMON0
|
||||
start-stop-daemon --start --background --quiet --chuid $ROOTUSERID:$ROOTUSERID --make-pidfile --pidfile $PIDFILE0 --startas $DAEMON0
|
||||
monit monitor airtime-playout >/dev/null 2>&1
|
||||
|
||||
liquidsoap_start
|
||||
|
|
@ -80,7 +81,7 @@ monit_restart() {
|
|||
}
|
||||
|
||||
start_no_monit() {
|
||||
start-stop-daemon --start --background --quiet --chuid $USERID:$GROUPID --make-pidfile --pidfile $PIDFILE0 --startas $DAEMON0
|
||||
start-stop-daemon --start --background --quiet --chuid $ROOTUSERID:$ROOTUSERID --make-pidfile --pidfile $PIDFILE0 --startas $DAEMON0
|
||||
liquidsoap_start
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -109,20 +109,6 @@ try:
|
|||
print e
|
||||
sys.exit(1)
|
||||
|
||||
"""
|
||||
logging.basicConfig(format='%(message)s')
|
||||
|
||||
#generate liquidsoap config file
|
||||
#access the DB and generate liquidsoap.cfg under /etc/airtime/
|
||||
ac = api_client.api_client_factory(config, logging.getLogger())
|
||||
ss = ac.get_stream_setting()
|
||||
|
||||
if ss is not None:
|
||||
generate_liquidsoap_config(ss)
|
||||
else:
|
||||
print "Unable to connect to the Airtime server."
|
||||
"""
|
||||
|
||||
#initialize init.d scripts
|
||||
subprocess.call("update-rc.d airtime-playout defaults >/dev/null 2>&1", shell=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -130,27 +130,29 @@ end
|
|||
|
||||
def append_dj_inputs(master_harbor_input_port, master_harbor_input_mount_point, dj_harbor_input_port, dj_harbor_input_mount_point, s) =
|
||||
if master_harbor_input_port != 0 and master_harbor_input_mount_point != "" and dj_harbor_input_port != 0 and dj_harbor_input_mount_point != "" then
|
||||
master_dj = mksafe(input.harbor(id="master_harbor", master_harbor_input_mount_point, port=master_harbor_input_port, auth=check_master_dj_client,
|
||||
max=40., on_connect=master_dj_connect, on_disconnect=master_dj_disconnect))
|
||||
dj_live = mksafe(input.harbor(id="live_dj_harbor", dj_harbor_input_mount_point, port=dj_harbor_input_port, auth=check_dj_client,
|
||||
max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect))
|
||||
|
||||
master_dj = mksafe(audio_to_stereo(input.harbor(id="master_harbor", master_harbor_input_mount_point, port=master_harbor_input_port, auth=check_master_dj_client,
|
||||
max=40., on_connect=master_dj_connect, on_disconnect=master_dj_disconnect)))
|
||||
dj_live = mksafe(audio_to_stereo(input.harbor(id="live_dj_harbor", dj_harbor_input_mount_point, port=dj_harbor_input_port, auth=check_dj_client,
|
||||
max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect)))
|
||||
|
||||
master_dj = rewrite_metadata([("artist","Airtime"), ("title", "Master Dj")],master_dj)
|
||||
dj_live = rewrite_metadata([("artist","Airtime"), ("title", "Live Dj")],dj_live)
|
||||
|
||||
|
||||
ignore(output.dummy(master_dj, fallible=true))
|
||||
ignore(output.dummy(dj_live, fallible=true))
|
||||
switch(id="master_dj_switch", track_sensitive=false, transitions=[transition, transition, transition], [({!master_dj_enabled},master_dj), ({!live_dj_enabled},dj_live), ({true}, s)])
|
||||
elsif master_harbor_input_port != 0 and master_harbor_input_mount_point != "" then
|
||||
master_dj = mksafe(input.harbor(id="master_harbor", master_harbor_input_mount_point, port=master_harbor_input_port, auth=check_master_dj_client,
|
||||
max=40., on_connect=master_dj_connect, on_disconnect=master_dj_disconnect))
|
||||
master_dj = mksafe(audio_to_stereo(input.harbor(id="master_harbor", master_harbor_input_mount_point, port=master_harbor_input_port, auth=check_master_dj_client,
|
||||
max=40., on_connect=master_dj_connect, on_disconnect=master_dj_disconnect)))
|
||||
ignore(output.dummy(master_dj, fallible=true))
|
||||
master_dj = rewrite_metadata([("artist","Airtime"), ("title", "Master Dj")],master_dj)
|
||||
switch(id="master_dj_switch", track_sensitive=false, transitions=[transition, transition], [({!master_dj_enabled},master_dj), ({true}, s)])
|
||||
elsif dj_harbor_input_port != 0 and dj_harbor_input_mount_point != "" then
|
||||
dj_live = mksafe(input.harbor(id="live_dj_harbor", dj_harbor_input_mount_point, port=dj_harbor_input_port, auth=check_dj_client,
|
||||
max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect))
|
||||
dj_live = mksafe(audio_to_stereo(input.harbor(id="live_dj_harbor", dj_harbor_input_mount_point, port=dj_harbor_input_port, auth=check_dj_client,
|
||||
max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect)))
|
||||
|
||||
dj_live = rewrite_metadata([("artist","Airtime"), ("title", "Live Dj")],dj_live)
|
||||
|
||||
ignore(output.dummy(dj_live, fallible=true))
|
||||
switch(id="live_dj_switch", track_sensitive=false, transitions=[transition, transition], [({!live_dj_enabled},dj_live), ({true}, s)])
|
||||
else
|
||||
|
|
|
|||
|
|
@ -10,4 +10,4 @@ SCRIPT=`readlink -f $0`
|
|||
# Absolute path this script is in
|
||||
SCRIPTPATH=`dirname $SCRIPT`
|
||||
|
||||
cd ${SCRIPTPATH}/../ && python pypo-notify.py "$@"
|
||||
cd ${SCRIPTPATH}/../ && python pyponotify.py "$@"
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
[loggers]
|
||||
keys=root,fetch,push,recorder,message_h,notify
|
||||
keys=root,fetch,push,recorder,message_h
|
||||
|
||||
[handlers]
|
||||
keys=pypo,recorder,message_h,notify
|
||||
keys=pypo,recorder,message_h
|
||||
|
||||
[formatters]
|
||||
keys=simpleFormatter
|
||||
|
|
@ -35,18 +35,6 @@ handlers=message_h
|
|||
qualname=message_h
|
||||
propagate=0
|
||||
|
||||
[logger_notify]
|
||||
level=DEBUG
|
||||
handlers=notify
|
||||
qualname=notify
|
||||
propagate=0
|
||||
|
||||
[handler_notify]
|
||||
class=logging.handlers.RotatingFileHandler
|
||||
level=DEBUG
|
||||
formatter=simpleFormatter
|
||||
args=("/var/log/airtime/pypo/notify.log", 'a', 1000000, 5,)
|
||||
|
||||
[handler_pypo]
|
||||
class=logging.handlers.RotatingFileHandler
|
||||
level=DEBUG
|
||||
|
|
|
|||
28
python_apps/pypo/notify_logging.cfg
Normal file
28
python_apps/pypo/notify_logging.cfg
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
[loggers]
|
||||
keys=root,notify
|
||||
|
||||
[handlers]
|
||||
keys=notify
|
||||
|
||||
[formatters]
|
||||
keys=simpleFormatter
|
||||
|
||||
[logger_root]
|
||||
level=DEBUG
|
||||
handlers=notify
|
||||
|
||||
[logger_notify]
|
||||
level=DEBUG
|
||||
handlers=notify
|
||||
qualname=notify
|
||||
propagate=0
|
||||
|
||||
[handler_notify]
|
||||
class=logging.handlers.RotatingFileHandler
|
||||
level=DEBUG
|
||||
formatter=simpleFormatter
|
||||
args=("/var/log/airtime/pypo/notify.log", 'a', 1000000, 5,)
|
||||
|
||||
[formatter_simpleFormatter]
|
||||
format=%(asctime)s %(levelname)s - [%(filename)s : %(funcName)s() : line %(lineno)d] - %(message)s
|
||||
datefmt=
|
||||
|
|
@ -2,13 +2,15 @@
|
|||
Python part of radio playout (pypo)
|
||||
"""
|
||||
|
||||
from optparse import OptionParser
|
||||
from datetime import datetime
|
||||
|
||||
import telnetlib
|
||||
|
||||
import time
|
||||
from optparse import *
|
||||
import sys
|
||||
import signal
|
||||
import logging
|
||||
import logging.config
|
||||
import logging.handlers
|
||||
import locale
|
||||
import os
|
||||
from Queue import Queue
|
||||
|
|
@ -50,40 +52,6 @@ parser.add_option("-c", "--check", help="Check the cached schedule and exit", de
|
|||
#need to wait for Python 2.7 for this..
|
||||
#logging.captureWarnings(True)
|
||||
|
||||
def configure_locale():
|
||||
logger.debug("Before %s", locale.nl_langinfo(locale.CODESET))
|
||||
current_locale = locale.getlocale()
|
||||
|
||||
if current_locale[1] is None:
|
||||
logger.debug("No locale currently set. Attempting to get default locale.")
|
||||
default_locale = locale.getdefaultlocale()
|
||||
|
||||
if default_locale[1] is None:
|
||||
logger.debug("No default locale exists. Let's try loading from /etc/default/locale")
|
||||
if os.path.exists("/etc/default/locale"):
|
||||
config = ConfigObj('/etc/default/locale')
|
||||
lang = config.get('LANG')
|
||||
new_locale = lang
|
||||
else:
|
||||
logger.error("/etc/default/locale could not be found! Please run 'sudo update-locale' from command-line.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
new_locale = default_locale
|
||||
|
||||
logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale))
|
||||
|
||||
|
||||
|
||||
reload(sys)
|
||||
sys.setdefaultencoding("UTF-8")
|
||||
current_locale_encoding = locale.getlocale()[1].lower()
|
||||
logger.debug("sys default encoding %s", sys.getdefaultencoding())
|
||||
logger.debug("After %s", locale.nl_langinfo(locale.CODESET))
|
||||
|
||||
if current_locale_encoding not in ['utf-8', 'utf8']:
|
||||
logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding)
|
||||
sys.exit(1)
|
||||
|
||||
# configure logging
|
||||
try:
|
||||
logging.config.fileConfig("logging.cfg")
|
||||
|
|
@ -92,7 +60,42 @@ try:
|
|||
except Exception, e:
|
||||
print "Couldn't configure logging"
|
||||
sys.exit()
|
||||
|
||||
|
||||
def configure_locale():
|
||||
logger.debug("Before %s", locale.nl_langinfo(locale.CODESET))
|
||||
current_locale = locale.getlocale()
|
||||
|
||||
if current_locale[1] is None:
|
||||
logger.debug("No locale currently set. Attempting to get default locale.")
|
||||
default_locale = locale.getdefaultlocale()
|
||||
|
||||
if default_locale[1] is None:
|
||||
logger.debug("No default locale exists. Let's try loading from /etc/default/locale")
|
||||
if os.path.exists("/etc/default/locale"):
|
||||
locale_config = ConfigObj('/etc/default/locale')
|
||||
lang = locale_config.get('LANG')
|
||||
new_locale = lang
|
||||
else:
|
||||
logger.error("/etc/default/locale could not be found! Please run 'sudo update-locale' from command-line.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
new_locale = default_locale
|
||||
|
||||
logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale))
|
||||
|
||||
|
||||
|
||||
reload(sys)
|
||||
sys.setdefaultencoding("UTF-8")
|
||||
current_locale_encoding = locale.getlocale()[1].lower()
|
||||
logger.debug("sys default encoding %s", sys.getdefaultencoding())
|
||||
logger.debug("After %s", locale.nl_langinfo(locale.CODESET))
|
||||
|
||||
if current_locale_encoding not in ['utf-8', 'utf8']:
|
||||
logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
configure_locale()
|
||||
|
||||
# loading config file
|
||||
|
|
@ -105,63 +108,65 @@ except Exception, e:
|
|||
class Global:
|
||||
def __init__(self):
|
||||
self.api_client = api_client.api_client_factory(config)
|
||||
|
||||
|
||||
def selfcheck(self):
|
||||
self.api_client = api_client.api_client_factory(config)
|
||||
return self.api_client.is_server_compatible()
|
||||
|
||||
|
||||
def test_api(self):
|
||||
self.api_client.test()
|
||||
|
||||
"""
|
||||
def check_schedule(self):
|
||||
logger = logging.getLogger()
|
||||
|
||||
try:
|
||||
schedule_file = open(self.schedule_file, "r")
|
||||
schedule = pickle.load(schedule_file)
|
||||
schedule_file.close()
|
||||
|
||||
except Exception, e:
|
||||
logger.error("%s", e)
|
||||
schedule = None
|
||||
|
||||
for pkey in sorted(schedule.iterkeys()):
|
||||
playlist = schedule[pkey]
|
||||
print '*****************************************'
|
||||
print '\033[0;32m%s %s\033[m' % ('scheduled at:', str(pkey))
|
||||
print 'cached at : ' + self.cache_dir + str(pkey)
|
||||
print 'played: ' + str(playlist['played'])
|
||||
print 'schedule id: ' + str(playlist['schedule_id'])
|
||||
print 'duration: ' + str(playlist['duration'])
|
||||
print 'source id: ' + str(playlist['x_ident'])
|
||||
print '-----------------------------------------'
|
||||
|
||||
for media in playlist['medias']:
|
||||
print media
|
||||
"""
|
||||
|
||||
def keyboardInterruptHandler(signum, frame):
|
||||
logger = logging.getLogger()
|
||||
logger.info('\nKeyboard Interrupt\n')
|
||||
sys.exit(0)
|
||||
|
||||
def liquidsoap_running_test(telnet_lock, host, port, logger):
|
||||
logger.debug("Checking to see if Liquidsoap is running")
|
||||
success = True
|
||||
try:
|
||||
telnet_lock.acquire()
|
||||
tn = telnetlib.Telnet(host, port)
|
||||
msg = "version\n"
|
||||
tn.write(msg)
|
||||
tn.write("exit\n")
|
||||
logger.info("Liquidsoap version %s", tn.read_all())
|
||||
except Exception, e:
|
||||
logger.error(str(e))
|
||||
success = False
|
||||
finally:
|
||||
telnet_lock.release()
|
||||
|
||||
return success
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
logger = logging.getLogger()
|
||||
logger.info('###########################################')
|
||||
logger.info('# *** pypo *** #')
|
||||
logger.info('# Liquidsoap Scheduled Playout System #')
|
||||
logger.info('###########################################')
|
||||
|
||||
#Although all of our calculations are in UTC, it is useful to know what timezone
|
||||
#the local machine is, so that we have a reference for what time the actual
|
||||
#log entries were made
|
||||
logger.info("Timezone: %s" % str(time.tzname))
|
||||
logger.info("UTC time: %s" % str(datetime.utcnow()))
|
||||
|
||||
signal.signal(signal.SIGINT, keyboardInterruptHandler)
|
||||
|
||||
# initialize
|
||||
g = Global()
|
||||
|
||||
while not g.selfcheck(): time.sleep(5)
|
||||
|
||||
logger = logging.getLogger()
|
||||
while not g.selfcheck():
|
||||
time.sleep(5)
|
||||
|
||||
telnet_lock = Lock()
|
||||
|
||||
ls_host = config['ls_host']
|
||||
ls_port = config['ls_port']
|
||||
while not liquidsoap_running_test(telnet_lock, ls_host, ls_port, logger):
|
||||
logger.warning("Liquidsoap not started yet. Sleeping one second and trying again")
|
||||
time.sleep(1)
|
||||
|
||||
if options.test:
|
||||
g.test_api()
|
||||
|
|
@ -173,9 +178,9 @@ if __name__ == '__main__':
|
|||
pypoFetch_q = Queue()
|
||||
recorder_q = Queue()
|
||||
pypoPush_q = Queue()
|
||||
|
||||
telnet_lock = Lock()
|
||||
|
||||
|
||||
|
||||
|
||||
"""
|
||||
This queue is shared between pypo-fetch and pypo-file, where pypo-file
|
||||
is the receiver. Pypo-fetch will send every schedule it gets to pypo-file
|
||||
|
|
@ -183,19 +188,19 @@ if __name__ == '__main__':
|
|||
priority, and will retrieve it.
|
||||
"""
|
||||
media_q = Queue()
|
||||
|
||||
|
||||
pmh = PypoMessageHandler(pypoFetch_q, recorder_q)
|
||||
pmh.daemon = True
|
||||
pmh.start()
|
||||
|
||||
|
||||
pfile = PypoFile(media_q)
|
||||
pfile.daemon = True
|
||||
pfile.start()
|
||||
|
||||
|
||||
pf = PypoFetch(pypoFetch_q, pypoPush_q, media_q, telnet_lock)
|
||||
pf.daemon = True
|
||||
pf.start()
|
||||
|
||||
|
||||
pp = PypoPush(pypoPush_q, telnet_lock)
|
||||
pp.daemon = True
|
||||
pp.start()
|
||||
|
|
@ -204,23 +209,12 @@ if __name__ == '__main__':
|
|||
recorder.daemon = True
|
||||
recorder.start()
|
||||
|
||||
# all join() are commented out becase we want to exit entire pypo
|
||||
# all join() are commented out because we want to exit entire pypo
|
||||
# if pypofetch is exiting
|
||||
#pmh.join()
|
||||
#recorder.join()
|
||||
#pp.join()
|
||||
pf.join()
|
||||
|
||||
|
||||
logger.info("pypo fetch exit")
|
||||
sys.exit()
|
||||
"""
|
||||
if options.check:
|
||||
try: g.check_schedule()
|
||||
except Exception, e:
|
||||
print e
|
||||
|
||||
if options.cleanup:
|
||||
try: pf.cleanup('scheduler')
|
||||
except Exception, e:
|
||||
print e
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -3,14 +3,14 @@
|
|||
import os
|
||||
import sys
|
||||
import time
|
||||
import logging
|
||||
import logging.config
|
||||
import shutil
|
||||
import json
|
||||
import telnetlib
|
||||
import copy
|
||||
from threading import Thread
|
||||
|
||||
from Queue import Empty
|
||||
|
||||
from api_clients import api_client
|
||||
from std_err_override import LogWriter
|
||||
|
||||
|
|
@ -29,7 +29,9 @@ try:
|
|||
config = ConfigObj('/etc/airtime/pypo.cfg')
|
||||
LS_HOST = config['ls_host']
|
||||
LS_PORT = config['ls_port']
|
||||
POLL_INTERVAL = int(config['poll_interval'])
|
||||
#POLL_INTERVAL = int(config['poll_interval'])
|
||||
POLL_INTERVAL = 1800
|
||||
|
||||
|
||||
except Exception, e:
|
||||
logger.error('Error loading config file: %s', e)
|
||||
|
|
@ -43,12 +45,12 @@ class PypoFetch(Thread):
|
|||
self.push_queue = pypoPush_q
|
||||
self.media_prepare_queue = media_q
|
||||
self.last_update_schedule_timestamp = time.time()
|
||||
self.listener_timeout = 3600
|
||||
|
||||
self.listener_timeout = POLL_INTERVAL
|
||||
|
||||
self.telnet_lock = telnet_lock
|
||||
|
||||
|
||||
self.logger = logging.getLogger();
|
||||
|
||||
|
||||
self.cache_dir = os.path.join(config["cache_dir"], "scheduler")
|
||||
self.logger.debug("Cache dir %s", self.cache_dir)
|
||||
|
||||
|
|
@ -63,24 +65,24 @@ class PypoFetch(Thread):
|
|||
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:
|
||||
try:
|
||||
self.logger.info("Received event from Pypo Message Handler: %s" % message)
|
||||
|
||||
m = json.loads(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.schedule_data = m['schedule']
|
||||
self.process_schedule(self.schedule_data)
|
||||
elif command == 'update_stream_setting':
|
||||
self.logger.info("Updating stream setting...")
|
||||
|
|
@ -100,12 +102,12 @@ class PypoFetch(Thread):
|
|||
elif command == 'disconnect_source':
|
||||
self.logger.info("disconnect_on_source show command received...")
|
||||
self.disconnect_source(self.logger, self.telnet_lock, m['sourcename'])
|
||||
|
||||
|
||||
# update timeout value
|
||||
if command == 'update_schedule':
|
||||
self.listener_timeout = 3600
|
||||
self.listener_timeout = POLL_INTERVAL
|
||||
else:
|
||||
self.listener_timeout = self.last_update_schedule_timestamp - time.time() + 3600
|
||||
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)
|
||||
|
|
@ -115,7 +117,7 @@ class PypoFetch(Thread):
|
|||
self.logger.error('Exception: %s', e)
|
||||
self.logger.error("traceback: %s", top)
|
||||
self.logger.error("Exception in handling Message Handler message: %s", e)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def disconnect_source(logger, lock, sourcename):
|
||||
logger.debug('Disconnecting source: %s', sourcename)
|
||||
|
|
@ -124,7 +126,7 @@ class PypoFetch(Thread):
|
|||
command += "master_harbor.kick\n"
|
||||
elif(sourcename == "live_dj"):
|
||||
command += "live_dj_harbor.kick\n"
|
||||
|
||||
|
||||
lock.acquire()
|
||||
try:
|
||||
tn = telnetlib.Telnet(LS_HOST, LS_PORT)
|
||||
|
|
@ -135,7 +137,7 @@ class PypoFetch(Thread):
|
|||
logger.error(str(e))
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def switch_source(logger, lock, sourcename, status):
|
||||
logger.debug('Switching source: %s to "%s" status', sourcename, status)
|
||||
|
|
@ -146,12 +148,12 @@ class PypoFetch(Thread):
|
|||
command += "live_dj_"
|
||||
elif(sourcename == "scheduled_play"):
|
||||
command += "scheduled_play_"
|
||||
|
||||
|
||||
if(status == "on"):
|
||||
command += "start\n"
|
||||
else:
|
||||
command += "stop\n"
|
||||
|
||||
|
||||
lock.acquire()
|
||||
try:
|
||||
tn = telnetlib.Telnet(LS_HOST, LS_PORT)
|
||||
|
|
@ -162,7 +164,7 @@ class PypoFetch(Thread):
|
|||
logger.error(str(e))
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
"""
|
||||
grabs some information that are needed to be set on bootstrap time
|
||||
and configures them
|
||||
|
|
@ -171,16 +173,16 @@ class PypoFetch(Thread):
|
|||
self.logger.debug('Getting information needed on bootstrap from Airtime')
|
||||
info = self.api_client.get_bootstrap_info()
|
||||
if info == None:
|
||||
self.logger.error('Unable to get bootstrap info.. Existing pypo...')
|
||||
sys.exit(0)
|
||||
self.logger.error('Unable to get bootstrap info.. Exiting pypo...')
|
||||
sys.exit(1)
|
||||
else:
|
||||
self.logger.debug('info:%s',info)
|
||||
self.logger.debug('info:%s', info)
|
||||
for k, v in info['switch_status'].iteritems():
|
||||
self.switch_source(self.logger, self.telnet_lock, k, v)
|
||||
self.update_liquidsoap_stream_format(info['stream_label'])
|
||||
self.update_liquidsoap_station_name(info['station_name'])
|
||||
self.update_liquidsoap_transition_fade(info['transition_fade'])
|
||||
|
||||
|
||||
def write_liquidsoap_config(self, setting):
|
||||
fh = open('/etc/airtime/liquidsoap.cfg', 'w')
|
||||
self.logger.info("Rewriting liquidsoap.cfg...")
|
||||
|
|
@ -197,7 +199,7 @@ class PypoFetch(Thread):
|
|||
if temp == "":
|
||||
temp = "0"
|
||||
buffer_str += temp
|
||||
|
||||
|
||||
buffer_str += "\n"
|
||||
fh.write(api_client.encode_to(buffer_str))
|
||||
fh.write("log_file = \"/var/log/airtime/pypo-liquidsoap/<script>.log\"\n");
|
||||
|
|
@ -206,18 +208,18 @@ class PypoFetch(Thread):
|
|||
# we could just restart liquidsoap but it take more time somehow.
|
||||
self.logger.info("Restarting pypo...")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
def regenerateLiquidsoapConf(self, setting):
|
||||
existing = {}
|
||||
# create a temp file
|
||||
|
||||
|
||||
setting = sorted(setting.items())
|
||||
try:
|
||||
fh = open('/etc/airtime/liquidsoap.cfg', 'r')
|
||||
except IOError, e:
|
||||
#file does not exist
|
||||
self.write_liquidsoap_config(setting)
|
||||
|
||||
|
||||
self.logger.info("Reading existing config...")
|
||||
# read existing conf file and build dict
|
||||
while True:
|
||||
|
|
@ -226,9 +228,9 @@ class PypoFetch(Thread):
|
|||
# empty line means EOF
|
||||
if not line:
|
||||
break
|
||||
|
||||
|
||||
line = line.strip()
|
||||
|
||||
|
||||
if line[0] == "#":
|
||||
continue
|
||||
|
||||
|
|
@ -243,7 +245,7 @@ class PypoFetch(Thread):
|
|||
value = ''
|
||||
existing[key] = value
|
||||
fh.close()
|
||||
|
||||
|
||||
# dict flag for any change in cofig
|
||||
change = {}
|
||||
# this flag is to detect disable -> disable change
|
||||
|
|
@ -251,7 +253,7 @@ class PypoFetch(Thread):
|
|||
state_change_restart = {}
|
||||
#restart flag
|
||||
restart = False
|
||||
|
||||
|
||||
self.logger.info("Looking for changes...")
|
||||
# look for changes
|
||||
for k, s in setting:
|
||||
|
|
@ -267,13 +269,13 @@ class PypoFetch(Thread):
|
|||
self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
|
||||
restart = True;
|
||||
else:
|
||||
stream, dump = s[u'keyname'].split('_',1)
|
||||
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'):
|
||||
elif (s[u'value'] != 'disabled'):
|
||||
state_change_restart[stream] = True
|
||||
else:
|
||||
state_change_restart[stream] = False
|
||||
|
|
@ -284,10 +286,10 @@ class PypoFetch(Thread):
|
|||
if not (s[u'value'] == existing[s[u'keyname']]):
|
||||
self.logger.info("Keyname: %s, Curent 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
|
||||
|
|
@ -306,7 +308,7 @@ class PypoFetch(Thread):
|
|||
updates the status of liquidsoap connection to the streaming server
|
||||
This fucntion updates the bootup time variable in liquidsoap script
|
||||
"""
|
||||
|
||||
|
||||
self.telnet_lock.acquire()
|
||||
try:
|
||||
tn = telnetlib.Telnet(LS_HOST, LS_PORT)
|
||||
|
|
@ -314,25 +316,25 @@ class PypoFetch(Thread):
|
|||
# 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"
|
||||
boot_up_time_command = "vars.bootup_time " + str(current_time) + "\n"
|
||||
tn.write(boot_up_time_command)
|
||||
tn.write("streams.connection_status\n")
|
||||
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(':')
|
||||
|
|
@ -340,7 +342,7 @@ class PypoFetch(Thread):
|
|||
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!!!
|
||||
|
|
@ -356,7 +358,7 @@ class PypoFetch(Thread):
|
|||
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!!!
|
||||
|
|
@ -372,14 +374,14 @@ class PypoFetch(Thread):
|
|||
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:
|
||||
self.logger.info(LS_HOST)
|
||||
self.logger.info(LS_PORT)
|
||||
|
||||
|
||||
self.telnet_lock.acquire()
|
||||
try:
|
||||
tn = telnetlib.Telnet(LS_HOST, LS_PORT)
|
||||
|
|
@ -387,7 +389,7 @@ class PypoFetch(Thread):
|
|||
self.logger.info(command)
|
||||
tn.write(command)
|
||||
tn.write('exit\n')
|
||||
tn.read_all()
|
||||
tn.read_all()
|
||||
except Exception, e:
|
||||
self.logger.error(str(e))
|
||||
finally:
|
||||
|
|
@ -403,7 +405,7 @@ class PypoFetch(Thread):
|
|||
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):
|
||||
def process_schedule(self, schedule_data):
|
||||
self.last_update_schedule_timestamp = time.time()
|
||||
self.logger.debug(schedule_data)
|
||||
media = schedule_data["media"]
|
||||
|
|
@ -411,7 +413,7 @@ class PypoFetch(Thread):
|
|||
|
||||
# Download all the media and put playlists in liquidsoap "annotate" format
|
||||
try:
|
||||
|
||||
|
||||
"""
|
||||
Make sure cache_dir exists
|
||||
"""
|
||||
|
|
@ -420,15 +422,16 @@ class PypoFetch(Thread):
|
|||
os.makedirs(download_dir)
|
||||
except Exception, e:
|
||||
pass
|
||||
|
||||
|
||||
for key in media:
|
||||
media_item = media[key]
|
||||
if(media_item['type'] == 'file'):
|
||||
fileExt = os.path.splitext(media_item['uri'])[1]
|
||||
dst = os.path.join(download_dir, media_item['id']+fileExt)
|
||||
dst = os.path.join(download_dir, media_item['id'] + fileExt)
|
||||
media_item['dst'] = dst
|
||||
media_item['started_copying'] = False
|
||||
media_filtered[key] = media_item
|
||||
|
||||
|
||||
self.media_prepare_queue.put(copy.copy(media_filtered))
|
||||
except Exception, e: self.logger.error("%s", e)
|
||||
|
||||
|
|
@ -440,7 +443,7 @@ class PypoFetch(Thread):
|
|||
# cleanup
|
||||
try: self.cache_cleanup(media)
|
||||
except Exception, e: self.logger.error("%s", e)
|
||||
|
||||
|
||||
def cache_cleanup(self, media):
|
||||
"""
|
||||
Get list of all files in the cache dir and remove them if they aren't being used anymore.
|
||||
|
|
@ -449,18 +452,18 @@ class PypoFetch(Thread):
|
|||
"""
|
||||
cached_file_set = set(os.listdir(self.cache_dir))
|
||||
scheduled_file_set = set()
|
||||
|
||||
|
||||
for mkey in media:
|
||||
media_item = media[mkey]
|
||||
fileExt = os.path.splitext(media_item['uri'])[1]
|
||||
scheduled_file_set.add(media_item["id"] + fileExt)
|
||||
|
||||
|
||||
unneeded_files = cached_file_set - scheduled_file_set
|
||||
|
||||
|
||||
self.logger.debug("Files to remove " + str(unneeded_files))
|
||||
for file in unneeded_files:
|
||||
self.logger.debug("Removing %s" % os.path.join(self.cache_dir, file))
|
||||
os.remove(os.path.join(self.cache_dir, file))
|
||||
for f in unneeded_files:
|
||||
self.logger.debug("Removing %s" % os.path.join(self.cache_dir, f))
|
||||
os.remove(os.path.join(self.cache_dir, f))
|
||||
|
||||
def main(self):
|
||||
# Bootstrap: since we are just starting up, we need to grab the
|
||||
|
|
@ -471,10 +474,10 @@ class PypoFetch(Thread):
|
|||
self.process_schedule(self.schedule_data)
|
||||
self.set_bootstrap_variables()
|
||||
|
||||
loops = 1
|
||||
loops = 1
|
||||
while True:
|
||||
self.logger.info("Loop #%s", loops)
|
||||
try:
|
||||
try:
|
||||
"""
|
||||
our simple_queue.get() requires a timeout, in which case we
|
||||
fetch the Airtime schedule manually. It is important to fetch
|
||||
|
|
@ -486,18 +489,20 @@ class PypoFetch(Thread):
|
|||
sent, and we will have very stale (or non-existent!) data about the
|
||||
schedule.
|
||||
|
||||
Currently we are checking every 3600 seconds (1 hour)
|
||||
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")
|
||||
except Exception, e:
|
||||
import traceback
|
||||
top = traceback.format_exc()
|
||||
self.logger.error('Exception: %s', e)
|
||||
self.logger.error("traceback: %s", top)
|
||||
|
||||
|
||||
success, self.schedule_data = self.api_client.get_schedule()
|
||||
if success:
|
||||
self.process_schedule(self.schedule_data)
|
||||
|
|
|
|||
|
|
@ -5,10 +5,10 @@ from Queue import Empty
|
|||
from configobj import ConfigObj
|
||||
|
||||
import logging
|
||||
import logging.config
|
||||
import shutil
|
||||
import os
|
||||
import sys
|
||||
import stat
|
||||
|
||||
from std_err_override import LogWriter
|
||||
|
||||
|
|
@ -33,21 +33,21 @@ except Exception, e:
|
|||
|
||||
|
||||
class PypoFile(Thread):
|
||||
|
||||
|
||||
def __init__(self, schedule_queue):
|
||||
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:
|
||||
|
|
@ -59,24 +59,31 @@ class PypoFile(Thread):
|
|||
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:
|
||||
do_copy = True
|
||||
|
||||
|
||||
if do_copy:
|
||||
self.logger.debug("copying from %s to local cache %s" % (src, dst))
|
||||
try:
|
||||
|
||||
media_item['started_copying'] = True
|
||||
|
||||
"""
|
||||
copy will overwrite dst if it already exists
|
||||
"""
|
||||
shutil.copy(src, dst)
|
||||
except:
|
||||
|
||||
#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
|
||||
|
|
@ -84,17 +91,17 @@ class PypoFile(Thread):
|
|||
"""
|
||||
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
|
||||
|
|
@ -103,11 +110,11 @@ class PypoFile(Thread):
|
|||
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]
|
||||
|
||||
del schedule[highest_priority]
|
||||
|
||||
return media_item
|
||||
|
||||
|
||||
|
||||
|
||||
def main(self):
|
||||
while True:
|
||||
try:
|
||||
|
|
@ -128,7 +135,7 @@ class PypoFile(Thread):
|
|||
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:
|
||||
|
|
@ -139,7 +146,7 @@ class PypoFile(Thread):
|
|||
self.logger.error(str(e))
|
||||
self.logger.error(top)
|
||||
raise
|
||||
|
||||
|
||||
def run(self):
|
||||
"""
|
||||
Entry point of the thread
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import logging
|
||||
import logging.config
|
||||
import sys
|
||||
from configobj import ConfigObj
|
||||
from threading import Thread
|
||||
|
|
@ -39,7 +38,7 @@ class PypoMessageHandler(Thread):
|
|||
self.logger = logging.getLogger('message_h')
|
||||
self.pypo_queue = pq
|
||||
self.recorder_queue = rq
|
||||
|
||||
|
||||
def init_rabbit_mq(self):
|
||||
self.logger.info("Initializing RabbitMQ stuff")
|
||||
try:
|
||||
|
|
@ -51,21 +50,21 @@ class PypoMessageHandler(Thread):
|
|||
except Exception, e:
|
||||
self.logger.error(e)
|
||||
return False
|
||||
|
||||
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
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:
|
||||
try:
|
||||
self.logger.info("Received event from RabbitMQ: %s" % message)
|
||||
|
||||
m = json.loads(message)
|
||||
|
||||
m = json.loads(message)
|
||||
command = m['event_type']
|
||||
self.logger.info("Handling command: " + command)
|
||||
|
||||
|
||||
if command == 'update_schedule':
|
||||
self.logger.info("Updating schdule...")
|
||||
self.pypo_queue.put(message)
|
||||
|
|
@ -121,15 +120,13 @@ class PypoMessageHandler(Thread):
|
|||
while loop and eat all the CPU
|
||||
"""
|
||||
time.sleep(5)
|
||||
|
||||
|
||||
"""
|
||||
There is a problem with the RabbitMq messenger service. Let's
|
||||
log the error and get the schedule via HTTP polling
|
||||
"""
|
||||
import traceback
|
||||
top = traceback.format_exc()
|
||||
self.logger.error('Exception: %s', e)
|
||||
self.logger.error("traceback: %s", top)
|
||||
self.logger.error("traceback: %s", traceback.format_exc())
|
||||
|
||||
loops += 1
|
||||
|
||||
|
|
|
|||
|
|
@ -15,19 +15,9 @@ Main case:
|
|||
|
||||
"""
|
||||
|
||||
# python defaults (debian default)
|
||||
import time
|
||||
import os
|
||||
import traceback
|
||||
from optparse import *
|
||||
from optparse import OptionParser
|
||||
import sys
|
||||
import time
|
||||
import datetime
|
||||
import logging
|
||||
import logging.config
|
||||
import urllib
|
||||
import urllib2
|
||||
import string
|
||||
import json
|
||||
|
||||
# additional modules (should be checked)
|
||||
|
|
@ -38,9 +28,6 @@ from configobj import ConfigObj
|
|||
from api_clients import *
|
||||
from std_err_override import LogWriter
|
||||
|
||||
# Set up command-line options
|
||||
parser = OptionParser()
|
||||
|
||||
# help screeen / info
|
||||
usage = "%prog [options]" + " - notification gateway"
|
||||
parser = OptionParser(usage=usage)
|
||||
|
|
@ -59,8 +46,8 @@ parser.add_option("-y", "--source-status", help="source connection stauts", meta
|
|||
(options, args) = parser.parse_args()
|
||||
|
||||
# configure logging
|
||||
logging.config.fileConfig("logging.cfg")
|
||||
logger = logging.getLogger()
|
||||
logging.config.fileConfig("notify_logging.cfg")
|
||||
logger = logging.getLogger('notify')
|
||||
LogWriter.override_std_err(logger)
|
||||
|
||||
#need to wait for Python 2.7 for this..
|
||||
|
|
@ -69,54 +56,54 @@ LogWriter.override_std_err(logger)
|
|||
# loading config file
|
||||
try:
|
||||
config = ConfigObj('/etc/airtime/pypo.cfg')
|
||||
|
||||
|
||||
except Exception, e:
|
||||
logger.error('Error loading config file: %s', e)
|
||||
sys.exit()
|
||||
|
||||
|
||||
|
||||
|
||||
class Notify:
|
||||
def __init__(self):
|
||||
self.api_client = api_client.api_client_factory(config)
|
||||
|
||||
|
||||
def notify_media_start_playing(self, data, media_id):
|
||||
logger = logging.getLogger("notify")
|
||||
|
||||
|
||||
logger.debug('#################################################')
|
||||
logger.debug('# Calling server to update about what\'s playing #')
|
||||
logger.debug('#################################################')
|
||||
logger.debug('data = '+ str(data))
|
||||
response = self.api_client.notify_media_item_start_playing(data, media_id)
|
||||
logger.debug("Response: "+json.dumps(response))
|
||||
|
||||
logger.debug('data = ' + str(data))
|
||||
response = self.api_client.notify_media_item_start_playing(data, media_id)
|
||||
logger.debug("Response: " + json.dumps(response))
|
||||
|
||||
# @pram time: time that LS started
|
||||
def notify_liquidsoap_status(self, msg, stream_id, time):
|
||||
logger = logging.getLogger("notify")
|
||||
|
||||
|
||||
logger.debug('#################################################')
|
||||
logger.debug('# Calling server to update liquidsoap status #')
|
||||
logger.debug('#################################################')
|
||||
logger.debug('msg = '+ str(msg))
|
||||
response = self.api_client.notify_liquidsoap_status(msg, stream_id, time)
|
||||
logger.debug("Response: "+json.dumps(response))
|
||||
|
||||
logger.debug('msg = ' + str(msg))
|
||||
response = self.api_client.notify_liquidsoap_status(msg, stream_id, time)
|
||||
logger.debug("Response: " + json.dumps(response))
|
||||
|
||||
def notify_source_status(self, source_name, status):
|
||||
logger = logging.getLogger("notify")
|
||||
|
||||
|
||||
logger.debug('#################################################')
|
||||
logger.debug('# Calling server to update source status #')
|
||||
logger.debug('#################################################')
|
||||
logger.debug('msg = '+ str(source_name) + ' : ' + str(status))
|
||||
response = self.api_client.notify_source_status(source_name, status)
|
||||
logger.debug("Response: "+json.dumps(response))
|
||||
|
||||
logger.debug('msg = ' + str(source_name) + ' : ' + str(status))
|
||||
response = self.api_client.notify_source_status(source_name, status)
|
||||
logger.debug("Response: " + json.dumps(response))
|
||||
|
||||
if __name__ == '__main__':
|
||||
print
|
||||
print '#########################################'
|
||||
print '# *** pypo *** #'
|
||||
print '# pypo notification gateway #'
|
||||
print '#########################################'
|
||||
|
||||
|
||||
# initialize
|
||||
logger = logging.getLogger("notify")
|
||||
if options.error and options.stream_id:
|
||||
|
|
@ -141,11 +128,11 @@ if __name__ == '__main__':
|
|||
if not options.data:
|
||||
print "NOTICE: 'data' command-line argument not given."
|
||||
sys.exit()
|
||||
|
||||
|
||||
if not options.media_id:
|
||||
print "NOTICE: 'media_id' command-line argument not given."
|
||||
sys.exit()
|
||||
|
||||
|
||||
try:
|
||||
n = Notify()
|
||||
n.notify_media_start_playing(options.data, options.media_id)
|
||||
|
|
@ -5,11 +5,9 @@ from datetime import timedelta
|
|||
|
||||
import sys
|
||||
import time
|
||||
import logging
|
||||
import logging.config
|
||||
import telnetlib
|
||||
import calendar
|
||||
import json
|
||||
import math
|
||||
from pypofetch import PypoFetch
|
||||
|
||||
|
|
@ -51,52 +49,58 @@ class PypoPush(Thread):
|
|||
|
||||
self.pushed_objects = {}
|
||||
self.logger = logging.getLogger('push')
|
||||
|
||||
|
||||
def main(self):
|
||||
loops = 0
|
||||
heartbeat_period = math.floor(30/PUSH_INTERVAL)
|
||||
|
||||
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)
|
||||
|
||||
chains = self.get_all_chains(media_schedule)
|
||||
|
||||
|
||||
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()
|
||||
|
||||
current_event_chain = self.get_current_chain(chains)
|
||||
tnow = datetime.utcnow()
|
||||
current_event_chain, original_chain = self.get_current_chain(chains, tnow)
|
||||
if len(current_event_chain) > 0 and len(liquidsoap_queue_approx) == 0:
|
||||
#Something is scheduled but Liquidsoap is not playing anything!
|
||||
#Need to schedule it immediately..this might happen if Liquidsoap crashed.
|
||||
chains.remove(current_event_chain)
|
||||
|
||||
try:
|
||||
chains.remove(original_chain)
|
||||
except ValueError, e:
|
||||
self.logger.error(str(e))
|
||||
|
||||
self.modify_cue_point(current_event_chain[0])
|
||||
next_media_item_chain = current_event_chain
|
||||
time_until_next_play = 0
|
||||
#sleep for 0.2 seconds to give pypo-file time to copy. This is a quick
|
||||
#fix that will be improved in 2.1.1
|
||||
#sleep for 0.2 seconds to give pypo-file time to copy.
|
||||
time.sleep(0.2)
|
||||
else:
|
||||
media_chain = filter(lambda item: (item["type"] == "file"), current_event_chain)
|
||||
self.handle_new_media_schedule(media_schedule, liquidsoap_queue_approx, media_chain)
|
||||
|
||||
next_media_item_chain = self.get_next_schedule_chain(chains)
|
||||
|
||||
self.logger.debug("Next schedule chain: %s", next_media_item_chain)
|
||||
|
||||
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:
|
||||
chains.remove(next_media_item_chain)
|
||||
tnow = datetime.utcnow()
|
||||
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 - tnow)
|
||||
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")
|
||||
|
|
@ -104,17 +108,18 @@ class PypoPush(Thread):
|
|||
except Empty, e:
|
||||
#We only get here when a new chain of tracks are ready to be played.
|
||||
self.push_to_liquidsoap(next_media_item_chain)
|
||||
|
||||
next_media_item_chain = self.get_next_schedule_chain(chains)
|
||||
|
||||
next_media_item_chain = self.get_next_schedule_chain(chains, datetime.utcnow())
|
||||
if next_media_item_chain is not None:
|
||||
tnow = datetime.utcnow()
|
||||
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 - tnow)
|
||||
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
|
||||
|
||||
except Exception, e:
|
||||
self.logger.error(str(e))
|
||||
|
||||
if loops % heartbeat_period == 0:
|
||||
self.logger.info("heartbeat")
|
||||
loops = 0
|
||||
|
|
@ -127,7 +132,7 @@ class PypoPush(Thread):
|
|||
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")
|
||||
|
|
@ -138,14 +143,14 @@ class PypoPush(Thread):
|
|||
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])
|
||||
|
|
@ -158,9 +163,9 @@ class PypoPush(Thread):
|
|||
self.clear_liquidsoap_queue()
|
||||
liquidsoap_queue_approx = []
|
||||
break
|
||||
|
||||
|
||||
return liquidsoap_queue_approx
|
||||
|
||||
|
||||
def handle_new_media_schedule(self, media_schedule, liquidsoap_queue_approx, media_chain):
|
||||
"""
|
||||
This function's purpose is to gracefully handle situations where
|
||||
|
|
@ -169,27 +174,27 @@ class PypoPush(Thread):
|
|||
call other functions that will connect to Liquidsoap and alter its
|
||||
queue.
|
||||
"""
|
||||
|
||||
problem_at_iteration, problem_start_time = self.find_removed_items(media_schedule, liquidsoap_queue_approx)
|
||||
|
||||
|
||||
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(media_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 = media_chain[problem_at_iteration:]
|
||||
if len(chain_to_push) > 0:
|
||||
self.modify_cue_point(chain_to_push[0])
|
||||
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
|
||||
|
|
@ -201,7 +206,6 @@ class PypoPush(Thread):
|
|||
#see if they are the same as the newly received schedule
|
||||
iteration = 0
|
||||
problem_at_iteration = None
|
||||
problem_start_time = None
|
||||
for queue_item in liquidsoap_queue_approx:
|
||||
if queue_item['start'] in media_schedule.keys():
|
||||
media_item = media_schedule[queue_item['start']]
|
||||
|
|
@ -211,33 +215,30 @@ class PypoPush(Thread):
|
|||
pass
|
||||
else:
|
||||
problem_at_iteration = iteration
|
||||
problem_start_time = queue_item['start']
|
||||
break
|
||||
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
|
||||
problem_start_time = queue_item['start']
|
||||
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
|
||||
problem_start_time = queue_item['start']
|
||||
break
|
||||
iteration+=1
|
||||
return (problem_at_iteration, problem_start_time)
|
||||
|
||||
|
||||
|
||||
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['type'] == "event":
|
||||
|
|
@ -251,54 +252,68 @@ class PypoPush(Thread):
|
|||
#Start a new one instead
|
||||
chains.append(current_chain)
|
||||
current_chain = [media_item]
|
||||
|
||||
|
||||
if len(current_chain) > 0:
|
||||
chains.append(current_chain)
|
||||
|
||||
|
||||
return chains
|
||||
|
||||
|
||||
def modify_cue_point(self, link):
|
||||
tnow = datetime.utcnow()
|
||||
|
||||
|
||||
link_start = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S")
|
||||
|
||||
|
||||
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
|
||||
|
||||
"""
|
||||
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
|
||||
|
||||
def get_current_chain(self, chains):
|
||||
tnow = datetime.utcnow()
|
||||
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 = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S")
|
||||
link_end = datetime.strptime(link['end'], "%Y-%m-%d-%H-%M-%S")
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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):
|
||||
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.
|
||||
tnow = datetime.utcnow()
|
||||
closest_start = None
|
||||
closest_chain = None
|
||||
for chain in chains:
|
||||
|
|
@ -307,19 +322,32 @@ class PypoPush(Thread):
|
|||
if (closest_start == None or chain_start < closest_start) and chain_start > tnow:
|
||||
closest_start = chain_start
|
||||
closest_chain = chain
|
||||
|
||||
|
||||
return closest_chain
|
||||
|
||||
|
||||
|
||||
|
||||
def date_interval_to_seconds(self, interval):
|
||||
return (interval.microseconds + (interval.seconds + interval.days * 24 * 3600) * 10**6) / float(10**6)
|
||||
|
||||
return (interval.microseconds + (interval.seconds + interval.days * 24 * 3600) * 10 ** 6) / float(10 ** 6)
|
||||
|
||||
def push_to_liquidsoap(self, event_chain):
|
||||
|
||||
|
||||
try:
|
||||
for media_item in event_chain:
|
||||
if media_item['type'] == "file":
|
||||
self.telnet_to_liquidsoap(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['started_copying'] and iter_num < 50:
|
||||
time.sleep(0.1)
|
||||
iter_num += 1
|
||||
|
||||
if media_item['started_copying']:
|
||||
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")
|
||||
|
|
@ -327,27 +355,27 @@ class PypoPush(Thread):
|
|||
PypoFetch.switch_source(self.logger, self.telnet_lock, "live_dj", "off")
|
||||
except Exception, e:
|
||||
self.logger.error('Pypo Push Exception: %s', e)
|
||||
|
||||
|
||||
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(msg)
|
||||
tn.write("exit\n")
|
||||
tn.read_all()
|
||||
except Exception, e:
|
||||
self.logger.error(str(e))
|
||||
finally:
|
||||
self.telnet_lock.release()
|
||||
|
||||
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)
|
||||
|
|
@ -355,13 +383,13 @@ class PypoPush(Thread):
|
|||
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
|
||||
|
|
@ -370,33 +398,33 @@ class PypoPush(Thread):
|
|||
msg = "source.skip\n"
|
||||
self.logger.debug(msg)
|
||||
tn.write(msg)
|
||||
|
||||
|
||||
msg = "queue.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)
|
||||
|
||||
|
|
@ -417,43 +445,43 @@ class PypoPush(Thread):
|
|||
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):
|
||||
|
||||
def create_liquidsoap_annotation(self, media):
|
||||
# we need lia_start_next value in the annotate. That is the value that controlls 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":%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['dst'])
|
||||
|
||||
% (media['id'], float(media['fade_in']) / 1000, float(media['fade_out']) / 1000, float(media['cue_in']), float(media['cue_out']), media['row_id'], media['dst'])
|
||||
|
||||
def run(self):
|
||||
try: self.main()
|
||||
except Exception, e:
|
||||
import traceback
|
||||
top = traceback.format_exc()
|
||||
self.logger.error('Pypo Push Exception: %s', top)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import logging
|
||||
import logging.config
|
||||
import json
|
||||
import time
|
||||
import datetime
|
||||
|
|
@ -55,7 +54,7 @@ class ShowRecorder(Thread):
|
|||
self.p = None
|
||||
|
||||
def record_show(self):
|
||||
length = str(self.filelength)+".0"
|
||||
length = str(self.filelength) + ".0"
|
||||
filename = self.start_time
|
||||
filename = filename.replace(" ", "-")
|
||||
|
||||
|
|
@ -128,7 +127,7 @@ class ShowRecorder(Thread):
|
|||
time = md[1].replace(":", "-")
|
||||
self.logger.info("time: %s" % time)
|
||||
|
||||
name = time+"-"+self.show_name
|
||||
name = time + "-" + self.show_name
|
||||
artist = "Airtime Show Recorder"
|
||||
|
||||
#set some metadata for our file daemon
|
||||
|
|
@ -181,7 +180,7 @@ class Recorder(Thread):
|
|||
def handle_message(self):
|
||||
if not self.queue.empty():
|
||||
message = self.queue.get()
|
||||
msg = json.loads(message)
|
||||
msg = json.loads(message)
|
||||
command = msg["event_type"]
|
||||
self.logger.info("Received msg from Pypo Message Handler: %s", msg)
|
||||
if command == 'cancel_recording':
|
||||
|
|
@ -190,10 +189,10 @@ class Recorder(Thread):
|
|||
else:
|
||||
self.process_recorder_schedule(msg)
|
||||
self.loops = 0
|
||||
|
||||
|
||||
if self.shows_to_record:
|
||||
self.start_record()
|
||||
|
||||
|
||||
def process_recorder_schedule(self, m):
|
||||
self.logger.info("Parsing recording show schedules...")
|
||||
temp_shows_to_record = {}
|
||||
|
|
@ -217,7 +216,7 @@ class Recorder(Thread):
|
|||
delta = next_show - tnow
|
||||
s = '%s.%s' % (delta.seconds, delta.microseconds)
|
||||
out = float(s)
|
||||
|
||||
|
||||
if out < 5:
|
||||
self.logger.debug("Shows %s", self.shows_to_record)
|
||||
self.logger.debug("Next show %s", next_show)
|
||||
|
|
@ -231,26 +230,26 @@ class Recorder(Thread):
|
|||
if delta < 5:
|
||||
self.logger.debug("sleeping %s seconds until show", delta)
|
||||
time.sleep(delta)
|
||||
|
||||
|
||||
sorted_show_keys = sorted(self.shows_to_record.keys())
|
||||
start_time = sorted_show_keys[0]
|
||||
show_length = self.shows_to_record[start_time][0]
|
||||
show_instance = self.shows_to_record[start_time][1]
|
||||
show_name = self.shows_to_record[start_time][2]
|
||||
server_timezone = self.shows_to_record[start_time][3]
|
||||
|
||||
|
||||
T = pytz.timezone(server_timezone)
|
||||
start_time_on_UTC = getDateTimeObj(start_time)
|
||||
start_time_on_server = start_time_on_UTC.replace(tzinfo=pytz.utc).astimezone(T)
|
||||
start_time_formatted = '%(year)d-%(month)02d-%(day)02d %(hour)02d:%(min)02d:%(sec)02d' % \
|
||||
{'year': start_time_on_server.year, 'month': start_time_on_server.month, 'day': start_time_on_server.day,\
|
||||
{'year': start_time_on_server.year, 'month': start_time_on_server.month, 'day': start_time_on_server.day, \
|
||||
'hour': start_time_on_server.hour, 'min': start_time_on_server.minute, 'sec': start_time_on_server.second}
|
||||
self.sr = ShowRecorder(show_instance, show_name, show_length.seconds, start_time_formatted)
|
||||
self.sr.start()
|
||||
#remove show from shows to record.
|
||||
del self.shows_to_record[start_time]
|
||||
#self.time_till_next_show = self.get_time_till_next_show()
|
||||
except Exception,e :
|
||||
except Exception, e :
|
||||
import traceback
|
||||
top = traceback.format_exc()
|
||||
self.logger.error('Exception: %s', e)
|
||||
|
|
@ -273,12 +272,12 @@ class Recorder(Thread):
|
|||
self.logger.info("Bootstrap recorder schedule received: %s", temp)
|
||||
except Exception, e:
|
||||
self.logger.error(e)
|
||||
|
||||
|
||||
self.logger.info("Bootstrap complete: got initial copy of the schedule")
|
||||
|
||||
|
||||
self.loops = 0
|
||||
heartbeat_period = math.floor(30/PUSH_INTERVAL)
|
||||
|
||||
heartbeat_period = math.floor(30 / PUSH_INTERVAL)
|
||||
|
||||
while True:
|
||||
if self.loops % heartbeat_period == 0:
|
||||
self.logger.info("heartbeat")
|
||||
|
|
@ -299,7 +298,7 @@ class Recorder(Thread):
|
|||
self.logger.error('Pypo Recorder Exception: %s', e)
|
||||
time.sleep(PUSH_INTERVAL)
|
||||
self.loops += 1
|
||||
except Exception,e :
|
||||
except Exception, e :
|
||||
import traceback
|
||||
top = traceback.format_exc()
|
||||
self.logger.error('Exception: %s', e)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue