events.py

#
# Aura Engine (https://gitlab.servus.at/aura/engine)
#
# Copyright (C) 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 datetime

from threading                  import Thread

from src.base.config            import AuraConfig
from src.plugins.mailer         import AuraMailer
from src.plugins.monitor        import AuraMonitor
from src.plugins.trackservice   import TrackServiceHandler
from src.scheduling.fallback    import FallbackManager


class EventBinding():
    """
    A binding between the event dispatcher and some event handler.

    It allows you to subscribe to events in a chained way:

        ```
        binding = dispatcher.attach(AuraMonitor)
        binding.subscribe("on_boot").subscribe("on_play")
        ```
    """
    dispatcher = None
    instance = None


    def __init__(self, dispatcher, instance):
        """
        Constructor
        """
        self.dispatcher = dispatcher
        self.instance = instance


    def subscribe(self, event_type):
        """
        Subscribes the instance to some event identified by the `event_type` string.
        """
        self.dispatcher.subscribe(self.instance, event_type)
        return self


    def get_instance(self):
        """
        Returns the object within that binding.
        """
        return self.instance



class EngineEventDispatcher():
    """
    Executes handlers for engine events.
    """
    logger = None
    config = None

    subscriber_registry = None
    engine = None
    scheduler = None
    monitor = None


    def __init__(self, engine):
        """
        Constructor
        """
        self.subscriber_registry = dict()
        self.logger = logging.getLogger("AuraEngine")
        self.config = AuraConfig.config()
        self.engine = engine

        binding = self.attach(AuraMailer)
        binding.subscribe("on_critical")
        binding.subscribe("on_sick")
        binding.subscribe("on_resurrect")
        binding.subscribe("on_fallback_active")

        binding = self.attach(AuraMonitor)
        binding.subscribe("on_boot")
        binding.subscribe("on_sick")
        binding.subscribe("on_resurrect")

        binding = self.attach(TrackServiceHandler)
        binding.subscribe("on_timeslot_start")
        binding.subscribe("on_timeslot_end")
        binding.subscribe("on_play")
        binding.subscribe("on_metadata")
        binding.subscribe("on_queue")



    #
    #   Methods
    #


    def attach(self, clazz):
        """
        Creates an instance of the given Class.
        """
        instance = clazz(self.engine)
        return EventBinding(self, instance)


    def subscribe(self, instance, event_type):
        """
        Subscribes to some event type. Preferably use it via `EventBinding.subscribe(..)`.
        """
        if not event_type in self.subscriber_registry:
            self.subscriber_registry[event_type] = []
        self.subscriber_registry[event_type].append(instance)


    def call_event(self, event_type, *args):
        """
        Calls all subscribers for the given event type.
        """
        if not event_type in self.subscriber_registry:
            return
        listeners = self.subscriber_registry[event_type]
        if not listeners:
            return
        for listener in listeners:
            method = getattr(listener, event_type)
            if method:
                if args and len(args) > 0:
                    method(*args)
                else:
                    method()


    #
    #   Events
    #


    def on_initialized(self):
        """
        Called when the engine is initialized, just before

        Important: Subsequent events are called synchronously, hence blocking.
        """
        self.logger.debug("on_initialized(..)")
        from src.scheduling.scheduler import AuraScheduler
        self.fallback_manager = FallbackManager(self.engine)
        self.scheduler = AuraScheduler(self.engine, self.fallback_manager)
        self.call_event("on_initialized", None)


    def on_boot(self):
        """
        Called when the engine is starting up. This happens after the initialization step.
        Connection to Liquidsoap should be available here.

        Important: Subsequent events are called synchronously, hence blocking.
        """
        self.logger.debug("on_boot(..)")
        self.call_event("on_boot")


    def on_ready(self):
        """
        Called when the engine has finished booting and is ready to play.
        """
        def func(self, param):
            self.logger.debug("on_ready(..)")
            self.scheduler.on_ready()
            self.call_event("on_ready", param)

        thread = Thread(target = func, args = (self, None))
        thread.start()


    def on_timeslot_start(self, timeslot):
        """
        Called when a timeslot starts.
        """
        def func(self, timeslot):
            self.logger.debug("on_timeslot_start(..)")
            self.fallback_manager.on_timeslot_start(timeslot)
            self.call_event("on_timeslot_start", timeslot)

        thread = Thread(target = func, args = (self, timeslot))
        thread.start()


    def on_timeslot_end(self, timeslot):
        """
        Called when a timeslot ends.
        """
        def func(self, timeslot):
            self.logger.debug("on_timeslot_end(..)")
            self.fallback_manager.on_timeslot_end(timeslot)
            self.call_event("on_timeslot_end", timeslot)

        thread = Thread(target = func, args = (self, timeslot))
        thread.start()


    def on_play(self, entry):
        """
        Event Handler which is called by the engine when some play command to Liquidsoap is issued.
        This does not indicate that Liquidsoap started playing actually, only that the command has
        been issued. To get the metadata update issued by Liquidsoap use `on_metadata` instead.

        Args:
            source (String):    The `PlaylistEntry` object
        """
        def func(self, entry):
            self.logger.debug("on_play(..)")
            # Assign timestamp indicating start play time. Use the actual playtime when possible.
            entry.entry_start_actual = datetime.datetime.now()
            self.scheduler.on_play(entry)
            self.call_event("on_play", entry)

        thread = Thread(target = func, args = (self, entry))
        thread.start()


    def on_metadata(self, data):
        """
        Event called by the soundsystem implementation (i.e. Liquidsoap) when some entry is actually playing.
        This does not include live or stream sources, since they ain't have metadata and are triggered from
        engine core (see `on_play(..)`).

        Args:
            data (dict):    A collection of metadata related to the current track
        """
        def func(self, data):
            self.logger.debug("on_metadata(..)")
            self.fallback_manager.on_metadata(data)
            self.call_event("on_metadata", data)

        thread = Thread(target = func, args = (self, data))
        thread.start()


    def on_stop(self, entry):
        """
        The entry on the assigned channel has been stopped playing.
        """
        def func(self, entry):
            self.logger.debug("on_stop(..)")
            self.call_event("on_stop", entry)

        thread = Thread(target = func, args = (self, entry))
        thread.start()


    # def on_fallback_updated(self, playlist_uri):
    #     """
    #     Called when the scheduled fallback playlist has been updated.
    #     This event does not indicate that the fallback is actually playing.
    #     """
    #     def func(self, playlist_uri):
    #         self.logger.debug("on_fallback_updated(..)")
    #         self.call_event("on_fallback_updated", playlist_uri)

    #     thread = Thread(target = func, args = (self, playlist_uri))
    #     thread.start()


    # def on_fallback_cleaned(self, cleaned_channel):
    #     """
    #     Called when the scheduled fallback queue has been cleaned up.
    #     This event does not indicate that some fallback is actually playing.
    #     """
    #     def func(self, cleaned_channel):
    #         self.logger.debug("on_fallback_cleaned(..)")
    #         self.call_event("on_fallback_cleaned", cleaned_channel)

    #     thread = Thread(target = func, args = (self, cleaned_channel))
    #     thread.start()


    def on_fallback_active(self, timeslot, fallback_type):
        """
        Called when a fallback is activated for the given timeslot,
        since no default playlist is available.
        """
        def func(self, timeslot, fallback_type):
            self.logger.debug("on_fallback_active(..)")
            self.call_event("on_fallback_active", timeslot, fallback_type)

        thread = Thread(target = func, args = (self, timeslot, fallback_type))
        thread.start()


    def on_queue(self, entries):
        """
        One or more entries have been queued and are currently pre-loaded.
        """
        def func(self, entries):
            self.logger.debug("on_queue(..)")
            self.call_event("on_queue", entries)

        thread = Thread(target = func, args = (self, entries))
        thread.start()


    def on_sick(self, data):
        """
        Called when the engine is in some unhealthy state.
        """
        def func(self, data):
            self.logger.debug("on_sick(..)")
            self.call_event("on_sick", data)

        thread = Thread(target = func, args = (self, data))
        thread.start()


    def on_resurrect(self, data):
        """
        Called when the engine turned healthy again after being sick.
        """
        def func(self, data):
            self.logger.debug("on_resurrect(..)")
            self.call_event("on_resurrect", data)

        thread = Thread(target = func, args = (self, data))
        thread.start()


    def on_critical(self, subject, message, data=None):
        """
        Callend when some critical event occurs
        """
        def func(self, subject, message, data):
            self.logger.debug("on_critical(..)")
            self.call_event("on_critical", (subject, message, data))

        thread = Thread(target = func, args = (self, subject, message, data))
        thread.start()