engine.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 time
import logging

from contextlib             import suppress
from threading              import Thread

import meta

from src.base.config        import AuraConfig
from src.base.utils         import SimpleUtil as SU
from src.base.exceptions    import LQConnectionError, InvalidChannelException, LQStreamException, LoadSourceException
from src.resources          import ResourceClass, ResourceUtil
from src.channels           import ChannelType, TransitionType, LiquidsoapResponse, EntryPlayState, ResourceType, ChannelRouter, ChannelResolver
from src.events             import EngineEventDispatcher
from src.control            import EngineControlInterface
from src.mixer              import Mixer, MixerType
from src.client.connector   import PlayerConnector




class Engine():
    """
    The Engine.
    """
    instance = None
    engine_time_offset = 0.0
    logger = None
    eci = None
    channels = None
    channel_router = None
    scheduler = None
    event_dispatcher = None
    plugins = None
    connector = None


    def __init__(self):
        """
        Constructor
        """
        if Engine.instance:
            raise Exception("Engine is already running!")
        Engine.instance = self
        self.logger = logging.getLogger("AuraEngine")
        self.config = AuraConfig.config()
        Engine.engine_time_offset = float(self.config.get("engine_latency_offset"))

        self.plugins = dict()
        self.channel_router = ChannelRouter(self.config, self.logger)
        self.start()



    def start(self):
        """
        Starts the engine. Called when the connection to the sound-system implementation
        has been established.
        """
        self.event_dispatcher = EngineEventDispatcher(self)
        self.eci = EngineControlInterface(self, self.event_dispatcher)
        self.connector = PlayerConnector(self.event_dispatcher)
        self.event_dispatcher.on_initialized()

        while not self.is_connected():
            self.logger.info(SU.yellow("Waiting for Liquidsoap to be running ..."))
            time.sleep(2)
        self.logger.info(SU.green("Engine Core ------[ connected ]-------- Liquidsoap"))

        self.player = Player(self.connector, self.event_dispatcher)
        self.event_dispatcher.on_boot()
        self.logger.info(EngineSplash.splash_screen("Engine Core", meta.__version__))
        self.event_dispatcher.on_ready()



    #
    #   Basic Methods
    #


    def is_connected(self):
        """
        Checks if there's a valid connection to Liquidsoap.
        """
        has_connection = False
        try:
            self.uptime()
            has_connection = True
        except LQConnectionError as e:
            self.logger.info("Liquidsoap is not running so far")
        except Exception as e:
            self.logger.error("Cannot check if Liquidsoap is running. Reason: " + str(e))

        return has_connection


    def engine_state(self):
        """
        Retrieves the state of all inputs and outputs.
        """
        state = self.connector.send_lqc_command("engine", "state")
        return state


    def version(self):
        """
        Get the version of Liquidsoap.
        """
        data = self.connector.send_lqc_command("version", "")
        return data


    def uptime(self):
        """
        Retrieves the uptime of Liquidsoap.
        """
        self.connector.enable_transaction()
        data = self.connector.send_lqc_command("uptime", "")
        self.connector.disable_transaction()
        return data


    @staticmethod
    def engine_time():
        """
        Liquidsoap is slow in executing commands, therefore it's needed to schedule
        actions by (n) seconds in advance, as defined in the configuration file by
        the property `engine_latency_offset`. it's important to note that this method
        requires the class variable `EngineUtil.engine_time_offset` to be set on
        Engine initialization.

        Returns:
            (Integer):  the Unix epoch timestamp including the offset
        """
        return SU.timestamp() + Engine.engine_time_offset


    @staticmethod
    def get_instance():
        """
        Returns the one and only engine.
        """
        return Engine.instance


    def terminate(self):
        """
        Terminates the engine and all related processes.
        """
        if self.eci: self.eci.terminate()



#
#   PLAYER
#


class Player:
    """
    Engine Player.
    """
    config = None
    logger = None
    connector = None
    channels = None
    channel_router = None
    event_dispatcher = None
    mixer = None
    mixer_fallback = None



    def __init__(self, connector, event_dispatcher):
        """
        Constructor

        Args:
            config (AuraConfig):    The configuration
        """
        self.config = AuraConfig.config()
        self.logger = logging.getLogger("AuraEngine")
        self.event_dispatcher = event_dispatcher
        self.connector = connector
        self.channel_router = ChannelRouter(self.config, self.logger)
        self.mixer = Mixer(self.config, MixerType.MAIN, self.connector)
        self.mixer_fallback = Mixer(self.config, MixerType.FALLBACK, self.connector)



    def preload(self, entry):
        """
        Pre-Load the entry. This is required before the actual `play(..)` can happen.

        Be aware when using this method to queue a very short entry (shorter than ``) this may
        result in sitations with incorrect timing. In this case bundle multiple short entries as
        one queue using `preload_playlist(self, entries)`.

        It's important to note, that his method is blocking until loading has finished. If this
        method is called asynchronously, the progress on the preloading state can be looked up in
        `entry.state`.

        Args:
            entries ([Entry]):    An array holding filesystem entries
        """
        entry.status = EntryPlayState.LOADING
        self.logger.info("Loading entry '%s'" % entry)
        is_ready = False

        # LIVE
        if entry.get_content_type() in ResourceClass.LIVE.types:
            entry.channel = ChannelResolver.live_channel_for_resource(entry.source)
            if entry.channel == None:
                self.logger.critical(SU.red("Invalid live channel '{entry.source}' requested!"))
            entry.previous_channel = None
            is_ready = True
        else:
            channel_type = self.channel_router.type_for_resource(entry.get_content_type())
            entry.previous_channel, entry.channel = self.channel_router.channel_swap(channel_type)

        # QUEUE
        if entry.get_content_type() in ResourceClass.FILE.types:
            is_ready = self.queue_push(entry.channel, entry.source)

        # STREAM
        elif entry.get_content_type() in ResourceClass.STREAM.types:
            is_ready = self.stream_load_entry(entry)

        if is_ready:
            entry.status = EntryPlayState.READY

        self.event_dispatcher.on_queue([entry])



    def preload_group(self, entries, channel_type=ChannelType.QUEUE):
        """
        Pre-Load multiple filesystem entries at once. This call is required before the
        actual `play(..)` can happen. Due to their nature, non-filesystem entries cannot be queued
        using this method. In this case use `preload(self, entry)` instead. This method also allows
        queuing of very short files, such as jingles.

        It's important to note, that his method is blocking until loading has finished. If this
        method is called asynchronously, the progress on the preloading state can be looked up in
        `entry.state`.

        Args:
            entries ([Entry]):          An array holding filesystem entries
            channel_type (ChannelType): The type of channel where it should be queued (optional)
        """
        channels = None

        # Validate entry type
        for entry in entries:
            if entry.get_content_type() != ResourceType.FILE:
                raise InvalidChannelException

        # Determine channel
        channels = self.channel_router.channel_swap(channel_type)

        # Queue entries
        for entry in entries:
            entry.status = EntryPlayState.LOADING
            self.logger.info("Loading entry '%s'" % entry)

            # Choose and save the input channel
            entry.previous_channel, entry.channel = channels

            if self.queue_push(entry.channel, entry.source) == True:
                entry.status = EntryPlayState.READY

        self.event_dispatcher.on_queue(entries)
        return channels


    def play(self, entry, transition):
        """
        Plays a new `Entry`. In case of a new timeslot (or some intented, immediate transition),
        a clean channel is selected and transitions between old and new channel is performed.

        This method expects that the entry is pre-loaded using `preload(..)` or `preload_group(self, entries)`
        before being played. In case the pre-roll has happened for a group of entries, only the
        first entry of the group needs to be passed.

        Args:
            entry (PlaylistEntry):          The audio source to be played
            transition (TransitionType):    The type of transition to use e.g. fade-in or instant volume level.
            queue (Boolean):                If `True` the entry is queued if the `ChannelType` does allow so;
                otherwise a new channel of the same type is activated

        """
        with suppress(LQConnectionError):

            channel_type = self.channel_router.type_of_channel(entry.channel)
            mixer = self.mixer
            if channel_type == ChannelType.FALLBACK_QUEUE:
                mixer = self.mixer_fallback

            # Instant activation or fade-in
            self.connector.enable_transaction()
            if transition == TransitionType.FADE:
                mixer.channel_select(entry.channel.value, True)
                mixer.fade_in(entry.channel, entry.volume)
            else:
                mixer.channel_activate(entry.channel.value, True)
            self.connector.disable_transaction()

            # Update active channel for the current channel type
            self.channel_router.set_active(channel_type, entry.channel)

            # Dear filesystem channels, please leave the room as you would like to find it!
            if entry.previous_channel and \
                entry.previous_channel in ChannelType.QUEUE.channels and \
                entry.previous_channel in ChannelType.FALLBACK_QUEUE.channels:

                def clean_up():
                    # Wait a little, if there is some long fade-out. Note, this also means,
                    # this channel should not be used for at least some seconds (including clearing time).
                    time.sleep(2)
                    self.connector.enable_transaction()
                    mixer.channel_activate(entry.previous_channel.value, False)
                    res = self.queue_clear(entry.previous_channel)
                    self.logger.info("Clear Queue Response: " + res)
                    self.connector.disable_transaction()
                Thread(target=clean_up).start()

            self.event_dispatcher.on_play(entry)



    def stop(self, entry, transition):
        """
        Stops the currently playing entry.

        Args:
            entry (Entry):                  The entry to stop playing
            transition (TransitionType):    The type of transition to use e.g. fade-out.
        """
        with suppress(LQConnectionError):
            self.connector.enable_transaction()

            if not entry.channel:
                self.logger.warn(SU.red("Trying to stop entry %s, but it has no channel assigned" % entry))
                return

            if transition == TransitionType.FADE:
                self.mixer.fade_out(entry.channel)
            else:
                self.mixer.channel_volume(entry.channel, 0)

            self.logger.info(SU.pink("Stopped channel '%s' for entry %s" % (entry.channel, entry)))
            self.connector.disable_transaction()
            self.event_dispatcher.on_stop(entry)



    def start_fallback_playlist(self, entries):
        """
        Sets any scheduled fallback playlist and performs a fade-in.

        Args:
            entries ([Entry]):    The playlist entries
        """
        self.preload_group(entries, ChannelType.FALLBACK_QUEUE)
        self.play(entries[0], TransitionType.FADE)
        self.event_dispatcher.on_fallback_updated(entries)



    def stop_fallback_playlist(self):
        """
        Performs a fade-out and clears any scheduled fallback playlist.
        """
        dirty_channel = self.channel_router.get_active(ChannelType.FALLBACK_QUEUE)

        self.logger.info(f"Fading out channel '{dirty_channel}'")
        self.connector.enable_transaction()
        self.mixer_fallback.fade_out(dirty_channel)
        self.connector.disable_transaction()

        def clean_up():
            # Wait a little, if there is some long fade-out. Note, this also means,
            # this channel should not be used for at least some seconds (including clearing time).
            time.sleep(2)
            self.connector.enable_transaction()
            self.mixer_fallback.channel_activate(dirty_channel.value, False)
            res = self.queue_clear(dirty_channel)
            self.logger.info("Clear Fallback Queue Response: " + res)
            self.connector.disable_transaction()
            self.event_dispatcher.on_fallback_cleaned(dirty_channel)
        Thread(target=clean_up).start()



    #
    #   Channel Type - Stream
    #


    def stream_load_entry(self, entry):
        """
        Loads the given stream entry and updates the entries's status codes.

        Args:
            entry (Entry):  The entry to be pre-loaded

        Returns:
            (Boolean):  `True` if successfull
        """
        self.stream_load(entry.channel, entry.source)
        time.sleep(1)

        retry_delay = self.config.get("input_stream_retry_delay")
        max_retries =  self.config.get("input_stream_max_retries")
        retries = 0

        while not self.stream_is_ready(entry.channel, entry.source):
            self.logger.info("Loading Stream ...")
            if retries >= max_retries:
                raise LoadSourceException("Could not connect to stream while waiting for %s seconds!" % str(retries*retry_delay))
            time.sleep(retry_delay)
            retries += 1

        return True



    def stream_load(self, channel, url):
        """
        Preloads the stream URL on the given channel. Note this method is blocking
        some serious amount of time; hence it's worth being called asynchroneously.

        Args:
            channel (Channel): The stream channel
            url (String):      The stream URL

        Returns:
            (Boolean):  `True` if successful
        """
        result = None

        self.connector.enable_transaction()
        result = self.connector.send_lqc_command(channel, "stream_stop")

        if result != LiquidsoapResponse.SUCCESS.value:
            self.logger.error("%s.stop result: %s" % (channel, result))
            raise LQStreamException("Error while stopping stream!")

        result = self.connector.send_lqc_command(channel, "stream_set_url", url)

        if result != LiquidsoapResponse.SUCCESS.value:
            self.logger.error("%s.set_url result: %s" % (channel, result))
            raise LQStreamException("Error while setting stream URL!")

        # Liquidsoap ignores commands sent without a certain timeout
        time.sleep(2)

        result = self.connector.send_lqc_command(channel, "stream_start")
        self.logger.info("%s.start result: %s" % (channel, result))

        self.connector.disable_transaction()
        return result



    def stream_is_ready(self, channel, url):
        """
        Checks if the stream on the given channel is ready to play. Note this method is blocking
        some serious amount of time even when successfull; hence it's worth being called asynchroneously.

        Args:
            channel (Channel): The stream channel
            url (String):      The stream URL

        Returns:
            (Boolean):  `True` if successful
        """
        result = None

        self.connector.enable_transaction()

        result = self.connector.send_lqc_command(channel, "stream_status")
        self.logger.info("%s.status result: %s" % (channel, result))

        if not result.startswith(LiquidsoapResponse.STREAM_STATUS_CONNECTED.value):
            return False

        lqs_url = result.split(" ")[1]
        if not url == lqs_url:
            self.logger.error("Wrong URL '%s' set for channel '%s', expected: '%s'." % (lqs_url, channel, url))
            return False

        self.connector.disable_transaction()

        stream_buffer = self.config.get("input_stream_buffer")
        self.logger.info("Ready to play stream, but wait %s seconds until the buffer is filled..." % str(stream_buffer))
        time.sleep(round(float(stream_buffer)))
        return True



    #
    #   Channel Type - Queue
    #


    def queue_push(self, channel, source):
        """
        Adds an filesystem URI to the given `ChannelType.QUEUE` channel.

        Args:
            channel (Channel): The channel to push the file to
            source (String):   The URI of the file

        Returns:
            (Boolean):  `True` if successful
        """
        if channel not in ChannelType.QUEUE.channels and \
            channel not in ChannelType.FALLBACK_QUEUE.channels:
                raise InvalidChannelException

        self.connector.enable_transaction()
        audio_store = self.config.abs_audio_store_path()
        extension = self.config.get("audio_source_extension")
        filepath = ResourceUtil.source_to_filepath(audio_store, source, extension)
        self.logger.info(SU.pink(f"{channel}.queue_push('{filepath}')"))
        result = self.connector.send_lqc_command(channel, "queue_push", filepath)
        self.logger.info("%s.queue_push result: %s" % (channel, result))
        self.connector.disable_transaction()

        # If successful, Liquidsoap returns a resource ID of the queued track
        resource_id = -1
        try:
            resource_id = int(result)
        except ValueError:
            self.logger.error(SU.red("Got an invalid resource ID: '%s'" % result))
            return False

        return resource_id >= 0



    def queue_seek(self, channel, seconds_to_seek):
        """
        Forwards the player of the given `ChannelType.QUEUE` channel by (n) seconds.

        Args:
            channel (Channel): The channel to push the file to
            seconds_to_seeks (Float):   The seconds to skip

        Returns:
            (String):   Liquidsoap response
        """
        if channel not in ChannelType.QUEUE.channels and \
            channel not in ChannelType.FALLBACK_QUEUE.channels:
                raise InvalidChannelException

        self.connector.enable_transaction()
        result = self.connector.send_lqc_command(channel, "queue_seek", str(seconds_to_seek))
        self.logger.info("%s.seek result: %s" % (channel, result))
        self.connector.disable_transaction()

        return result



    def queue_clear(self, channel):
        """
        Removes all tracks currently queued in the given `ChannelType.QUEUE` channel.

        Args:
            channel (Channel): The channel to push the file to

        Returns:
            (String):   Liquidsoap response
        """
        if channel not in ChannelType.QUEUE.channels and \
            channel not in ChannelType.FALLBACK_QUEUE.channels:
                raise InvalidChannelException

        self.logger.info(SU.pink("Clearing filesystem queue '%s'!" % channel))

        self.connector.enable_transaction()
        result = self.connector.send_lqc_command(channel, "queue_clear")
        self.logger.info("%s.clear result: %s" % (channel, result))
        self.connector.disable_transaction()

        return result



    #
    #   Channel Type - Playlist
    #


    def playlist_set_uri(self, channel, playlist_uri):
        """
        Sets the URI of a playlist.

        Args:
            channel (Channel): The channel to push the file to
            playlist_uri (String): The path to the playlist

        Returns:
            (String):   Liquidsoap response
        """
        self.logger.info(SU.pink("Setting URI of playlist '%s' to '%s'" % (channel, playlist_uri)))

        self.connector.enable_transaction()
        result = self.connector.send_lqc_command(channel, "playlist_uri_set", playlist_uri)
        self.logger.info("%s.playlist_uri result: %s" % (channel, result))
        self.connector.disable_transaction()

        return result



    def playlist_clear_uri(self, channel):
        """
        Clears the URI of a playlist.

        Args:
            channel (Channel): The channel to push the file to

        Returns:
            (String):   Liquidsoap response
        """
        self.logger.info(SU.pink("Clearing URI of playlist '%s'" % (channel)))

        self.connector.enable_transaction()
        result = self.connector.send_lqc_command(channel, "playlist_uri_clear")
        self.logger.info("%s.playlist_uri_clear result: %s" % (channel, result))
        self.connector.disable_transaction()

        return result





class EngineSplash:

    @staticmethod
    def splash_screen(component, version):
        """
        Prints the engine logo and version info.
        """
        return """\n
             █████╗ ██╗   ██╗██████╗  █████╗     ███████╗███╗   ██╗ ██████╗ ██╗███╗   ██╗███████╗
            ██╔══██╗██║   ██║██╔══██╗██╔══██╗    ██╔════╝████╗  ██║██╔════╝ ██║████╗  ██║██╔════╝
            ███████║██║   ██║██████╔╝███████║    █████╗  ██╔██╗ ██║██║  ███╗██║██╔██╗ ██║█████╗
            ██╔══██║██║   ██║██╔══██╗██╔══██║    ██╔══╝  ██║╚██╗██║██║   ██║██║██║╚██╗██║██╔══╝
            ██║  ██║╚██████╔╝██║  ██║██║  ██║    ███████╗██║ ╚████║╚██████╔╝██║██║ ╚████║███████╗
            ╚═╝  ╚═╝ ╚═════╝ ╚═╝  ╚═╝╚═╝  ╚═╝    ╚══════╝╚═╝  ╚═══╝ ╚═════╝ ╚═╝╚═╝  ╚═══╝╚══════╝
            %s v%s - Ready to play!
        \n""" % (component, version)