scheduler.py

#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 2017-2020 - The Aura Engine Team.

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program 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 Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


import logging
import threading
import time


from src.base.config            import AuraConfig
from src.base.utils             import SimpleUtil as SU
from src.scheduling.models      import AuraDatabaseModel
from src.base.exceptions        import NoActiveTimeslotException, LoadSourceException
from src.control                import EngineExecutor
from src.engine                 import Engine
from src.channels               import ChannelType, TransitionType, EntryPlayState
from src.resources              import ResourceClass, ResourceUtil

from src.scheduling.utils       import TimeslotRenderer
from src.scheduling.programme   import ProgrammeService



class AuraScheduler(threading.Thread):
    """
    Aura Scheduler Class

    - Retrieves data from Steering and Tank
    - Executes engine actions in an automated fashion

    """
    config = None
    logger = None
    engine = None
    exit_event = None
    timeslot_renderer = None
    programme = None
    message_timer = []
    fallback = None
    is_initialized = None
    is_initialized = None



    def __init__(self, engine, fallback_manager):
        """
        Constructor

        Args:
            config (AuraConfig):        Reads the engine configuration
            engine (Engine):            The engine to play the schedule on
            func_on_init (Function):    The function to be called when the scheduler is initialized
        """
        self.config = AuraConfig.config()
        self.logger = logging.getLogger("AuraEngine")
        self.programme = ProgrammeService()
        self.timeslot_renderer = TimeslotRenderer(self)
        self.fallback = fallback_manager
        self.engine = engine
        self.engine.scheduler = self
        self.is_soundsytem_init = False

        # Scheduler Initialization
        AuraDatabaseModel.init_database()
        self.is_initialized = False
        self.is_engine_ready = False

        # Init scheduling thread
        threading.Thread.__init__(self)
        self.exit_event = threading.Event()
        self.start()



    def run(self):
        """
        Called when thread is started via `start()`. It does the following:

            1. `self.fetch_new_programme()` periodically from the API depending on the `fetching_frequency` defined in the engine configuration.
            2. Loads the latest programme from the database and sets the instance state `self.programme` with current timeslots.
            3. Queues all timeslots of the programme, if the soundssystem is ready to accept commands.

        On every cycle the configuration file is reloaded, to allow modifications while running the engine.
        """
        while not self.exit_event.is_set():
            try:
                self.config.load_config()
                seconds_to_wait = int(self.config.get("fetching_frequency"))
                self.logger.info(SU.cyan(f"== start fetching new timeslots (every {seconds_to_wait} seconds) =="))

                # Load some stuff from the API in any case
                self.programme.refresh()

                # Queue only when the engine is ready to play
                if self.is_initialized == True:
                    self.queue_programme()

            except Exception as e:
                self.logger.critical(SU.red(f"Unhandled error while fetching & scheduling new programme! ({str(e)})"))
                # Keep on working anyway

            EngineExecutor.log_commands()
            self.exit_event.wait(seconds_to_wait)



    #
    #   EVENT HANDLERS
    #


    def on_ready(self):
        """
        Called when the engine has finished booting and is ready to play.
        """
        self.is_initialized = True
        self.logger.info(self.timeslot_renderer.get_ascii_timeslots())

        try:
            self.play_active_entry()
            self.queue_startup_entries()
        except NoActiveTimeslotException:
            # That's not good, but keep on working...
            pass


    def on_play(self, entry):
        """
        Event Handler which is called by the engine when some entry is actually playing.
        Ignores entries which are part of a scheduled fallback, because they handle their
        stuff by themselves.

        Args:
            source (String):    The `PlaylistEntry` object
        """
        if entry.channel in ChannelType.FALLBACK_QUEUE.channels:
            return

        current_timeslot = self.programme.get_current_timeslot()
        if current_timeslot:
            current_timeslot.set_active_entry(entry)



    #
    #   METHODS
    #


    def get_programme(self):
        """
        Returns the current programme.
        """
        return self.programme


    def play_active_entry(self):
        """
        Plays the entry scheduled for the very current moment and forwards to the scheduled position in time.
        Usually called when the Engine boots.

        Raises:
            (NoActiveTimeslotException):    If there's no timeslot in the programme, within the scheduling window
        """
        sleep_offset = 10
        active_timeslot = self.programme.get_current_timeslot()

        # Schedule any available fallback playlist
        if active_timeslot:
            # Create command timer to indicate the start of the timeslot
            TimeslotCommand(self.engine, active_timeslot)
            self.fallback.queue_fallback_playlist(active_timeslot)

        active_entry = self.programme.get_current_entry()
        if not active_entry:
            raise NoActiveTimeslotException

        # In case of a file-system source, we need to fast-foward to the current marker as per timeslot
        if active_entry.get_content_type() in ResourceClass.FILE.types:

            # Calculate the seconds we have to fast-forward
            now_unix = Engine.engine_time()
            seconds_to_seek = now_unix - active_entry.start_unix

            # If the seek exceeds the length of the current track,
            # there's no need to do anything - the scheduler takes care of the rest
            if (seconds_to_seek + sleep_offset) > active_entry.duration:
                self.logger.info("The FFWD [>>] range exceeds the length of the entry. Drink some tea and wait for the sound of the next entry.")
            else:
                # Preload and play active entry
                PlayCommand(self.engine, [active_entry])

                # Fast-forward to the scheduled position
                if seconds_to_seek > 0:
                    # Without plenty of timeout (10s) the seek doesn't work
                    def async_cue_seek(seconds_to_seek):
                        seconds_to_seek += sleep_offset
                        time.sleep(sleep_offset)
                        self.logger.info("Going to fast-forward %s seconds" % seconds_to_seek)
                        response = self.engine.player.queue_seek(active_entry.channel, seconds_to_seek)
                        self.logger.info("Sound-system seek response: " + response)

                    thread = threading.Thread(target = async_cue_seek, args = (seconds_to_seek,))
                    thread.start()

        elif active_entry.get_content_type() in ResourceClass.STREAM.types \
            or active_entry.get_content_type() in ResourceClass.LIVE.types:

            # Preload and play active entry
            PlayCommand(self.engine, [active_entry])

        else:
            self.logger.critical("Unknown Entry Type: %s" % active_entry)



    def get_active_playlist(self):
        """
        Retrieves the currently playing playlist.

        Returns:
            (FallbackType, Playlist): The resolved playlist
        """
        timeslot = self.programme.get_current_timeslot()
        if timeslot:
            return self.fallback.resolve_playlist(timeslot)
        return (None, None)



    def queue_programme(self):
        """
        Queues the current programme (playlists as per timeslot) by creating
        timed commands to the sound-system to enable the individual tracks of playlists.
        """

        # Get a clean set of the timeslots within the scheduling window
        timeslots = self.programme.get_next_timeslots()
        timeslots = self.filter_scheduling_window(timeslots)

        # Queue the timeslots, their playlists and entries
        if timeslots:
            for next_timeslot in timeslots:
                # Create command timer to indicate the start of the timeslot
                TimeslotCommand(self.engine, next_timeslot)
                # Schedule any available fallback playlist
                self.fallback.queue_fallback_playlist(next_timeslot)

                playlist = self.programme.get_current_playlist(next_timeslot)
                if playlist:
                    self.queue_playlist_entries(next_timeslot, playlist.entries, False, True)

        self.logger.info(SU.green("Finished queuing programme."))



    def queue_startup_entries(self):
        """
        Queues all entries after the one currently playing upon startup. Don't use
        this method in any other scenario, as it doesn't respect the scheduling window.
        """
        current_timeslot = self.programme.get_current_timeslot()

        # Queue the (rest of the) currently playing timeslot upon startup
        if current_timeslot:
            current_playlist = self.programme.get_current_playlist(current_timeslot)

            if current_playlist:
                active_entry = self.programme.get_current_entry()
                if active_entry:
                    # Queue open entries for current playlist
                    rest_of_playlist = active_entry.get_next_entries(True)
                    self.queue_playlist_entries(current_timeslot, rest_of_playlist, False, True)



    def queue_playlist_entries(self, timeslot, entries, fade_in, fade_out):
        """
        Creates sound-system player commands for all playlist items to be executed at the scheduled time.

        Since each scheduled playlist can consist of multiple entry types such as *file*, *live*,
        and *stream*, the play-out of the timeslot is actually a bit more complex. Before any playlist
        entries of the timeslot can be turned into sound, they need to be aggregated, queued and pre-loaded.

        1. First, all entries are aggregated when they hold filesystem entries.
            Given you have a playlist with 10 entries, the first 4 are consisting of files, the next two
            of a a stream and a live source. The last 4 are files again. These entries are now
            aggregated into 4 groups: one for the files, one for the stream, one for the live entry
            and another one for files. For each group a timer for executing the next step is created.

        2. Now, the playlist entries are going to be "pre-loaded". This means that filesystem
            entries are queued and pre-loaded and entries which are based on audio streams are buffered.
            This is required to allow a seamless play-out, when its time to do so (in the next stage).
            Due to their nature, playlist entries which hold live audio sources are not affected by
            this stage at all.

        Args:
            timeslot (Timeslot):        The timeslot this entries belong to
            entries ([PlaylistEntry]):  The playlist entries to be scheduled for playout
            fade_in (Boolean):          Fade-in at the beginning of the set of entries
            fade_out (Boolean):         Fade-out at the end of the set of entries

        Returns:
            (String):                   Formatted string to display playlist entries in log
        """
        entry_groups = []
        entry_groups.append([])
        previous_entry = None
        index = 0

        # Mark entries which start after the end of their timeslot or are cut
        # clean_entries = self.preprocess_entries(entries, True)

        # Group/aggregate all filesystem entries, allowing them to be queued at once
        for entry in entries:
            if previous_entry == None or \
                (previous_entry != None and \
                 previous_entry.get_content_type() == entry.get_content_type() and \
                 entry.get_content_type() in ResourceClass.FILE.types):

                entry_groups[index].append(entry)
            else:
                index += 1
                entry_groups.append([])
                entry_groups[index].append(entry)
            previous_entry = entry
        self.logger.info("Built %s entry group(s)" % len(entry_groups))

        # Timeslot function calls
        if len(entries) > 0 and len(entry_groups) > 0:
            for entries in entry_groups:
                if not isinstance(entries, list):
                    raise ValueError("Invalid Entry Group: %s" % str(entries))

                # Create command timers for each entry group
                PlayCommand(self.engine, entries)
        else:
            self.logger.warn(SU.red("Nothing to schedule ..."))



    def filter_scheduling_window(self, timeslots):
        """
        Ignore timeslots which are before the start of scheduling window (start of timeslot - `scheduling_window_start`)
        or after the end of the scheduling window (end of timeslot -`scheduling_window_end`).

        Before the scheduling window:
            - Timeslots can still be deleted in Steering and the playout will respect this

        During the scheduling window:
            - Timeslots and it's playlists are queued as timed commands

        After the scheduling window:
            - Such timeslots are ignored, because it doesn't make sense anymore to schedule them before the next
              timeslot starts

        """
        if not timeslots:
            return timeslots

        now_unix = Engine.engine_time()
        len_before = len(timeslots)
        window_start = self.config.get("scheduling_window_start")
        window_end = self.config.get("scheduling_window_end")
        timeslots = list(filter(lambda t: (t.start_unix - window_start) < now_unix and now_unix < (t.end_unix - window_end), timeslots))
        len_after = len(timeslots)
        self.logger.info("For now, skipped %s future timeslot(s) which are out of the scheduling window (T¹-%ss to T²-%ss)" % ((len_before - len_after), window_start, window_end))

        return timeslots



    def terminate(self):
        """
        Called when thread is stopped or a signal to terminate is received.
        """
        self.logger.info("Shutting down scheduler ...")
        self.programme.terminate()
        self.exit_event.set()





#
#   EngineExecutor Commands
#


class TimeslotCommand(EngineExecutor):
    """
    Command for triggering start and end of timeslot events.
    """
    engine = None
    config = None

    def __init__(self, engine, timeslot):
        """
        Constructor

        Args:
            engine (Engine):        The engine
            timeslot (Timeslot):    The timeslot which is starting at this time
        """
        self.config = AuraConfig()
        self.engine = engine

        fade_out_time = float(self.config.get("fade_out_time"))
        start_fade_out = timeslot.end_unix - fade_out_time
        self.logger.info(f"Fading out timeslot in {start_fade_out} seconds at {timeslot.timeslot_end} | Timeslot: {timeslot}")
        # Initialize the "fade in" EngineExecuter and instatiate a connected child EngineExecuter for "fade out" when the parent is ready
        super().__init__("TIMESLOT", None, timeslot.start_unix, self.do_start_timeslot, timeslot)
        self.on_ready(lambda: EngineExecutor("TIMESLOT", self, start_fade_out, self.do_end_timeslot, timeslot))


    def do_start_timeslot(self, timeslot):
        """
        Initiates the start of the timeslot.
        """
        self.logger.info(SU.cyan(f"=== on_timeslot_start('{timeslot}') ==="))
        self.engine.event_dispatcher.on_timeslot_start(timeslot)


    def do_end_timeslot(self, timeslot):
        """
        Initiates the end of the timeslot.
        """
        self.logger.info(SU.cyan(f"=== on_timeslot_end('{timeslot}') ==="))
        self.engine.event_dispatcher.on_timeslot_end(timeslot)

        recent_entry = timeslot.get_recent_entry()
        if recent_entry:
            self.engine.player.stop(recent_entry, TransitionType.FADE)
        else:
            self.logger.warning(SU.red(f"Interestingly timeslot {timeslot} has no entry to be faded out?"))



class PlayCommand(EngineExecutor):
    """
    Command for triggering timed preloading and playing as a child command.
    """
    engine = None
    config = None

    def __init__(self, engine, entries):
        """
        Constructor

        Args:
            engine (Engine):            The engine
            entries (PlaylistEntry):    One or more playlist entries to be started
        """
        self.config = AuraConfig()
        self.engine = engine

        start_preload = entries[0].start_unix - self.config.get("preload_offset")
        start_play = entries[0].start_unix
        # Initialize the "preload" EngineExecuter and attach a child `PlayCommand` to the "on_ready" event handler
        preload_timer = super().__init__("PRELOAD", None, start_preload, self.do_preload, entries)
        self.on_ready(lambda: EngineExecutor("PLAY", self, start_play, self.do_play, entries))


    def do_preload(self, entries):
        """
        Preload the entries.
        """
        try:
            if entries[0].get_content_type() in ResourceClass.FILE.types:
                self.logger.info(SU.cyan("=== preload_group('%s') ===" % ResourceUtil.get_entries_string(entries)))
                self.engine.player.preload_group(entries, ChannelType.QUEUE)
            else:
                self.logger.info(SU.cyan("=== preload('%s') ===" % ResourceUtil.get_entries_string(entries)))
                self.engine.player.preload(entries[0])
        except LoadSourceException as e:
            self.logger.critical(SU.red("Could not preload entries %s" % ResourceUtil.get_entries_string(entries)), e)

        if entries[-1].status != EntryPlayState.READY:
            self.logger.critical(SU.red("Entries didn't reach 'ready' state during preloading (Entries: %s)" % ResourceUtil.get_entries_string(entries)))


    def do_play(self, entries):
        """
        Play the entries.
        """
        self.logger.info(SU.cyan("=== play('%s') ===" % ResourceUtil.get_entries_string(entries)))
        if entries[-1].status != EntryPlayState.READY:
            # Let 'em play anyway ...
            self.logger.critical(SU.red("PLAY: The entry/entries are not yet ready to be played (Entries: %s)" % ResourceUtil.get_entries_string(entries)))
            while (entries[-1].status != EntryPlayState.READY):
                self.logger.info("PLAY: Wait a little until preloading is done ...")
                time.sleep(2)

        self.engine.player.play(entries[0], TransitionType.FADE)
        self.logger.info(self.engine.scheduler.timeslot_renderer.get_ascii_timeslots())