<commit_before>import os
import sys
import time
import logging
import logging.config
import shutil
import pickle
import random
import string
import json
import telnetlib
import math
from threading import Thread
from subprocess import Popen, PIPE

# For RabbitMQ
from kombu.connection import BrokerConnection
from kombu.messaging import Exchange, Queue, Consumer, Producer

from api_clients import api_client
from util import CueFile

from configobj import ConfigObj

# configure logging
logging.config.fileConfig("logging.cfg")

# loading config file
try:
    config = ConfigObj('config.cfg')
    LS_HOST = config['ls_host']
    LS_PORT = config['ls_port']
    POLL_INTERVAL = int(config['poll_interval'])

except Exception, e:
    print 'Error loading config file: ', e
    sys.exit()

# Yuk - using a global, i know!
SCHEDULE_PUSH_MSG = []

"""
Handle a message from RabbitMQ, put it into our yucky global var.
Hopefully there is a better way to do this.
"""
def handle_message(body, message):
    logger = logging.getLogger('fetch')
    global SCHEDULE_PUSH_MSG
    logger.info("Received schedule from RabbitMQ: " + message.body)
    SCHEDULE_PUSH_MSG = json.loads(message.body)
    # ACK the message to take it off the queue
    message.ack()


class PypoFetch(Thread):
    def __init__(self, q):
        Thread.__init__(self)
        logger = logging.getLogger('fetch')
        self.api_client = api_client.api_client_factory(config)
        self.cue_file = CueFile()
        self.set_export_source('scheduler')
        self.queue = q

        logger.info("Initializing RabbitMQ stuff")
        schedule_exchange = Exchange("airtime-schedule", "direct", durable=True, auto_delete=True)
        schedule_queue = Queue("pypo-fetch", exchange=schedule_exchange, key="foo")
        self.connection = BrokerConnection(config["rabbitmq_host"], config["rabbitmq_user"], config["rabbitmq_password"], "/")
        channel = self.connection.channel()
        consumer = Consumer(channel, schedule_queue)
        consumer.register_callback(handle_message)
        consumer.consume()
        
        logger.info("PypoFetch: init complete")


    def set_export_source(self, export_source):
        self.export_source = export_source
        self.cache_dir = config["cache_dir"] + self.export_source + '/'

    def check_matching_timezones(self, server_timezone):
        logger = logging.getLogger('fetch')

        process = Popen(["date", "+%z"], stdout=PIPE)
        pypo_timezone = (process.communicate()[0]).strip(' \r\n\t')

        if server_timezone != pypo_timezone:
            logger.error("Server and pypo timezone offsets do not match. Audio playback may not start when expected!")
            logger.error("Server timezone offset: %s", server_timezone)
            logger.error("Pypo timezone offset: %s", pypo_timezone)
    
    """
    Process the schedule
     - Reads the scheduled entries of a given range (actual time +/- "prepare_ahead" / "cache_for")
     - Saves a serialized file of the schedule
     - playlists are prepared. (brought to liquidsoap format) and, if not mounted via nsf, files are copied
       to the cache dir (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss)
     - runs the cleanup routine, to get rid of unused cashed files
    """
    def process_schedule(self, schedule_data, export_source):
        logger = logging.getLogger('fetch')
        self.schedule = schedule_data["playlists"]

        self.check_matching_timezones(schedule_data["server_timezone"])
            
        # Push stream metadata to liquidsoap
        # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
        stream_metadata = schedule_data['stream_metadata']
        try:
            tn = telnetlib.Telnet(LS_HOST, LS_PORT)
            #encode in latin-1 due to telnet protocol not supporting utf-8
            tn.write(('vars.stream_metadata_type %s\n' % stream_metadata['format']).encode('latin-1'))<commit_msg>Better error message when timezones dont match up between pypo and airtime.<commit_after>import os
import sys
import time
import logging
import logging.config
import shutil
import pickle
import random
import string
import json
import telnetlib
import math
from threading import Thread
from subprocess import Popen, PIPE

# For RabbitMQ
from kombu.connection import BrokerConnection
from kombu.messaging import Exchange, Queue, Consumer, Producer

from api_clients import api_client
from util import CueFile

from configobj import ConfigObj

# configure logging
logging.config.fileConfig("logging.cfg")

# loading config file
try:
    config = ConfigObj('config.cfg')
    LS_HOST = config['ls_host']
    LS_PORT = config['ls_port']
    POLL_INTERVAL = int(config['poll_interval'])

except Exception, e:
    print 'Error loading config file: ', e
    sys.exit()

# Yuk - using a global, i know!
SCHEDULE_PUSH_MSG = []

"""
Handle a message from RabbitMQ, put it into our yucky global var.
Hopefully there is a better way to do this.
"""
def handle_message(body, message):
    logger = logging.getLogger('fetch')
    global SCHEDULE_PUSH_MSG
    logger.info("Received schedule from RabbitMQ: " + message.body)
    SCHEDULE_PUSH_MSG = json.loads(message.body)
    # ACK the message to take it off the queue
    message.ack()


class PypoFetch(Thread):
    def __init__(self, q):
        Thread.__init__(self)
        logger = logging.getLogger('fetch')
        self.api_client = api_client.api_client_factory(config)
        self.cue_file = CueFile()
        self.set_export_source('scheduler')
        self.queue = q

        logger.info("Initializing RabbitMQ stuff")
        schedule_exchange = Exchange("airtime-schedule", "direct", durable=True, auto_delete=True)
        schedule_queue = Queue("pypo-fetch", exchange=schedule_exchange, key="foo")
        self.connection = BrokerConnection(config["rabbitmq_host"], config["rabbitmq_user"], config["rabbitmq_password"], "/")
        channel = self.connection.channel()
        consumer = Consumer(channel, schedule_queue)
        consumer.register_callback(handle_message)
        consumer.consume()
        
        logger.info("PypoFetch: init complete")


    def set_export_source(self, export_source):
        self.export_source = export_source
        self.cache_dir = config["cache_dir"] + self.export_source + '/'

    def check_matching_timezones(self, server_timezone):
        logger = logging.getLogger('fetch')

        process = Popen(["date", "+%z"], stdout=PIPE)
        pypo_timezone = (process.communicate()[0]).strip(' \r\n\t')

        if server_timezone != pypo_timezone:
            logger.error("ERROR: Airtime server and pypo timezone offsets do not match. Audio playback will not start when expected!!!")
            logger.error("  * Server timezone offset: %s", server_timezone)
            logger.error("  * Pypo timezone offset: %s", pypo_timezone)
            logger.error("  * To fix this, you need to set the 'date.timezone' value in your php.ini file and restart apache.")
            logger.error("  * See this page for more info (v1.7): http://wiki.sourcefabric.org/x/BQBF")
            logger.error("  * and also the 'FAQ and Support' page underneath it.")  
    
    """
    Process the schedule
     - Reads the scheduled entries of a given range (actual time +/- "prepare_ahead" / "cache_for")
     - Saves a serialized file of the schedule
     - playlists are prepared. (brought to liquidsoap format) and, if not mounted via nsf, files are copied
       to the cache dir (Folder-structure: cache/YYYY-MM-DD-hh-mm-ss)
     - runs the cleanup routine, to get rid of unused cashed files
    """
    def process_schedule(self, schedule_data, export_source):
        logger = logging.getLogger('fetch')
        self.schedule = schedule_data["playlists"]

        self.check_matching_timezones(schedule_data["server_timezone"])
            
        # Push stream metadata to liquidsoap
        # TODO: THIS LIQUIDSOAP STUFF NEEDS TO BE MOVED TO PYPO-PUSH!!!
        stream_metadata = schedule_data['stream_metadata']
        try:
            tn = telnetlib.Telnet(LS_HOST, LS_PORT)
            #encode in latin-1 due to telnet protocol not supporting utf-8
            tn.write(('vars.stream_metadata_type %s\n' % stream_metadata['format']).encode('latin-1'))
{'fraction_non_alphanumeric': 0.09873264936632468, 'fraction_numerical': 0.0012070006035003018, 'mean_word_length': 3.3610526315789473, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7365255', 'n_tokens_mistral': 2478, 'n_tokens_neox': 2398, 'n_words': 695}