calendar.py

#
#  engine
#
#  Playout Daemon for autoradio project
#
#
#  Copyright (C) 2017-2018 Gottfried Gaisbauer <gottfried.gaisbauer@servus.at>
#
#  This file is part of engine.
#
#  engine is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  any later version.
#
#  engine is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with engine. If not, see <http://www.gnu.org/licenses/>.
#

import os
import sys
import threading
import json
import queue
import traceback
import urllib
import logging

from mutagen.flac import FLAC
from datetime import datetime, timedelta

from libraries.database.broadcasts import Schedule, ScheduleEntry
from libraries.enum.auraenumerations import ScheduleEntryType
from modules.communication.redis.messenger import RedisMessenger


class AuraCalendarService(threading.Thread):
    messenger = None
    until = ""
    playlistdir = ""
    xmlplaylist = range(0)

    has_already_fetched = False
    queue = None
    config = None
    debug = False
    _stop_event = None
    logger = None
    url = dict()
    data = dict()
    # another crutch because of the missing TANK
    used_random_playlist_ids = list()

    """
    Fetching playlist data, write it into the database and notify service
    """
    def __init__(self, config, datefrom="", dateto=""):
        threading.Thread.__init__(self)
        self.config = config
        self.messenger = RedisMessenger(config)
        self.logger = logging.getLogger("AuraEngine")

        self.messenger.set_channel("aura")
        self.messenger.set_section("calendar")

        self.datefrom = str(datefrom)
        self.dateto = dateto

        self.queue = queue.Queue()

        self._stop_event = threading.Event()

        self.__set_url__("calendar")
        self.__set_url__("importer")

    # ------------------------------------------------------------------------------------------ #
    def set_date_from(self, date):
        self.datefrom = str(date).replace(" ", "T")

    # ------------------------------------------------------------------------------------------ #
    def set_date_to(self, date):
        self.dateto = str(date).replace(" ", "T")

    # ------------------------------------------------------------------------------------------ #
    def set_until_time(self, timestring):
        self.until = timestring

    # ------------------------------------------------------------------------------------------ #
    def set_playlist_store(self, path):
        self.playlistdir = path

    # ------------------------------------------------------------------------------------------ #
    def get_duration(self, start, end):
        return self.__calc_duration__(start, end)

    # ------------------------------------------------------------------------------------------ #
    def get_queue(self):
        return self.queue

    # ------------------------------------------------------------------------------------------ #
    def get_uri(self):
        if not self.playlistdir:
            return False
        if not self.datefrom:
            return False
        if not self.__calc_date_to__():
            return

        hostname = self.get("servername");
        port = self.get("serviceport");
        date_from = self.datefrom[0:16] + ":00";
        date_to = self.dateto[0:16] + ":00";
        uri = "http://" + hostname  + ":" + port + "/playlist/" + date_from + "/" + date_to

        return uri

    # ------------------------------------------------------------------------------------------ #
    def run(self):
        """
        Fetch calendar data and store it in the database
        """

        try:
            # fetch upcoming schedules from STEERING
            self.logger.debug("Fetching schedules from STEERING")
            self.__fetch_schedule_data__()
            # fetch playlist and fallbacks to the schedules from TANK
            self.logger.debug("Fetching playlists from TANK")
            self.__fetch_schedule_entry_data__()

            # drop everything what is more than 30 minutes in the future to avoid strange sync errors
            # the programme is still in the memory of engine and reloaded, when this fetching is finished.
            self.drop_the_future(timedelta(minutes=30))

            for schedule in self.fetched_schedule_data:
                if "start" not in schedule:
                    self.logger.warning("No start of schedule given. skipping the schedule: "+str(schedule))
                    continue
                if "end" not in schedule:
                    self.logger.warning("No end of schedule given. skipping the schedule: "+str(schedule))
                    continue

                # store the schedule
                schedule_db = self.store_schedule(schedule)

                # store playlists to play
                self.store_schedule_playlist(schedule_db, schedule, "playlist")
                self.store_schedule_playlist(schedule_db, schedule, "schedule_fallback", 1)
                self.store_schedule_playlist(schedule_db, schedule, "show_fallback", 2)
                self.store_schedule_playlist(schedule_db, schedule, "station_fallback", 3)

            # release the mutex
            self.queue.put(schedule) #"fetching_finished")
        except Exception as e:
            self.queue.put("fetching_aborted " + str(e))

        # terminate the thread
        return

    # ------------------------------------------------------------------------------------------ #
    def drop_the_future(self, time_in_the_future):
        ScheduleEntry.drop_the_future(time_in_the_future)
        Schedule.drop_the_future(time_in_the_future)

    # ------------------------------------------------------------------------------------------ #
    def store_schedule(self, schedule):
        schedule_db = Schedule.select_show_on_datetime(schedule["start"])
        havetoadd = False

        if not schedule_db:
            self.logger.debug("no schedule with given schedule id in database => create new")
            schedule_db = Schedule()
            havetoadd = True

        # calc duration
        duration = self.__calc_duration__(schedule["start"], schedule["end"])
        schedule["duration"] = timedelta(seconds=duration).__str__()

        schedule_db.show_id = schedule["show_id"]
        schedule_db.schedule_id = schedule["schedule_id"]
        schedule_db.schedule_start = schedule["start"]
        schedule_db.schedule_end = schedule["end"]
        schedule_db.show_name = schedule["show_name"]
        schedule_db.show_hosts = schedule["show_hosts"]
        schedule_db.is_repetition = schedule["is_repetition"]
        schedule_db.funding_category = schedule["show_fundingcategory"]
        schedule_db.languages = schedule["show_languages"]
        schedule_db.type = schedule["show_type"]
        schedule_db.category = schedule["show_categories"]
        schedule_db.topic = schedule["show_topics"]
        schedule_db.musicfocus = schedule["show_musicfocus"]

        schedule_db.playlist_id = schedule["playlist_id"]
        schedule_db.schedule_fallback_id = schedule["schedule_fallback_id"]
        schedule_db.show_fallback_id = schedule["show_fallback_id"]
        schedule_db.station_fallback_id = schedule["station_fallback_id"]

        schedule_db.store(add=havetoadd, commit=True)

        return schedule_db

    # ------------------------------------------------------------------------------------------ #
    def store_schedule_playlist(self, schedule_db, schedule, playlistname, fallbackplaylist_type=0):
        playlist = schedule[playlistname]

        debug = "Schedule playlist (" + playlistname + ") for " + schedule_db.show_name + " stored"
        warning = "No scheduleentries for playlist #" + str(playlist['playlist_id']) + " in schedule #" + str(schedule_db.schedule_id) + " found"
        entrynum = 0

        if "entries" in playlist:
            lastentry = None

            for entry in playlist["entries"]:
                lastentry = self.store_playlist_entry(schedule_db, playlist, entry, lastentry, entrynum, fallbackplaylist_type)
                entrynum = entrynum + 1

            if lastentry is None:
                self.logger.warning(warning)
            else:
                self.logger.debug(debug)
        else:
            self.logger.warning(warning)

    # ------------------------------------------------------------------------------------------ #
    def store_playlist_entry(self, schedule_db, playlist, entry, lastentry, entrynum, fallbackplaylist_type=0):
        schedule_entry_db = ScheduleEntry.select_one_playlist_entry_for_show(schedule_db.schedule_id, fallbackplaylist_type, entrynum)
        havetoadd = False

        if not schedule_entry_db:
            self.logger.debug("no scheduleentry with id " + str(playlist["playlist_id"]) + " and pos " + str(entrynum) + " in database => creating a new one")
            schedule_entry_db = ScheduleEntry()
            havetoadd = True

        schedule_entry_db.playlist_id = playlist["playlist_id"]
        schedule_entry_db.entry_num = entrynum
        schedule_entry_db.schedule_id = schedule_db.schedule_id
        schedule_entry_db.source = entry["source"]
        schedule_entry_db.fallback_type = fallbackplaylist_type
        schedule_entry_db.entry_start = schedule_db.schedule_start + timedelta(seconds=self.get_length(lastentry))

        schedule_entry_db.calc_unix_times()
        if havetoadd:
            schedule_entry_db.define_clean_source()

        self.logger.debug("Storing entries... playlist_id: " + str(playlist["playlist_id"]) + " schedule_id: " + str(schedule_db.schedule_id) + " num: " + str(entrynum))

        schedule_entry_db.store(add=havetoadd, commit=True)

        return schedule_entry_db

    # ------------------------------------------------------------------------------------------ #
    def __calc_date_to__(self):
        if self.dateto:
            return True
        if not self.until:
            return False
        if not self.datefrom:
            return False

        date_start = datetime.strptime(self.datefrom.replace("T"," "), "%Y-%m-%d %H:%M:%S")
        time_start = date_start.strftime("%H:%M")
        day_offset = 1 if (time_start > self.until) else 0
        end_date = date_start + timedelta(day_offset)
        self.dateto = end_date.strftime("%F") + "T" + self.until

        return True

    # ------------------------------------------------------------------------------------------ #
    @staticmethod
    def __calc_duration__(start, end):
        """
        Berechnet Zeit in Sekunden aus Differenz zwischen Start und Enddatum
        @type start: datetime
        @param start: Startzeit
        @type end: datetime
        @param end: Endzeit
        @rtype:       int
        @return:      Zeit in Sekunden
        """
        sec1 = int(datetime.strptime(start[0:16].replace(" ","T"),"%Y-%m-%dT%H:%M").strftime("%s"));
        sec2 = int(datetime.strptime(end[0:16].replace(" ","T"),"%Y-%m-%dT%H:%M").strftime("%s"));
        return (sec2 - sec1);

    # ------------------------------------------------------------------------------------------ #
    def __fetch_schedule_entry_data__(self):
        # store fetched entries => do not have to fetch playlist_id more than once
        fetched_entries=[]

        try:
            for schedule in self.fetched_schedule_data:
                # retrieve playlist and the fallbacks for every schedule
                # if a playlist (like station_fallback) is already fetched, it is not fetched again but reused
                schedule["playlist"]          = self.__fetch_schedule_entries__(schedule, "playlist_id",          fetched_entries)
                schedule["schedule_fallback"] = self.__fetch_schedule_entries__(schedule, "schedule_fallback_id", fetched_entries)
                schedule["show_fallback"]     = self.__fetch_schedule_entries__(schedule, "show_fallback_id",     fetched_entries)
                schedule["station_fallback"]  = self.__fetch_schedule_entries__(schedule, "station_fallback_id",  fetched_entries)

                self.logger.info(str(schedule))

        except Exception as e:
            self.logger.error(str(e))

    # ------------------------------------------------------------------------------------------ #
    def __fetch_schedule_entries__(self, schedule, id_name, fetched_schedule_entries):
        servicetype = "importer"
        use_testdata = False

        # fetch data from importer
        json_response = self.__fetch_data__(servicetype)
        if not json_response and self.config.get("use_test_data"):
            use_testdata = True

        # if a playlist is already fetched, do not fetch it again
        for entry in fetched_schedule_entries:
            if entry["playlist_id"] == schedule[id_name]:
                self.logger.debug("playlist #" + str(schedule[id_name]) + " already fetched")
                return entry

        # generate testdata
        if use_testdata:
            json_response = self.create_test_data(id_name, schedule)

        # convert to list
        try:
            schedule_entries = json.loads(json_response)
        except Exception as e:
            self.logger.critical("Cannot convert playlist from importer into list")
            schedule_entries = list()

        if "entries" in schedule_entries:
            for entry in schedule_entries["entries"]:
                if entry["source"].startswith("file"):
                    e = entry["source"][7:] # filter file:// out
                    if not os.path.isfile(e):
                        self.logger.warning("File", e, "does not exist!")

            fetched_schedule_entries.append(schedule_entries)

        return schedule_entries

    def create_test_data(self, id_name, schedule):
        import random
        rand_id = random.randint(1, 10000)

        while rand_id in self.used_random_playlist_ids:
            rand_id = random.randint(1, 10000)

        self.used_random_playlist_ids.append(rand_id)

        # HARDCODED Testdata
        if id_name != "playlist_id":
            # FALLBACK TESTDATA

            if rand_id % 3 == 0:  # playlist fallback
                json_response = '{"playlist_id":' + str(
                    rand_id) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"file:///var/audio/fallback/NightmaresOnWax/DJ-Kicks/02 - Only Child - Breakneck.flac"}]}'
            elif rand_id % 2 == 0:  # stream fallback
                json_response = '{"playlist_id":' + str(
                    rand_id) + ',"entries":[{"source":"http://chill.out.airtime.pro:8000/chill_a"}]}'
            else:  # pool fallback
                json_response = '{"playlist_id":' + str(rand_id) + ',"entries":[{"source":"pool:///liedermacherei"}]}'

            schedule[id_name] = rand_id

        elif schedule[id_name] == 0 or schedule[id_name] is None:
            # this happens when playlist id is not filled out in pv
            # json_response = '{"playlist_id": 0}'

            if rand_id % 4 == 0:  # playlist with two files
                json_response = '{"playlist_id":' + str(
                    rand_id) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"file:///var/audio/fallback/NightmaresOnWax/DJ-Kicks/02 - Only Child - Breakneck.flac"}]}'
            elif rand_id % 3 == 0:  # playlist with jingle and then linein
                json_response = '{"playlist_id":' + str(
                    rand_id) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"linein://1"}]}'
            elif rand_id % 2 == 0:  # playlist with jingle and then http stream
                json_response = '{"playlist_id":' + str(
                    rand_id) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"http://chill.out.airtime.pro:8000/chill_a"}]}'
            else:  # pool playlist
                json_response = '{"playlist_id":' + str(rand_id) + ',"entries":[{"source":"pool:///hiphop"}]}'

            schedule[id_name] = rand_id

        elif schedule[id_name] % 4 == 0:  # playlist with two files
            json_response = '{"playlist_id":' + str(schedule[id_name]) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"file:///var/audio/fallback/NightmaresOnWax/DJ-Kicks/01 - Type - Slow Process.flac"}]}'
        elif schedule[id_name] % 3 == 0:  # playlist with jingle and then http stream
            json_response = '{"playlist_id":' + str(schedule[id_name]) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"linein://0"}]}'
        elif schedule[id_name] % 2 == 0:  # playlist with jingle and then linein
            json_response = '{"playlist_id":' + str(schedule[id_name]) + ',"entries":[{"source":"file:///var/audio/fallback/music.flac"},{"source":"http://stream.fro.at:80/fro-128.ogg"}]}'
        else:  # pool playlist
            json_response = '{"playlist_id":' + str(schedule[id_name]) + ',"entries":[{"source":"pool:///chillout"}]}'

        self.logger.info("Using 'randomized' playlist: " + json_response + " for " + id_name[:-3] + " for show " + schedule["show_name"] + " starting @ " + schedule["start"])

        return json_response

    # ------------------------------------------------------------------------------------------ #
    def __fetch_schedule_data__(self):
        servicetype = "calendar"
        use_testdata = False

        html_response = self.__fetch_data__(servicetype)
        if not html_response or html_response == b"[]":
            self.logger.debug("Got no response: Using testdata")
            use_testdata = True

        # if an error occours => use testdata
        if use_testdata:
            html_response = '[{"schedule_id":1,"start":"' + (datetime.now() + timedelta(hours=0)).strftime('%Y-%m-%d %H:00:00') + '","end":"' + (datetime.now() + timedelta(hours=1)).strftime('%Y-%m-%d %H:00:00') + '","show_id":9,"show_name":"FROzine","show_hosts":"Sandra Hochholzer, Martina Schweiger","is_repetition":false,"playlist_id":2,"schedule_fallback_id":12,"show_fallback_id":92,"station_fallback_id":1,"rtr_category":"string","comment":"Kommentar","languages":"Sprachen","type":"Typ","category":"Kategorie","topic":"Topic","musicfocus":"Fokus"},{"schedule_id":2,"schedule_start":"' + (datetime.now()+timedelta(hours=1)).strftime('%Y-%m-%d %H:00:00') + '","schedule_end":"' + (datetime.now()+timedelta(hours=2)).strftime('%Y-%m-%d %H:00:00') + '","show_id":10,"show_name":"FROMat","show_hosts":"Sandra Hochholzer, Martina Schweiger","is_repetition":false,"playlist_id":4,"schedule_fallback_id":22,"show_fallback_id":102,"station_fallback_id":1,"rtr_category":"string","comment":"Kommentar","languages":"Sprachen","type":"Typ","category":"Kategorie","topic":"Topic","musicfocus":"Fokus"},{"schedule_id":3,"schedule_start":"' + (datetime.now()+timedelta(hours=2)).strftime('%Y-%m-%d %H:00:00') + '","schedule_end":"' + (datetime.now() + timedelta(hours=3)).strftime('%Y-%m-%d %H:00:00') + '","show_id":11,"show_name":"Radio für Senioren","show_hosts":"Sandra Hochholzer, Martina Schweiger","is_repetition":false,"playlist_id":6,"schedule_fallback_id":32,"show_fallback_id":112,"station_fallback_id":1,"rtr_category":"string","comment":"Kommentar","languages":"Sprachen","type":"Typ","category":"Kategorie","topic":"Topic","musicfocus":"Fokus"}]'

        try:
            schedule_from_pv = json.loads(html_response)
        except Exception as e:
            self.logger.critical("Cannot fetch schedule entries from PV")
            sys.exit()

        # check data
        self.logger.critical("Hardcoded Response && no JSON data checks. I believe what i get here")

        d = self.remove_data_more_than_24h_in_the_future(schedule_from_pv)
        self.fetched_schedule_data = self.remove_data_in_the_past(d)

        return self.fetched_schedule_data

    # ------------------------------------------------------------------------------------------ #
    def remove_data_more_than_24h_in_the_future(self, schedule_from_pv):
        act_list = []
        now = datetime.now()
        now_plus_24hours = now + timedelta(hours=24)

        for s in schedule_from_pv:
            date_start = datetime.strptime(s["start"], "%Y-%m-%dT%H:%M:%S")

            # append only elements which are close enough to now
            if date_start <= now_plus_24hours and date_start >= now - timedelta(hours=1):
                act_list.append(s)

        return act_list

    # ------------------------------------------------------------------------------------------ #
    def remove_data_in_the_past(self, schedule_from_pv):
        act_list = []
        now = datetime.now()

        for index,curr in enumerate(schedule_from_pv[:-1]):
            date_start = datetime.strptime(curr["start"], "%Y-%m-%dT%H:%M:%S")
            date_next_start = datetime.strptime(schedule_from_pv[index+1]["start"], "%Y-%m-%dT%H:%M:%S")

            # append all elements in the future
            if date_start >= now:
                act_list.append(curr)
            # append the one which is now playing
            if date_start <= now and date_next_start >= now:
                act_list.append(curr)

        return act_list

    # ------------------------------------------------------------------------------------------ #
    def __fetch_data__(self, type):
        # init html_response
        html_response = ""

        # open an url and read the data
        try:
            if type not in self.data:
                if self.url[type] == "":
                    return False
                request = urllib.request.Request(self.url[type])
            else:
                request = urllib.request.Request(self.url[type], self.data[type])

            response = urllib.request.urlopen(request)
            html_response = response.read()

        except (urllib.error.URLError, IOError, ValueError) as e:
            self.logger.error("Cannot connect to " + self.url[type] + "! reason: " + str(e.reason))
            if not self.has_already_fetched: # first fetch
                sys.exit()

        self.has_already_fetched = True
        return html_response

    # ------------------------------------------------------------------------------------------ #
    def get_length(self, entry):
        if entry is None or entry.type == ScheduleEntryType.STREAM or entry.type == ScheduleEntryType.LIVE_0 or entry.type == ScheduleEntryType.LIVE_1 or entry.type == ScheduleEntryType.LIVE_2 or entry.type == ScheduleEntryType.LIVE_3 or entry.type == ScheduleEntryType.LIVE_4:
            return 0

        audio_file = FLAC(entry.cleansource)
        return audio_file.info.length

    # ------------------------------------------------------------------------------------------ #
    def __set_url__(self, type):
        url = self.config.get(type+"url")
        pos = url.find("?")

        if pos > 0:
            self.url[type] = url[0:pos]
            self.data[type] = url[pos:]
        else:
            self.url[type] = url

    # ------------------------------------------------------------------------------------------ #
    def stop(self):
        self._stop_event.set()