Congegni/sintonia
7
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch '2.1.x' of dev.sourcefabric.org:airtime into 2.1.x

Browse source
This commit is contained in:
James 2012-06-28 12:25:38 -04:00
commit ffe3bdab0c
18 changed files with 339 additions and 329 deletions
Download patch file Download diff file Expand all files Collapse all files

3
airtime_mvc/application/Bootstrap.php
View file

@ -25,8 +25,7 @@ date_default_timezone_set(Application_Model_Preference::GetTimezone());
global $CC_CONFIG; global $CC_CONFIG;
$airtime_version = Application_Model_Preference::GetAirtimeVersion(); $airtime_version = Application_Model_Preference::GetAirtimeVersion();
$uniqueid = Application_Model_Preference::GetUniqueId(); $uniqueid = Application_Model_Preference::GetUniqueId();
$CC_CONFIG['airtime_version'] = md5($airtime_version + $uniqueid); $CC_CONFIG['airtime_version'] = md5($airtime_version.$uniqueid);
require_once __DIR__."/configs/navigation.php"; require_once __DIR__."/configs/navigation.php";
Zend_Validate::setDefaultNamespaces("Zend"); Zend_Validate::setDefaultNamespaces("Zend");

6
airtime_mvc/application/controllers/ApiController.php
View file

@ -129,6 +129,12 @@ class ApiController extends Zend_Controller_Action
if(is_file($filepath)){ if(is_file($filepath)){
$full_path = $media->getPropelOrm()->getDbFilepath(); $full_path = $media->getPropelOrm()->getDbFilepath();
$file_base_name = strrchr($full_path, '/'); $file_base_name = strrchr($full_path, '/');
/* If $full_path does not contain a '/', strrchr will return false,
* in which case we can use $full_path as the base name.
*/
if (!$file_base_name) {
$file_base_name = $full_path;
}
$file_base_name = substr($file_base_name, 1); $file_base_name = substr($file_base_name, 1);
// possibly use fileinfo module here in the future. // possibly use fileinfo module here in the future.
// http://www.php.net/manual/en/book.fileinfo.php // http://www.php.net/manual/en/book.fileinfo.php

7
airtime_mvc/application/models/Schedule.php
View file

@ -548,10 +548,11 @@ class Application_Model_Schedule {
$data["media"][$kick_start]['end'] = $kick_start; $data["media"][$kick_start]['end'] = $kick_start;
$data["media"][$kick_start]['event_type'] = "kick_out"; $data["media"][$kick_start]['event_type'] = "kick_out";
$data["media"][$kick_start]['type'] = "event"; $data["media"][$kick_start]['type'] = "event";
if($kick_time !== $switch_off_time){ if($kick_time !== $switch_off_time){
$data["media"][$switch_start]['start'] = Application_Model_Schedule::AirtimeTimeToPypoTime($switch_off_time); $switch_start = Application_Model_Schedule::AirtimeTimeToPypoTime($switch_off_time);
$data["media"][$switch_start]['end'] = Application_Model_Schedule::AirtimeTimeToPypoTime($switch_off_time); $data["media"][$switch_start]['start'] = $switch_start;
$data["media"][$switch_start]['end'] = $switch_start;
$data["media"][$switch_start]['event_type'] = "switch_off"; $data["media"][$switch_start]['event_type'] = "switch_off";
$data["media"][$switch_start]['type'] = "event"; $data["media"][$switch_start]['type'] = "event";
} }

26
airtime_mvc/public/js/airtime/library/library.js
View file

@ -202,13 +202,19 @@ var AIRTIME = (function(AIRTIME) {
*/ */
mod.selectCurrentPage = function() { mod.selectCurrentPage = function() {
$.fn.reverse = [].reverse; $.fn.reverse = [].reverse;
var $trs = $libTable.find("tbody input:checkbox").parents("tr").reverse(); var $inputs = $libTable.find("tbody input:checkbox"),
$trs = $inputs.parents("tr").reverse();
$inputs.attr("checked", true);
$trs.addClass(LIB_SELECTED_CLASS);
$trs.each(function(i, el){ $trs.each(function(i, el){
$el = $(this); $el = $(this);
mod.addToChosen($el);
mod.selectItem($el);
}); });
mod.checkToolBarIcons();
}; };
/* /*
@ -216,14 +222,20 @@ var AIRTIME = (function(AIRTIME) {
* (behaviour taken from gmail) * (behaviour taken from gmail)
*/ */
mod.deselectCurrentPage = function() { mod.deselectCurrentPage = function() {
var $inputs = $libTable.find("tbody input:checkbox"),
$trs = $inputs.parents("tr"),
id;
var $trs = $libTable.find("tbody input:checkbox").filter(":checked").parents("tr"); $inputs.attr("checked", false);
$trs.removeClass(LIB_SELECTED_CLASS);
$trs.each(function(i, el){ $trs.each(function(i, el){
$el = $(this); $el = $(this);
id = $el.attr("id");
mod.deselectItem($el); delete chosenItems[id];
}); });
mod.checkToolBarIcons();
}; };
mod.selectNone = function() { mod.selectNone = function() {

34
airtime_mvc/public/js/airtime/schedule/add-show.js
View file

@ -91,6 +91,34 @@ function onEndTimeSelect(){
$("#add_show_end_time").trigger('input'); $("#add_show_end_time").trigger('input');
} }
function padZeroes(number, length)
{
var str = '' + number;
while (str.length < length) {str = '0' + str;}
return str;
}
function hashCode(str) { // java String#hashCode
var hash = 0;
for (var i = 0; i < str.length; i++) {
hash = str.charCodeAt(i) + ((hash << 5) - hash);
}
return hash;
}
function intToRGB(i){
return (padZeroes(((i>>16)&0xFF).toString(16), 2) +
padZeroes(((i>>8)&0xFF).toString(16), 2)+
padZeroes((i&0xFF).toString(16), 2)
);
}
function stringToColor(s)
{
return intToRGB(hashCode(s));
}
function setAddShowEvents() { function setAddShowEvents() {
var form = $("#add-show-form"); var form = $("#add-show-form");
@ -548,6 +576,12 @@ function setAddShowEvents() {
loadingIcon.hide(); loadingIcon.hide();
}); });
} }
var bgColorEle = $("#add_show_background_color");
$('#add_show_name').bind('input', 'change', function(){
var colorCode = stringToColor($(this).val());
bgColorEle.val(colorCode);
});
} }
function showErrorSections() { function showErrorSections() {

13
airtime_mvc/public/js/airtime/showbuilder/builder.js
View file

@ -210,7 +210,8 @@ var AIRTIME = (function(AIRTIME){
mod.fnItemCallback = function(json) { mod.fnItemCallback = function(json) {
checkError(json); checkError(json);
cursorIds = [];
cursors = $(".cursor-selected-row"); cursors = $(".cursor-selected-row");
for (i = 0; i < cursors.length; i++) { for (i = 0; i < cursors.length; i++) {
cursorIds.push(($(cursors.get(i)).attr("id"))); cursorIds.push(($(cursors.get(i)).attr("id")));
@ -422,8 +423,6 @@ var AIRTIME = (function(AIRTIME){
$nRow.addClass(sClass); $nRow.addClass(sClass);
}; };
$nRow.attr("id", aData.id);
if (aData.header === true) { if (aData.header === true) {
//remove the column classes from all tds. //remove the column classes from all tds.
@ -583,11 +582,12 @@ var AIRTIME = (function(AIRTIME){
$nRow.addClass("sb-future"); $nRow.addClass("sb-future");
} }
if (aData.allowed !== true) { if (aData.allowed !== true || aData.header === true) {
$nRow.addClass("sb-not-allowed"); $nRow.addClass("sb-not-allowed");
} }
else { else {
$nRow.addClass("sb-allowed"); $nRow.addClass("sb-allowed");
$nRow.attr("id", aData.id);
} }
//status used to colour tracks. //status used to colour tracks.
@ -601,7 +601,7 @@ var AIRTIME = (function(AIRTIME){
if (aData.currentShow === true) { if (aData.currentShow === true) {
$nRow.addClass("sb-current-show"); $nRow.addClass("sb-current-show");
} }
//call the context menu so we can prevent the event from propagating. //call the context menu so we can prevent the event from propagating.
$nRow.find('td:gt(1)').click(function(e){ $nRow.find('td:gt(1)').click(function(e){
@ -673,7 +673,6 @@ var AIRTIME = (function(AIRTIME){
$tr = $table.find("tr[id="+cursorIds[i]+"]"); $tr = $table.find("tr[id="+cursorIds[i]+"]");
mod.selectCursor($tr); mod.selectCursor($tr);
} }
cursorIds = [];
//if there is only 1 cursor on the page highlight it by default. //if there is only 1 cursor on the page highlight it by default.
if ($cursorRows.length === 1) { if ($cursorRows.length === 1) {
@ -699,7 +698,7 @@ var AIRTIME = (function(AIRTIME){
if (temp.length > 0) { if (temp.length > 0) {
aData = temp.data("aData"); aData = temp.data("aData");
// max time interval // max time interval
// setTimeout allow only up to 2^21 millisecs timeout value // setTimeout allows only up to (2^31)-1 millisecs timeout value
maxRefreshInterval = Math.pow(2, 31) - 1; maxRefreshInterval = Math.pow(2, 31) - 1;
refreshInterval = aData.refresh * 1000; refreshInterval = aData.refresh * 1000;
if(refreshInterval > maxRefreshInterval){ if(refreshInterval > maxRefreshInterval){

2
install_minimal/DoctrineMigrations/Version20110711161043.php
View file

@ -15,7 +15,7 @@ class Version20110711161043 extends AbstractMigration
{ {
public function up(Schema $schema) public function up(Schema $schema)
{ {
$ini = parse_ini_file(__DIR__."../include/airtime-install.ini"); $ini = parse_ini_file(__DIR__."/../include/airtime-install.ini");
$stor_dir = $ini["storage_dir"]; $stor_dir = $ini["storage_dir"];
/* 1) update cc_files table to include to "directory" column */ /* 1) update cc_files table to include to "directory" column */

10
install_minimal/include/airtime-install.php
View file

@ -61,6 +61,16 @@ AirtimeInstall::InstallStorageDirectory();
$db_install = getenv("nodb")!="t"; $db_install = getenv("nodb")!="t";
if ($db_install) { if ($db_install) {
echo "* Checking database for correct encoding".PHP_EOL;
exec('su -c \'psql -t -c "SHOW SERVER_ENCODING"\' postgres | grep -i "UTF.*8"', $out, $return_code);
if ($return_code != 0){
echo " * Unfortunately your postgresql database has not been created using a UTF-8 encoding.".PHP_EOL;
echo " * As of Airtime 2.1, installs will fail unless the encoding has been set to UTF-8. Please verify this is the case".PHP_EOL;
echo " * and try the install again".PHP_EOL;
exit(1);
}
if($newInstall) { if($newInstall) {
//call external script. "y" argument means force creation of database tables. //call external script. "y" argument means force creation of database tables.
passthru('php '.__DIR__.'/airtime-db-install.php y'); passthru('php '.__DIR__.'/airtime-db-install.php y');

14
python_apps/pypo/install/pypo-initialize.py
View file

@ -109,20 +109,6 @@ try:
print e print e
sys.exit(1) 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 #initialize init.d scripts
subprocess.call("update-rc.d airtime-playout defaults >/dev/null 2>&1", shell=True) subprocess.call("update-rc.d airtime-playout defaults >/dev/null 2>&1", shell=True)

16
python_apps/pypo/liquidsoap_scripts/ls_script.liq
View file

@ -130,21 +130,21 @@ end
def append_dj_inputs(master_harbor_input_port, master_harbor_input_mount_point, dj_harbor_input_port, dj_harbor_input_mount_point, s) = 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 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, 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)) 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, 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)) max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect)))
ignore(output.dummy(master_dj, fallible=true)) ignore(output.dummy(master_dj, fallible=true))
ignore(output.dummy(dj_live, 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)]) 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 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, 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)) max=40., on_connect=master_dj_connect, on_disconnect=master_dj_disconnect)))
ignore(output.dummy(master_dj, fallible=true)) ignore(output.dummy(master_dj, fallible=true))
switch(id="master_dj_switch", track_sensitive=false, transitions=[transition, transition], [({!master_dj_enabled},master_dj), ({true}, s)]) 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 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, 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)) max=40., on_connect=live_dj_connect, on_disconnect=live_dj_disconnect)))
ignore(output.dummy(dj_live, fallible=true)) 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)]) switch(id="live_dj_switch", track_sensitive=false, transitions=[transition, transition], [({!live_dj_enabled},dj_live), ({true}, s)])
else else

2
python_apps/pypo/liquidsoap_scripts/notify.sh
View file

@ -10,4 +10,4 @@ SCRIPT=`readlink -f $0`
# Absolute path this script is in # Absolute path this script is in
SCRIPTPATH=`dirname $SCRIPT` SCRIPTPATH=`dirname $SCRIPT`
cd ${SCRIPTPATH}/../ && python pypo-notify.py "$@" cd ${SCRIPTPATH}/../ && python pyponotify.py "$@"

89
python_apps/pypo/pypocli.py
View file

@ -2,13 +2,13 @@
Python part of radio playout (pypo) Python part of radio playout (pypo)
""" """
from optparse import OptionParser
from datetime import datetime
import time import time
from optparse import *
import sys import sys
import signal import signal
import logging import logging
import logging.config
import logging.handlers
import locale import locale
import os import os
from Queue import Queue from Queue import Queue
@ -53,11 +53,11 @@ parser.add_option("-c", "--check", help="Check the cached schedule and exit", de
def configure_locale(): def configure_locale():
logger.debug("Before %s", locale.nl_langinfo(locale.CODESET)) logger.debug("Before %s", locale.nl_langinfo(locale.CODESET))
current_locale = locale.getlocale() current_locale = locale.getlocale()
if current_locale[1] is None: if current_locale[1] is None:
logger.debug("No locale currently set. Attempting to get default locale.") logger.debug("No locale currently set. Attempting to get default locale.")
default_locale = locale.getdefaultlocale() default_locale = locale.getdefaultlocale()
if default_locale[1] is None: if default_locale[1] is None:
logger.debug("No default locale exists. Let's try loading from /etc/default/locale") logger.debug("No default locale exists. Let's try loading from /etc/default/locale")
if os.path.exists("/etc/default/locale"): if os.path.exists("/etc/default/locale"):
@ -69,17 +69,17 @@ def configure_locale():
sys.exit(1) sys.exit(1)
else: else:
new_locale = default_locale new_locale = default_locale
logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale)) logger.info("New locale set to: %s", locale.setlocale(locale.LC_ALL, new_locale))
reload(sys) reload(sys)
sys.setdefaultencoding("UTF-8") sys.setdefaultencoding("UTF-8")
current_locale_encoding = locale.getlocale()[1].lower() current_locale_encoding = locale.getlocale()[1].lower()
logger.debug("sys default encoding %s", sys.getdefaultencoding()) logger.debug("sys default encoding %s", sys.getdefaultencoding())
logger.debug("After %s", locale.nl_langinfo(locale.CODESET)) logger.debug("After %s", locale.nl_langinfo(locale.CODESET))
if current_locale_encoding not in ['utf-8', 'utf8']: if current_locale_encoding not in ['utf-8', 'utf8']:
logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding) logger.error("Need a UTF-8 locale. Currently '%s'. Exiting..." % current_locale_encoding)
sys.exit(1) sys.exit(1)
@ -92,7 +92,7 @@ try:
except Exception, e: except Exception, e:
print "Couldn't configure logging" print "Couldn't configure logging"
sys.exit() sys.exit()
configure_locale() configure_locale()
# loading config file # loading config file
@ -105,42 +105,14 @@ except Exception, e:
class Global: class Global:
def __init__(self): def __init__(self):
self.api_client = api_client.api_client_factory(config) self.api_client = api_client.api_client_factory(config)
def selfcheck(self): def selfcheck(self):
self.api_client = api_client.api_client_factory(config) self.api_client = api_client.api_client_factory(config)
return self.api_client.is_server_compatible() return self.api_client.is_server_compatible()
def test_api(self): def test_api(self):
self.api_client.test() 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): def keyboardInterruptHandler(signum, frame):
logger = logging.getLogger() logger = logging.getLogger()
logger.info('\nKeyboard Interrupt\n') logger.info('\nKeyboard Interrupt\n')
@ -154,13 +126,19 @@ if __name__ == '__main__':
logger.info('# Liquidsoap Scheduled Playout System #') logger.info('# Liquidsoap Scheduled Playout System #')
logger.info('###########################################') 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" % time.tzname)
logger.info("UTC time: %s" % datetime.utcnow())
signal.signal(signal.SIGINT, keyboardInterruptHandler) signal.signal(signal.SIGINT, keyboardInterruptHandler)
# initialize # initialize
g = Global() g = Global()
while not g.selfcheck(): time.sleep(5) while not g.selfcheck(): time.sleep(5)
logger = logging.getLogger() logger = logging.getLogger()
if options.test: if options.test:
@ -173,9 +151,9 @@ if __name__ == '__main__':
pypoFetch_q = Queue() pypoFetch_q = Queue()
recorder_q = Queue() recorder_q = Queue()
pypoPush_q = Queue() pypoPush_q = Queue()
telnet_lock = Lock() telnet_lock = Lock()
""" """
This queue is shared between pypo-fetch and pypo-file, where pypo-file 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 is the receiver. Pypo-fetch will send every schedule it gets to pypo-file
@ -183,19 +161,19 @@ if __name__ == '__main__':
priority, and will retrieve it. priority, and will retrieve it.
""" """
media_q = Queue() media_q = Queue()
pmh = PypoMessageHandler(pypoFetch_q, recorder_q) pmh = PypoMessageHandler(pypoFetch_q, recorder_q)
pmh.daemon = True pmh.daemon = True
pmh.start() pmh.start()
pfile = PypoFile(media_q) pfile = PypoFile(media_q)
pfile.daemon = True pfile.daemon = True
pfile.start() pfile.start()
pf = PypoFetch(pypoFetch_q, pypoPush_q, media_q, telnet_lock) pf = PypoFetch(pypoFetch_q, pypoPush_q, media_q, telnet_lock)
pf.daemon = True pf.daemon = True
pf.start() pf.start()
pp = PypoPush(pypoPush_q, telnet_lock) pp = PypoPush(pypoPush_q, telnet_lock)
pp.daemon = True pp.daemon = True
pp.start() pp.start()
@ -204,23 +182,12 @@ if __name__ == '__main__':
recorder.daemon = True recorder.daemon = True
recorder.start() 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 # if pypofetch is exiting
#pmh.join() #pmh.join()
#recorder.join() #recorder.join()
#pp.join() #pp.join()
pf.join() pf.join()
logger.info("pypo fetch exit") logger.info("pypo fetch exit")
sys.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
"""

141
python_apps/pypo/pypofetch.py
View file

@ -3,14 +3,14 @@
import os import os
import sys import sys
import time import time
import logging
import logging.config import logging.config
import shutil
import json import json
import telnetlib import telnetlib
import copy import copy
from threading import Thread from threading import Thread
from Queue import Empty
from api_clients import api_client from api_clients import api_client
from std_err_override import LogWriter from std_err_override import LogWriter
@ -29,7 +29,9 @@ try:
config = ConfigObj('/etc/airtime/pypo.cfg') config = ConfigObj('/etc/airtime/pypo.cfg')
LS_HOST = config['ls_host'] LS_HOST = config['ls_host']
LS_PORT = config['ls_port'] LS_PORT = config['ls_port']
POLL_INTERVAL = int(config['poll_interval']) #POLL_INTERVAL = int(config['poll_interval'])
POLL_INTERVAL = 1800
except Exception, e: except Exception, e:
logger.error('Error loading config file: %s', e) logger.error('Error loading config file: %s', e)
@ -43,12 +45,12 @@ class PypoFetch(Thread):
self.push_queue = pypoPush_q self.push_queue = pypoPush_q
self.media_prepare_queue = media_q self.media_prepare_queue = media_q
self.last_update_schedule_timestamp = time.time() self.last_update_schedule_timestamp = time.time()
self.listener_timeout = 3600 self.listener_timeout = POLL_INTERVAL
self.telnet_lock = telnet_lock self.telnet_lock = telnet_lock
self.logger = logging.getLogger(); self.logger = logging.getLogger();
self.cache_dir = os.path.join(config["cache_dir"], "scheduler") self.cache_dir = os.path.join(config["cache_dir"], "scheduler")
self.logger.debug("Cache dir %s", self.cache_dir) self.logger.debug("Cache dir %s", self.cache_dir)
@ -63,24 +65,24 @@ class PypoFetch(Thread):
os.makedirs(dir) os.makedirs(dir)
except Exception, e: except Exception, e:
pass pass
self.schedule_data = [] self.schedule_data = []
self.logger.info("PypoFetch: init complete") self.logger.info("PypoFetch: init complete")
""" """
Handle a message from RabbitMQ, put it into our yucky global var. Handle a message from RabbitMQ, put it into our yucky global var.
Hopefully there is a better way to do this. Hopefully there is a better way to do this.
""" """
def handle_message(self, message): def handle_message(self, message):
try: try:
self.logger.info("Received event from Pypo Message Handler: %s" % message) self.logger.info("Received event from Pypo Message Handler: %s" % message)
m = json.loads(message) m = json.loads(message)
command = m['event_type'] command = m['event_type']
self.logger.info("Handling command: " + command) self.logger.info("Handling command: " + command)
if command == 'update_schedule': if command == 'update_schedule':
self.schedule_data = m['schedule'] self.schedule_data = m['schedule']
self.process_schedule(self.schedule_data) self.process_schedule(self.schedule_data)
elif command == 'update_stream_setting': elif command == 'update_stream_setting':
self.logger.info("Updating stream setting...") self.logger.info("Updating stream setting...")
@ -100,12 +102,12 @@ class PypoFetch(Thread):
elif command == 'disconnect_source': elif command == 'disconnect_source':
self.logger.info("disconnect_on_source show command received...") self.logger.info("disconnect_on_source show command received...")
self.disconnect_source(self.logger, self.telnet_lock, m['sourcename']) self.disconnect_source(self.logger, self.telnet_lock, m['sourcename'])
# update timeout value # update timeout value
if command == 'update_schedule': if command == 'update_schedule':
self.listener_timeout = 3600 self.listener_timeout = POLL_INTERVAL
else: 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: if self.listener_timeout < 0:
self.listener_timeout = 0 self.listener_timeout = 0
self.logger.info("New timeout: %s" % self.listener_timeout) 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('Exception: %s', e)
self.logger.error("traceback: %s", top) self.logger.error("traceback: %s", top)
self.logger.error("Exception in handling Message Handler message: %s", e) self.logger.error("Exception in handling Message Handler message: %s", e)
@staticmethod @staticmethod
def disconnect_source(logger, lock, sourcename): def disconnect_source(logger, lock, sourcename):
logger.debug('Disconnecting source: %s', sourcename) logger.debug('Disconnecting source: %s', sourcename)
@ -124,7 +126,7 @@ class PypoFetch(Thread):
command += "master_harbor.kick\n" command += "master_harbor.kick\n"
elif(sourcename == "live_dj"): elif(sourcename == "live_dj"):
command += "live_dj_harbor.kick\n" command += "live_dj_harbor.kick\n"
lock.acquire() lock.acquire()
try: try:
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
@ -135,7 +137,7 @@ class PypoFetch(Thread):
logger.error(str(e)) logger.error(str(e))
finally: finally:
lock.release() lock.release()
@staticmethod @staticmethod
def switch_source(logger, lock, sourcename, status): def switch_source(logger, lock, sourcename, status):
logger.debug('Switching source: %s to "%s" status', sourcename, status) logger.debug('Switching source: %s to "%s" status', sourcename, status)
@ -146,12 +148,12 @@ class PypoFetch(Thread):
command += "live_dj_" command += "live_dj_"
elif(sourcename == "scheduled_play"): elif(sourcename == "scheduled_play"):
command += "scheduled_play_" command += "scheduled_play_"
if(status == "on"): if(status == "on"):
command += "start\n" command += "start\n"
else: else:
command += "stop\n" command += "stop\n"
lock.acquire() lock.acquire()
try: try:
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
@ -162,7 +164,7 @@ class PypoFetch(Thread):
logger.error(str(e)) logger.error(str(e))
finally: finally:
lock.release() lock.release()
""" """
grabs some information that are needed to be set on bootstrap time grabs some information that are needed to be set on bootstrap time
and configures them and configures them
@ -171,16 +173,16 @@ class PypoFetch(Thread):
self.logger.debug('Getting information needed on bootstrap from Airtime') self.logger.debug('Getting information needed on bootstrap from Airtime')
info = self.api_client.get_bootstrap_info() info = self.api_client.get_bootstrap_info()
if info == None: if info == None:
self.logger.error('Unable to get bootstrap info.. Existing pypo...') self.logger.error('Unable to get bootstrap info.. Exiting pypo...')
sys.exit(0) sys.exit(1)
else: else:
self.logger.debug('info:%s',info) self.logger.debug('info:%s', info)
for k, v in info['switch_status'].iteritems(): for k, v in info['switch_status'].iteritems():
self.switch_source(self.logger, self.telnet_lock, k, v) self.switch_source(self.logger, self.telnet_lock, k, v)
self.update_liquidsoap_stream_format(info['stream_label']) self.update_liquidsoap_stream_format(info['stream_label'])
self.update_liquidsoap_station_name(info['station_name']) self.update_liquidsoap_station_name(info['station_name'])
self.update_liquidsoap_transition_fade(info['transition_fade']) self.update_liquidsoap_transition_fade(info['transition_fade'])
def write_liquidsoap_config(self, setting): def write_liquidsoap_config(self, setting):
fh = open('/etc/airtime/liquidsoap.cfg', 'w') fh = open('/etc/airtime/liquidsoap.cfg', 'w')
self.logger.info("Rewriting liquidsoap.cfg...") self.logger.info("Rewriting liquidsoap.cfg...")
@ -197,7 +199,7 @@ class PypoFetch(Thread):
if temp == "": if temp == "":
temp = "0" temp = "0"
buffer_str += temp buffer_str += temp
buffer_str += "\n" buffer_str += "\n"
fh.write(api_client.encode_to(buffer_str)) fh.write(api_client.encode_to(buffer_str))
fh.write("log_file = \"/var/log/airtime/pypo-liquidsoap/<script>.log\"\n"); 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. # we could just restart liquidsoap but it take more time somehow.
self.logger.info("Restarting pypo...") self.logger.info("Restarting pypo...")
sys.exit(0) sys.exit(0)
def regenerateLiquidsoapConf(self, setting): def regenerateLiquidsoapConf(self, setting):
existing = {} existing = {}
# create a temp file # create a temp file
setting = sorted(setting.items()) setting = sorted(setting.items())
try: try:
fh = open('/etc/airtime/liquidsoap.cfg', 'r') fh = open('/etc/airtime/liquidsoap.cfg', 'r')
except IOError, e: except IOError, e:
#file does not exist #file does not exist
self.write_liquidsoap_config(setting) self.write_liquidsoap_config(setting)
self.logger.info("Reading existing config...") self.logger.info("Reading existing config...")
# read existing conf file and build dict # read existing conf file and build dict
while True: while True:
@ -226,9 +228,9 @@ class PypoFetch(Thread):
# empty line means EOF # empty line means EOF
if not line: if not line:
break break
line = line.strip() line = line.strip()
if line[0] == "#": if line[0] == "#":
continue continue
@ -243,7 +245,7 @@ class PypoFetch(Thread):
value = '' value = ''
existing[key] = value existing[key] = value
fh.close() fh.close()
# dict flag for any change in cofig # dict flag for any change in cofig
change = {} change = {}
# this flag is to detect disable -> disable change # this flag is to detect disable -> disable change
@ -251,7 +253,7 @@ class PypoFetch(Thread):
state_change_restart = {} state_change_restart = {}
#restart flag #restart flag
restart = False restart = False
self.logger.info("Looking for changes...") self.logger.info("Looking for changes...")
# look for changes # look for changes
for k, s in setting: 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']) self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
restart = True; restart = True;
else: else:
stream, dump = s[u'keyname'].split('_',1) stream, dump = s[u'keyname'].split('_', 1)
if "_output" in s[u'keyname']: if "_output" in s[u'keyname']:
if (existing[s[u'keyname']] != s[u'value']): if (existing[s[u'keyname']] != s[u'value']):
self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname']) self.logger.info("'Need-to-restart' state detected for %s...", s[u'keyname'])
restart = True; restart = True;
state_change_restart[stream] = True state_change_restart[stream] = True
elif ( s[u'value'] != 'disabled'): elif (s[u'value'] != 'disabled'):
state_change_restart[stream] = True state_change_restart[stream] = True
else: else:
state_change_restart[stream] = False state_change_restart[stream] = False
@ -284,10 +286,10 @@ class PypoFetch(Thread):
if not (s[u'value'] == existing[s[u'keyname']]): 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']) self.logger.info("Keyname: %s, Curent value: %s, New Value: %s", s[u'keyname'], existing[s[u'keyname']], s[u'value'])
change[stream] = True change[stream] = True
# set flag change for sound_device alway True # set flag change for sound_device alway True
self.logger.info("Change:%s, State_Change:%s...", change, state_change_restart) self.logger.info("Change:%s, State_Change:%s...", change, state_change_restart)
for k, v in state_change_restart.items(): for k, v in state_change_restart.items():
if k == "sound_device" and v: if k == "sound_device" and v:
restart = True restart = True
@ -306,7 +308,7 @@ class PypoFetch(Thread):
updates the status of liquidsoap connection to the streaming server updates the status of liquidsoap connection to the streaming server
This fucntion updates the bootup time variable in liquidsoap script This fucntion updates the bootup time variable in liquidsoap script
""" """
self.telnet_lock.acquire() self.telnet_lock.acquire()
try: try:
tn = telnetlib.Telnet(LS_HOST, LS_PORT) 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 # we are manually adjusting the bootup time variable so the status msg will get
# updated. # updated.
current_time = time.time() 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(boot_up_time_command)
tn.write("streams.connection_status\n") tn.write("streams.connection_status\n")
tn.write('exit\n') tn.write('exit\n')
output = tn.read_all() output = tn.read_all()
except Exception, e: except Exception, e:
self.logger.error(str(e)) self.logger.error(str(e))
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
output_list = output.split("\r\n") output_list = output.split("\r\n")
stream_info = output_list[2] stream_info = output_list[2]
# streamin info is in the form of: # streamin info is in the form of:
# eg. s1:true,2:true,3:false # eg. s1:true,2:true,3:false
streams = stream_info.split(",") streams = stream_info.split(",")
self.logger.info(streams) self.logger.info(streams)
fake_time = current_time + 1 fake_time = current_time + 1
for s in streams: for s in streams:
info = s.split(':') info = s.split(':')
@ -340,7 +342,7 @@ class PypoFetch(Thread):
status = info[1] status = info[1]
if(status == "true"): if(status == "true"):
self.api_client.notify_liquidsoap_status("OK", stream_id, str(fake_time)) self.api_client.notify_liquidsoap_status("OK", stream_id, str(fake_time))
def update_liquidsoap_stream_format(self, stream_format): def update_liquidsoap_stream_format(self, stream_format):
# Push stream metadata to liquidsoap # Push stream metadata to liquidsoap
# TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!! # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
@ -356,7 +358,7 @@ class PypoFetch(Thread):
self.logger.error("Exception %s", e) self.logger.error("Exception %s", e)
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
def update_liquidsoap_transition_fade(self, fade): def update_liquidsoap_transition_fade(self, fade):
# Push stream metadata to liquidsoap # Push stream metadata to liquidsoap
# TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!! # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
@ -372,14 +374,14 @@ class PypoFetch(Thread):
self.logger.error("Exception %s", e) self.logger.error("Exception %s", e)
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
def update_liquidsoap_station_name(self, station_name): def update_liquidsoap_station_name(self, station_name):
# Push stream metadata to liquidsoap # Push stream metadata to liquidsoap
# TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!! # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
try: try:
self.logger.info(LS_HOST) self.logger.info(LS_HOST)
self.logger.info(LS_PORT) self.logger.info(LS_PORT)
self.telnet_lock.acquire() self.telnet_lock.acquire()
try: try:
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
@ -387,7 +389,7 @@ class PypoFetch(Thread):
self.logger.info(command) self.logger.info(command)
tn.write(command) tn.write(command)
tn.write('exit\n') tn.write('exit\n')
tn.read_all() tn.read_all()
except Exception, e: except Exception, e:
self.logger.error(str(e)) self.logger.error(str(e))
finally: finally:
@ -403,7 +405,7 @@ class PypoFetch(Thread):
to the cache dir (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss) to the cache dir (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss)
- runs the cleanup routine, to get rid of unused cached files - 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.last_update_schedule_timestamp = time.time()
self.logger.debug(schedule_data) self.logger.debug(schedule_data)
media = schedule_data["media"] media = schedule_data["media"]
@ -411,7 +413,7 @@ class PypoFetch(Thread):
# Download all the media and put playlists in liquidsoap "annotate" format # Download all the media and put playlists in liquidsoap "annotate" format
try: try:
""" """
Make sure cache_dir exists Make sure cache_dir exists
""" """
@ -420,15 +422,16 @@ class PypoFetch(Thread):
os.makedirs(download_dir) os.makedirs(download_dir)
except Exception, e: except Exception, e:
pass pass
for key in media: for key in media:
media_item = media[key] media_item = media[key]
if(media_item['type'] == 'file'): if(media_item['type'] == 'file'):
fileExt = os.path.splitext(media_item['uri'])[1] 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['dst'] = dst
media_item['started_copying'] = False
media_filtered[key] = media_item media_filtered[key] = media_item
self.media_prepare_queue.put(copy.copy(media_filtered)) self.media_prepare_queue.put(copy.copy(media_filtered))
except Exception, e: self.logger.error("%s", e) except Exception, e: self.logger.error("%s", e)
@ -440,7 +443,7 @@ class PypoFetch(Thread):
# cleanup # cleanup
try: self.cache_cleanup(media) try: self.cache_cleanup(media)
except Exception, e: self.logger.error("%s", e) except Exception, e: self.logger.error("%s", e)
def cache_cleanup(self, media): def cache_cleanup(self, media):
""" """
Get list of all files in the cache dir and remove them if they aren't being used anymore. 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)) cached_file_set = set(os.listdir(self.cache_dir))
scheduled_file_set = set() scheduled_file_set = set()
for mkey in media: for mkey in media:
media_item = media[mkey] media_item = media[mkey]
fileExt = os.path.splitext(media_item['uri'])[1] fileExt = os.path.splitext(media_item['uri'])[1]
scheduled_file_set.add(media_item["id"] + fileExt) scheduled_file_set.add(media_item["id"] + fileExt)
unneeded_files = cached_file_set - scheduled_file_set unneeded_files = cached_file_set - scheduled_file_set
self.logger.debug("Files to remove " + str(unneeded_files)) self.logger.debug("Files to remove " + str(unneeded_files))
for file in unneeded_files: for f in unneeded_files:
self.logger.debug("Removing %s" % os.path.join(self.cache_dir, file)) self.logger.debug("Removing %s" % os.path.join(self.cache_dir, f))
os.remove(os.path.join(self.cache_dir, file)) os.remove(os.path.join(self.cache_dir, f))
def main(self): def main(self):
# Bootstrap: since we are just starting up, we need to grab the # 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.process_schedule(self.schedule_data)
self.set_bootstrap_variables() self.set_bootstrap_variables()
loops = 1 loops = 1
while True: while True:
self.logger.info("Loop #%s", loops) self.logger.info("Loop #%s", loops)
try: try:
""" """
our simple_queue.get() requires a timeout, in which case we our simple_queue.get() requires a timeout, in which case we
fetch the Airtime schedule manually. It is important to fetch 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 sent, and we will have very stale (or non-existent!) data about the
schedule. 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) message = self.fetch_queue.get(block=True, timeout=self.listener_timeout)
self.handle_message(message) self.handle_message(message)
except Empty, e:
self.logger.info("Queue timeout. Fetching schedule manually")
except Exception, e: except Exception, e:
import traceback import traceback
top = traceback.format_exc() top = traceback.format_exc()
self.logger.error('Exception: %s', e) self.logger.error('Exception: %s', e)
self.logger.error("traceback: %s", top) self.logger.error("traceback: %s", top)
success, self.schedule_data = self.api_client.get_schedule() success, self.schedule_data = self.api_client.get_schedule()
if success: if success:
self.process_schedule(self.schedule_data) self.process_schedule(self.schedule_data)

6
python_apps/pypo/pypofile.py
View file

@ -5,7 +5,6 @@ from Queue import Empty
from configobj import ConfigObj from configobj import ConfigObj
import logging import logging
import logging.config
import shutil import shutil
import os import os
import sys import sys
@ -71,13 +70,16 @@ class PypoFile(Thread):
if do_copy: if do_copy:
self.logger.debug("copying from %s to local cache %s" % (src, dst)) self.logger.debug("copying from %s to local cache %s" % (src, dst))
try: try:
media_item['started_copying'] = True
""" """
copy will overwrite dst if it already exists copy will overwrite dst if it already exists
""" """
shutil.copy(src, dst) shutil.copy(src, dst)
#make file world readable #make file world readable
os.chmod(dst, stat.S_IRUSR | stat.S_IRGRP | stat.S_IXOTH) os.chmod(dst, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
except Exception, e: except Exception, e:
self.logger.error("Could not copy from %s to %s" % (src, dst)) self.logger.error("Could not copy from %s to %s" % (src, dst))
self.logger.error(e) self.logger.error(e)

21
python_apps/pypo/pypomessagehandler.py
View file

@ -1,7 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import logging import logging
import logging.config
import sys import sys
from configobj import ConfigObj from configobj import ConfigObj
from threading import Thread from threading import Thread
@ -39,7 +38,7 @@ class PypoMessageHandler(Thread):
self.logger = logging.getLogger('message_h') self.logger = logging.getLogger('message_h')
self.pypo_queue = pq self.pypo_queue = pq
self.recorder_queue = rq self.recorder_queue = rq
def init_rabbit_mq(self): def init_rabbit_mq(self):
self.logger.info("Initializing RabbitMQ stuff") self.logger.info("Initializing RabbitMQ stuff")
try: try:
@ -51,21 +50,21 @@ class PypoMessageHandler(Thread):
except Exception, e: except Exception, e:
self.logger.error(e) self.logger.error(e)
return False return False
return True return True
""" """
Handle a message from RabbitMQ, put it into our yucky global var. Handle a message from RabbitMQ, put it into our yucky global var.
Hopefully there is a better way to do this. Hopefully there is a better way to do this.
""" """
def handle_message(self, message): def handle_message(self, message):
try: try:
self.logger.info("Received event from RabbitMQ: %s" % message) self.logger.info("Received event from RabbitMQ: %s" % message)
m = json.loads(message) m = json.loads(message)
command = m['event_type'] command = m['event_type']
self.logger.info("Handling command: " + command) self.logger.info("Handling command: " + command)
if command == 'update_schedule': if command == 'update_schedule':
self.logger.info("Updating schdule...") self.logger.info("Updating schdule...")
self.pypo_queue.put(message) self.pypo_queue.put(message)
@ -121,15 +120,13 @@ class PypoMessageHandler(Thread):
while loop and eat all the CPU while loop and eat all the CPU
""" """
time.sleep(5) time.sleep(5)
""" """
There is a problem with the RabbitMq messenger service. Let's There is a problem with the RabbitMq messenger service. Let's
log the error and get the schedule via HTTP polling 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('Exception: %s', e)
self.logger.error("traceback: %s", top) self.logger.error("traceback: %s", traceback.format_exc())
loops += 1 loops += 1

61
python_apps/pypo/pypo-notify.py → python_apps/pypo/pyponotify.py
View file

@ -15,19 +15,9 @@ Main case:
""" """
# python defaults (debian default) from optparse import OptionParser
import time
import os
import traceback
from optparse import *
import sys import sys
import time
import datetime
import logging
import logging.config import logging.config
import urllib
import urllib2
import string
import json import json
# additional modules (should be checked) # additional modules (should be checked)
@ -38,9 +28,6 @@ from configobj import ConfigObj
from api_clients import * from api_clients import *
from std_err_override import LogWriter from std_err_override import LogWriter
# Set up command-line options
parser = OptionParser()
# help screeen / info # help screeen / info
usage = "%prog [options]" + " - notification gateway" usage = "%prog [options]" + " - notification gateway"
parser = OptionParser(usage=usage) parser = OptionParser(usage=usage)
@ -60,7 +47,7 @@ parser.add_option("-y", "--source-status", help="source connection stauts", meta
# configure logging # configure logging
logging.config.fileConfig("logging.cfg") logging.config.fileConfig("logging.cfg")
logger = logging.getLogger() logger = logging.getLogger('notify')
LogWriter.override_std_err(logger) LogWriter.override_std_err(logger)
#need to wait for Python 2.7 for this.. #need to wait for Python 2.7 for this..
@ -69,54 +56,54 @@ LogWriter.override_std_err(logger)
# loading config file # loading config file
try: try:
config = ConfigObj('/etc/airtime/pypo.cfg') config = ConfigObj('/etc/airtime/pypo.cfg')
except Exception, e: except Exception, e:
logger.error('Error loading config file: %s', e) logger.error('Error loading config file: %s', e)
sys.exit() sys.exit()
class Notify: class Notify:
def __init__(self): def __init__(self):
self.api_client = api_client.api_client_factory(config) self.api_client = api_client.api_client_factory(config)
def notify_media_start_playing(self, data, media_id): def notify_media_start_playing(self, data, media_id):
logger = logging.getLogger("notify") logger = logging.getLogger("notify")
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('# Calling server to update about what\'s playing #') logger.debug('# Calling server to update about what\'s playing #')
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('data = '+ str(data)) logger.debug('data = ' + str(data))
response = self.api_client.notify_media_item_start_playing(data, media_id) response = self.api_client.notify_media_item_start_playing(data, media_id)
logger.debug("Response: "+json.dumps(response)) logger.debug("Response: " + json.dumps(response))
# @pram time: time that LS started # @pram time: time that LS started
def notify_liquidsoap_status(self, msg, stream_id, time): def notify_liquidsoap_status(self, msg, stream_id, time):
logger = logging.getLogger("notify") logger = logging.getLogger("notify")
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('# Calling server to update liquidsoap status #') logger.debug('# Calling server to update liquidsoap status #')
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('msg = '+ str(msg)) logger.debug('msg = ' + str(msg))
response = self.api_client.notify_liquidsoap_status(msg, stream_id, time) response = self.api_client.notify_liquidsoap_status(msg, stream_id, time)
logger.debug("Response: "+json.dumps(response)) logger.debug("Response: " + json.dumps(response))
def notify_source_status(self, source_name, status): def notify_source_status(self, source_name, status):
logger = logging.getLogger("notify") logger = logging.getLogger("notify")
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('# Calling server to update source status #') logger.debug('# Calling server to update source status #')
logger.debug('#################################################') logger.debug('#################################################')
logger.debug('msg = '+ str(source_name) + ' : ' + str(status)) logger.debug('msg = ' + str(source_name) + ' : ' + str(status))
response = self.api_client.notify_source_status(source_name, status) response = self.api_client.notify_source_status(source_name, status)
logger.debug("Response: "+json.dumps(response)) logger.debug("Response: " + json.dumps(response))
if __name__ == '__main__': if __name__ == '__main__':
print print
print '#########################################' print '#########################################'
print '# *** pypo *** #' print '# *** pypo *** #'
print '# pypo notification gateway #' print '# pypo notification gateway #'
print '#########################################' print '#########################################'
# initialize # initialize
logger = logging.getLogger("notify") logger = logging.getLogger("notify")
if options.error and options.stream_id: if options.error and options.stream_id:
@ -141,11 +128,11 @@ if __name__ == '__main__':
if not options.data: if not options.data:
print "NOTICE: 'data' command-line argument not given." print "NOTICE: 'data' command-line argument not given."
sys.exit() sys.exit()
if not options.media_id: if not options.media_id:
print "NOTICE: 'media_id' command-line argument not given." print "NOTICE: 'media_id' command-line argument not given."
sys.exit() sys.exit()
try: try:
n = Notify() n = Notify()
n.notify_media_start_playing(options.data, options.media_id) n.notify_media_start_playing(options.data, options.media_id)

186
python_apps/pypo/pypopush.py
View file

@ -5,11 +5,9 @@ from datetime import timedelta
import sys import sys
import time import time
import logging
import logging.config import logging.config
import telnetlib import telnetlib
import calendar import calendar
import json
import math import math
from pypofetch import PypoFetch from pypofetch import PypoFetch
@ -51,47 +49,46 @@ class PypoPush(Thread):
self.pushed_objects = {} self.pushed_objects = {}
self.logger = logging.getLogger('push') self.logger = logging.getLogger('push')
def main(self): def main(self):
loops = 0 loops = 0
heartbeat_period = math.floor(30/PUSH_INTERVAL) heartbeat_period = math.floor(30 / PUSH_INTERVAL)
next_media_item_chain = None next_media_item_chain = None
media_schedule = None media_schedule = None
time_until_next_play = None time_until_next_play = None
chains = None chains = None
while True: while True:
try: try:
if time_until_next_play is None: if time_until_next_play is None:
media_schedule = self.queue.get(block=True) media_schedule = self.queue.get(block=True)
else: else:
media_schedule = self.queue.get(block=True, timeout=time_until_next_play) 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. #We get to the following lines only if a schedule was received.
liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap() liquidsoap_queue_approx = self.get_queue_items_from_liquidsoap()
current_event_chain = self.get_current_chain(chains) current_event_chain = self.get_current_chain(chains)
if len(current_event_chain) > 0 and len(liquidsoap_queue_approx) == 0: if len(current_event_chain) > 0 and len(liquidsoap_queue_approx) == 0:
#Something is scheduled but Liquidsoap is not playing anything! #Something is scheduled but Liquidsoap is not playing anything!
#Need to schedule it immediately..this might happen if Liquidsoap crashed. #Need to schedule it immediately..this might happen if Liquidsoap crashed.
chains.remove(current_event_chain) chains.remove(current_event_chain)
self.modify_cue_point(current_event_chain[0]) self.modify_cue_point(current_event_chain[0])
next_media_item_chain = current_event_chain next_media_item_chain = current_event_chain
time_until_next_play = 0 time_until_next_play = 0
#sleep for 0.2 seconds to give pypo-file time to copy. This is a quick #sleep for 0.2 seconds to give pypo-file time to copy.
#fix that will be improved in 2.1.1
time.sleep(0.2) time.sleep(0.2)
else: else:
media_chain = filter(lambda item: (item["type"] == "file"), current_event_chain) media_chain = filter(lambda item: (item["type"] == "file"), current_event_chain)
self.handle_new_media_schedule(media_schedule, liquidsoap_queue_approx, media_chain) self.handle_new_media_schedule(media_schedule, liquidsoap_queue_approx, media_chain)
next_media_item_chain = self.get_next_schedule_chain(chains) next_media_item_chain = self.get_next_schedule_chain(chains)
self.logger.debug("Next schedule chain: %s", next_media_item_chain) self.logger.debug("Next schedule chain: %s", next_media_item_chain)
if next_media_item_chain is not None: if next_media_item_chain is not None:
chains.remove(next_media_item_chain) chains.remove(next_media_item_chain)
tnow = datetime.utcnow() tnow = datetime.utcnow()
@ -104,7 +101,7 @@ class PypoPush(Thread):
except Empty, e: except Empty, e:
#We only get here when a new chain of tracks are ready to be played. #We only get here when a new chain of tracks are ready to be played.
self.push_to_liquidsoap(next_media_item_chain) 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)
if next_media_item_chain is not None: if next_media_item_chain is not None:
tnow = datetime.utcnow() tnow = datetime.utcnow()
@ -114,7 +111,7 @@ class PypoPush(Thread):
else: else:
self.logger.debug("Blocking indefinitely since no show scheduled next") self.logger.debug("Blocking indefinitely since no show scheduled next")
time_until_next_play = None time_until_next_play = None
if loops % heartbeat_period == 0: if loops % heartbeat_period == 0:
self.logger.info("heartbeat") self.logger.info("heartbeat")
loops = 0 loops = 0
@ -127,7 +124,7 @@ class PypoPush(Thread):
try: try:
self.telnet_lock.acquire() self.telnet_lock.acquire()
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
msg = 'queue.queue\n' msg = 'queue.queue\n'
tn.write(msg) tn.write(msg)
response = tn.read_until("\r\n").strip(" \r\n") response = tn.read_until("\r\n").strip(" \r\n")
@ -138,14 +135,14 @@ class PypoPush(Thread):
response = [] response = []
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
liquidsoap_queue_approx = [] liquidsoap_queue_approx = []
if len(response) > 0: if len(response) > 0:
items_in_queue = response.split(" ") items_in_queue = response.split(" ")
self.logger.debug("items_in_queue: %s", items_in_queue) self.logger.debug("items_in_queue: %s", items_in_queue)
for item in items_in_queue: for item in items_in_queue:
if item in self.pushed_objects: if item in self.pushed_objects:
liquidsoap_queue_approx.append(self.pushed_objects[item]) liquidsoap_queue_approx.append(self.pushed_objects[item])
@ -158,9 +155,9 @@ class PypoPush(Thread):
self.clear_liquidsoap_queue() self.clear_liquidsoap_queue()
liquidsoap_queue_approx = [] liquidsoap_queue_approx = []
break break
return liquidsoap_queue_approx return liquidsoap_queue_approx
def handle_new_media_schedule(self, media_schedule, liquidsoap_queue_approx, media_chain): def handle_new_media_schedule(self, media_schedule, liquidsoap_queue_approx, media_chain):
""" """
This function's purpose is to gracefully handle situations where This function's purpose is to gracefully handle situations where
@ -169,27 +166,27 @@ class PypoPush(Thread):
call other functions that will connect to Liquidsoap and alter its call other functions that will connect to Liquidsoap and alter its
queue. 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: if problem_at_iteration is not None:
#Items that are in Liquidsoap's queue aren't scheduled anymore. We need to connect #Items that are in Liquidsoap's queue aren't scheduled anymore. We need to connect
#and remove these items. #and remove these items.
self.logger.debug("Change in link %s of current chain", problem_at_iteration) 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:]) 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): if problem_at_iteration is None and len(media_chain) > len(liquidsoap_queue_approx):
self.logger.debug("New schedule has longer current chain.") self.logger.debug("New schedule has longer current chain.")
problem_at_iteration = len(liquidsoap_queue_approx) problem_at_iteration = len(liquidsoap_queue_approx)
if problem_at_iteration is not None: if problem_at_iteration is not None:
self.logger.debug("Change in chain at link %s", problem_at_iteration) self.logger.debug("Change in chain at link %s", problem_at_iteration)
chain_to_push = media_chain[problem_at_iteration:] chain_to_push = media_chain[problem_at_iteration:]
if len(chain_to_push) > 0: if len(chain_to_push) > 0:
self.modify_cue_point(chain_to_push[0]) self.modify_cue_point(chain_to_push[0])
self.push_to_liquidsoap(chain_to_push) self.push_to_liquidsoap(chain_to_push)
""" """
Compare whats in the liquidsoap_queue to the new schedule we just Compare whats in the liquidsoap_queue to the new schedule we just
received in media_schedule. This function only iterates over liquidsoap_queue_approx received in media_schedule. This function only iterates over liquidsoap_queue_approx
@ -201,7 +198,6 @@ class PypoPush(Thread):
#see if they are the same as the newly received schedule #see if they are the same as the newly received schedule
iteration = 0 iteration = 0
problem_at_iteration = None problem_at_iteration = None
problem_start_time = None
for queue_item in liquidsoap_queue_approx: for queue_item in liquidsoap_queue_approx:
if queue_item['start'] in media_schedule.keys(): if queue_item['start'] in media_schedule.keys():
media_item = media_schedule[queue_item['start']] media_item = media_schedule[queue_item['start']]
@ -211,33 +207,30 @@ class PypoPush(Thread):
pass pass
else: else:
problem_at_iteration = iteration problem_at_iteration = iteration
problem_start_time = queue_item['start'] break
break
else: else:
#A different item has been scheduled at the same time! Need to remove #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 #all tracks from the Liquidsoap queue starting at this point, and re-add
#them. #them.
problem_at_iteration = iteration problem_at_iteration = iteration
problem_start_time = queue_item['start']
break break
else: else:
#There are no more items scheduled for this time! The user has shortened #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. #the playlist, so we simply need to remove tracks from the queue.
problem_at_iteration = iteration problem_at_iteration = iteration
problem_start_time = queue_item['start']
break break
iteration+=1 iteration += 1
return (problem_at_iteration, problem_start_time) return problem_at_iteration
def get_all_chains(self, media_schedule): def get_all_chains(self, media_schedule):
chains = [] chains = []
current_chain = [] current_chain = []
sorted_keys = sorted(media_schedule.keys()) sorted_keys = sorted(media_schedule.keys())
for mkey in sorted_keys: for mkey in sorted_keys:
media_item = media_schedule[mkey] media_item = media_schedule[mkey]
if media_item['type'] == "event": if media_item['type'] == "event":
@ -251,26 +244,26 @@ class PypoPush(Thread):
#Start a new one instead #Start a new one instead
chains.append(current_chain) chains.append(current_chain)
current_chain = [media_item] current_chain = [media_item]
if len(current_chain) > 0: if len(current_chain) > 0:
chains.append(current_chain) chains.append(current_chain)
return chains return chains
def modify_cue_point(self, link): def modify_cue_point(self, link):
tnow = datetime.utcnow() tnow = datetime.utcnow()
link_start = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S") link_start = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S")
diff_td = tnow - link_start diff_td = tnow - link_start
diff_sec = self.date_interval_to_seconds(diff_td) diff_sec = self.date_interval_to_seconds(diff_td)
if diff_sec > 0: if diff_sec > 0:
self.logger.debug("media item was supposed to start %s ago. Preparing to start..", diff_sec) 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'])) original_cue_in_td = timedelta(seconds=float(link['cue_in']))
link['cue_in'] = self.date_interval_to_seconds(original_cue_in_td) + diff_sec link['cue_in'] = self.date_interval_to_seconds(original_cue_in_td) + diff_sec
def get_current_chain(self, chains): def get_current_chain(self, chains):
tnow = datetime.utcnow() tnow = datetime.utcnow()
current_chain = [] current_chain = []
@ -280,22 +273,22 @@ class PypoPush(Thread):
for link in chain: for link in chain:
link_start = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S") link_start = datetime.strptime(link['start'], "%Y-%m-%d-%H-%M-%S")
link_end = datetime.strptime(link['end'], "%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) self.logger.debug("tnow %s, chain_start %s", tnow, link_start)
if link_start <= tnow and tnow < link_end: if link_start <= tnow and tnow < link_end:
current_chain = chain[iteration:] current_chain = chain[iteration:]
break break
iteration += 1 iteration += 1
return current_chain return current_chain
""" """
The purpose of this function is to take a look at the last received schedule from 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 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 of media_items where the end time of media_item 'n' is the start time of media_item
'n+1' 'n+1'
""" """
def get_next_schedule_chain(self, chains): def get_next_schedule_chain(self, chains):
#all media_items are now divided into chains. Let's find the one that #all media_items are now divided into chains. Let's find the one that
#starts closest in the future. #starts closest in the future.
tnow = datetime.utcnow() tnow = datetime.utcnow()
@ -307,19 +300,32 @@ class PypoPush(Thread):
if (closest_start == None or chain_start < closest_start) and chain_start > tnow: if (closest_start == None or chain_start < closest_start) and chain_start > tnow:
closest_start = chain_start closest_start = chain_start
closest_chain = chain closest_chain = chain
return closest_chain return closest_chain
def date_interval_to_seconds(self, interval): 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): def push_to_liquidsoap(self, event_chain):
try: try:
for media_item in event_chain: for media_item in event_chain:
if media_item['type'] == "file": 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": elif media_item['type'] == "event":
if media_item['event_type'] == "kick_out": if media_item['event_type'] == "kick_out":
PypoFetch.disconnect_source(self.logger, self.telnet_lock, "live_dj") PypoFetch.disconnect_source(self.logger, self.telnet_lock, "live_dj")
@ -327,27 +333,27 @@ class PypoPush(Thread):
PypoFetch.switch_source(self.logger, self.telnet_lock, "live_dj", "off") PypoFetch.switch_source(self.logger, self.telnet_lock, "live_dj", "off")
except Exception, e: except Exception, e:
self.logger.error('Pypo Push Exception: %s', e) self.logger.error('Pypo Push Exception: %s', e)
def clear_liquidsoap_queue(self): def clear_liquidsoap_queue(self):
self.logger.debug("Clearing Liquidsoap queue") self.logger.debug("Clearing Liquidsoap queue")
try: try:
self.telnet_lock.acquire() self.telnet_lock.acquire()
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
msg = "source.skip\n" msg = "source.skip\n"
tn.write(msg) tn.write(msg)
tn.write("exit\n") tn.write("exit\n")
tn.read_all() tn.read_all()
except Exception, e: except Exception, e:
self.logger.error(str(e)) self.logger.error(str(e))
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
def remove_from_liquidsoap_queue(self, problem_at_iteration, liquidsoap_queue_approx): def remove_from_liquidsoap_queue(self, problem_at_iteration, liquidsoap_queue_approx):
try: try:
self.telnet_lock.acquire() self.telnet_lock.acquire()
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
if problem_at_iteration == 0: if problem_at_iteration == 0:
msg = "source.skip\n" msg = "source.skip\n"
self.logger.debug(msg) self.logger.debug(msg)
@ -355,13 +361,13 @@ class PypoPush(Thread):
else: else:
# Remove things in reverse order. # Remove things in reverse order.
queue_copy = liquidsoap_queue_approx[::-1] queue_copy = liquidsoap_queue_approx[::-1]
for queue_item in queue_copy: for queue_item in queue_copy:
msg = "queue.remove %s\n" % queue_item['queue_id'] msg = "queue.remove %s\n" % queue_item['queue_id']
self.logger.debug(msg) self.logger.debug(msg)
tn.write(msg) tn.write(msg)
response = tn.read_until("\r\n").strip("\r\n") response = tn.read_until("\r\n").strip("\r\n")
if "No such request in my queue" in response: if "No such request in my queue" in response:
""" """
Cannot remove because Liquidsoap started playing the item. Need Cannot remove because Liquidsoap started playing the item. Need
@ -370,33 +376,33 @@ class PypoPush(Thread):
msg = "source.skip\n" msg = "source.skip\n"
self.logger.debug(msg) self.logger.debug(msg)
tn.write(msg) tn.write(msg)
msg = "queue.queue\n" msg = "queue.queue\n"
self.logger.debug(msg) self.logger.debug(msg)
tn.write(msg) tn.write(msg)
tn.write("exit\n") tn.write("exit\n")
self.logger.debug(tn.read_all()) self.logger.debug(tn.read_all())
except Exception, e: except Exception, e:
self.logger.error(str(e)) self.logger.error(str(e))
finally: finally:
self.telnet_lock.release() self.telnet_lock.release()
def sleep_until_start(self, media_item): def sleep_until_start(self, media_item):
""" """
The purpose of this function is to look at the difference between 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. "now" and when the media_item starts, and sleep for that period of time.
After waking from sleep, this function returns. After waking from sleep, this function returns.
""" """
mi_start = media_item['start'][0:19] mi_start = media_item['start'][0:19]
#strptime returns struct_time in local time #strptime returns struct_time in local time
epoch_start = calendar.timegm(time.strptime(mi_start, '%Y-%m-%d-%H-%M-%S')) 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. #Return the time as a floating point number expressed in seconds since the epoch, in UTC.
epoch_now = time.time() epoch_now = time.time()
self.logger.debug("Epoch start: %s" % epoch_start) self.logger.debug("Epoch start: %s" % epoch_start)
self.logger.debug("Epoch now: %s" % epoch_now) self.logger.debug("Epoch now: %s" % epoch_now)
@ -417,43 +423,43 @@ class PypoPush(Thread):
try: try:
self.telnet_lock.acquire() self.telnet_lock.acquire()
tn = telnetlib.Telnet(LS_HOST, LS_PORT) tn = telnetlib.Telnet(LS_HOST, LS_PORT)
#tn.write(("vars.pypo_data %s\n"%liquidsoap_data["schedule_id"]).encode('utf-8')) #tn.write(("vars.pypo_data %s\n"%liquidsoap_data["schedule_id"]).encode('utf-8'))
annotation = self.create_liquidsoap_annotation(media_item) annotation = self.create_liquidsoap_annotation(media_item)
msg = 'queue.push %s\n' % annotation.encode('utf-8') msg = 'queue.push %s\n' % annotation.encode('utf-8')
self.logger.debug(msg) self.logger.debug(msg)
tn.write(msg) tn.write(msg)
queue_id = tn.read_until("\r\n").strip("\r\n") queue_id = tn.read_until("\r\n").strip("\r\n")
#remember the media_item's queue id which we may use #remember the media_item's queue id which we may use
#later if we need to remove it from the queue. #later if we need to remove it from the queue.
media_item['queue_id'] = queue_id media_item['queue_id'] = queue_id
#add media_item to the end of our queue #add media_item to the end of our queue
self.pushed_objects[queue_id] = media_item self.pushed_objects[queue_id] = media_item
show_name = media_item['show_name'] show_name = media_item['show_name']
msg = 'vars.show_name %s\n' % show_name.encode('utf-8') msg = 'vars.show_name %s\n' % show_name.encode('utf-8')
tn.write(msg) tn.write(msg)
self.logger.debug(msg) self.logger.debug(msg)
tn.write("exit\n") tn.write("exit\n")
self.logger.debug(tn.read_all()) self.logger.debug(tn.read_all())
except Exception, e: except Exception, e:
self.logger.error(str(e)) self.logger.error(str(e))
finally: finally:
self.telnet_lock.release() 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. # 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' \ 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): def run(self):
try: self.main() try: self.main()
except Exception, e: except Exception, e:
import traceback import traceback
top = traceback.format_exc() top = traceback.format_exc()
self.logger.error('Pypo Push Exception: %s', top) self.logger.error('Pypo Push Exception: %s', top)

31
python_apps/pypo/recorder.py
View file

@ -1,7 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import logging import logging
import logging.config
import json import json
import time import time
import datetime import datetime
@ -55,7 +54,7 @@ class ShowRecorder(Thread):
self.p = None self.p = None
def record_show(self): def record_show(self):
length = str(self.filelength)+".0" length = str(self.filelength) + ".0"
filename = self.start_time filename = self.start_time
filename = filename.replace(" ", "-") filename = filename.replace(" ", "-")
@ -128,7 +127,7 @@ class ShowRecorder(Thread):
time = md[1].replace(":", "-") time = md[1].replace(":", "-")
self.logger.info("time: %s" % time) self.logger.info("time: %s" % time)
name = time+"-"+self.show_name name = time + "-" + self.show_name
artist = "Airtime Show Recorder" artist = "Airtime Show Recorder"
#set some metadata for our file daemon #set some metadata for our file daemon
@ -181,7 +180,7 @@ class Recorder(Thread):
def handle_message(self): def handle_message(self):
if not self.queue.empty(): if not self.queue.empty():
message = self.queue.get() message = self.queue.get()
msg = json.loads(message) msg = json.loads(message)
command = msg["event_type"] command = msg["event_type"]
self.logger.info("Received msg from Pypo Message Handler: %s", msg) self.logger.info("Received msg from Pypo Message Handler: %s", msg)
if command == 'cancel_recording': if command == 'cancel_recording':
@ -190,10 +189,10 @@ class Recorder(Thread):
else: else:
self.process_recorder_schedule(msg) self.process_recorder_schedule(msg)
self.loops = 0 self.loops = 0
if self.shows_to_record: if self.shows_to_record:
self.start_record() self.start_record()
def process_recorder_schedule(self, m): def process_recorder_schedule(self, m):
self.logger.info("Parsing recording show schedules...") self.logger.info("Parsing recording show schedules...")
temp_shows_to_record = {} temp_shows_to_record = {}
@ -217,7 +216,7 @@ class Recorder(Thread):
delta = next_show - tnow delta = next_show - tnow
s = '%s.%s' % (delta.seconds, delta.microseconds) s = '%s.%s' % (delta.seconds, delta.microseconds)
out = float(s) out = float(s)
if out < 5: if out < 5:
self.logger.debug("Shows %s", self.shows_to_record) self.logger.debug("Shows %s", self.shows_to_record)
self.logger.debug("Next show %s", next_show) self.logger.debug("Next show %s", next_show)
@ -231,26 +230,26 @@ class Recorder(Thread):
if delta < 5: if delta < 5:
self.logger.debug("sleeping %s seconds until show", delta) self.logger.debug("sleeping %s seconds until show", delta)
time.sleep(delta) time.sleep(delta)
sorted_show_keys = sorted(self.shows_to_record.keys()) sorted_show_keys = sorted(self.shows_to_record.keys())
start_time = sorted_show_keys[0] start_time = sorted_show_keys[0]
show_length = self.shows_to_record[start_time][0] show_length = self.shows_to_record[start_time][0]
show_instance = self.shows_to_record[start_time][1] show_instance = self.shows_to_record[start_time][1]
show_name = self.shows_to_record[start_time][2] show_name = self.shows_to_record[start_time][2]
server_timezone = self.shows_to_record[start_time][3] server_timezone = self.shows_to_record[start_time][3]
T = pytz.timezone(server_timezone) T = pytz.timezone(server_timezone)
start_time_on_UTC = getDateTimeObj(start_time) start_time_on_UTC = getDateTimeObj(start_time)
start_time_on_server = start_time_on_UTC.replace(tzinfo=pytz.utc).astimezone(T) 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' % \ 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} '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 = ShowRecorder(show_instance, show_name, show_length.seconds, start_time_formatted)
self.sr.start() self.sr.start()
#remove show from shows to record. #remove show from shows to record.
del self.shows_to_record[start_time] del self.shows_to_record[start_time]
#self.time_till_next_show = self.get_time_till_next_show() #self.time_till_next_show = self.get_time_till_next_show()
except Exception,e : except Exception, e :
import traceback import traceback
top = traceback.format_exc() top = traceback.format_exc()
self.logger.error('Exception: %s', e) self.logger.error('Exception: %s', e)
@ -273,12 +272,12 @@ class Recorder(Thread):
self.logger.info("Bootstrap recorder schedule received: %s", temp) self.logger.info("Bootstrap recorder schedule received: %s", temp)
except Exception, e: except Exception, e:
self.logger.error(e) self.logger.error(e)
self.logger.info("Bootstrap complete: got initial copy of the schedule") self.logger.info("Bootstrap complete: got initial copy of the schedule")
self.loops = 0 self.loops = 0
heartbeat_period = math.floor(30/PUSH_INTERVAL) heartbeat_period = math.floor(30 / PUSH_INTERVAL)
while True: while True:
if self.loops % heartbeat_period == 0: if self.loops % heartbeat_period == 0:
self.logger.info("heartbeat") self.logger.info("heartbeat")
@ -299,7 +298,7 @@ class Recorder(Thread):
self.logger.error('Pypo Recorder Exception: %s', e) self.logger.error('Pypo Recorder Exception: %s', e)
time.sleep(PUSH_INTERVAL) time.sleep(PUSH_INTERVAL)
self.loops += 1 self.loops += 1
except Exception,e : except Exception, e :
import traceback import traceback
top = traceback.format_exc() top = traceback.format_exc()
self.logger.error('Exception: %s', e) self.logger.error('Exception: %s', e)