Events for start of timeslot and fallbacks. #38

src/core/engine.py

@@ -383,7 +383,7 @@ class Player:
         """
         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):

src/core/events.py

@@ -23,10 +23,10 @@ import datetime
 from threading                  import Thread
 
 from src.base.config            import AuraConfig
-from src.base.utils             import SimpleUtil as SU
 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():
@@ -94,6 +94,7 @@ class EngineEventDispatcher():
         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")
@@ -101,12 +102,18 @@ class EngineEventDispatcher():
         binding.subscribe("on_resurrect")
 
         binding = self.attach(TrackServiceHandler)
-        binding.subscribe("on_timeslot")
+        binding.subscribe("on_timeslot_start")
         binding.subscribe("on_play")
         binding.subscribe("on_metadata")
         binding.subscribe("on_queue")
 
 
+
+    #
+    #   Methods
+    #
+
+
     def attach(self, clazz):
         """
         Creates an instance of the given Class.
@@ -124,7 +131,7 @@ class EngineEventDispatcher():
         self.subscriber_registry[event_type].append(instance)
 
 
-    def call_event(self, event_type, args):
+    def call_event(self, event_type, *args):
         """
         Calls all subscribers for the given event type.
         """
@@ -136,24 +143,27 @@ class EngineEventDispatcher():
         for listener in listeners:
             method = getattr(listener, event_type)
             if method:
-                if args:
-                    method(args)
+                if args and len(args) > 0:
+                    method(*args)
                 else:
                     method()
 
 
     #
-    # Events
+    #   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.scheduler = AuraScheduler(self.engine)
+        self.fallback_manager = FallbackManager(self.engine)
+        self.scheduler = AuraScheduler(self.engine, self.fallback_manager)
         self.call_event("on_initialized", None)
 
 
@@ -161,32 +171,37 @@ class EngineEventDispatcher():
         """
         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", None)
+        self.call_event("on_boot")
 
 
     def on_ready(self):
         """
-        Called when the engine is booted and ready to play.
+        Called when the engine has finished booting and is ready to play.
         """
-        self.logger.debug("on_ready(..)")
-        self.scheduler.on_ready()
+        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(self, timeslot):
-        """
-        Event Handler which is called by the scheduler when the current timeslot is refreshed.
 
-        Args:
-            source (String):    The `PlaylistEntry` object
+    def on_timeslot_start(self, timeslot):
+        """
+        Called when a timeslot starts.
         """
         def func(self, timeslot):        
-            self.logger.debug("on_timeslot(..)")
-            self.call_event("on_timeslot", 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()    
+        thread.start() 
 
 
     def on_play(self, entry):
@@ -208,14 +223,16 @@ class EngineEventDispatcher():
 
     def on_metadata(self, data):
         """
-        Event Handler which is called by the soundsystem implementation (i.e. Liquidsoap)
-        when some entry is actually playing.
+        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))
@@ -237,41 +254,39 @@ class EngineEventDispatcher():
     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.
         """
-        self.logger.debug("on_fallback_updated(..)")
-        self.call_event("on_fallback_updated", playlist_uri)
+        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.
         """
-        self.logger.debug("on_fallback_cleaned(..)")
-        self.call_event("on_fallback_cleaned", cleaned_channel)
-
-
-    def on_idle(self):
-        """
-        Callend when no entry is playing
-        """
-        def func(self):
-            self.logger.debug("on_idle(..)")
-            self.logger.error(SU.red("Currently there's nothing playing!"))
-            self.call_event("on_idle", None)
+        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, ))
+        thread = Thread(target = func, args = (self, cleaned_channel))
         thread.start() 
 
 
-    def on_timeslot_change(self, timeslot):
+    def on_fallback_active(self, timeslot, fallback_type):
         """
-        Called when the playlist or entries of the current timeslot have changed.
+        Called when a fallback is activated for the given timeslot,
+        since no default playlist is available.
         """
-        def func(self, timeslot):        
-            self.logger.debug("on_timeslot_change(..)")
-            self.call_event("on_timeslot_change", timeslot)
+        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))
+        thread = Thread(target = func, args = (self, timeslot, fallback_type))
         thread.start() 
 
 

src/plugins/trackservice.py

@@ -53,7 +53,7 @@ class TrackServiceHandler():
 
 
 
-    def on_timeslot(self, timeslot=None):
+    def on_timeslot_start(self, timeslot=None):
         """
         Some new timeslot has just started.
         """

src/scheduling/scheduler.py

@@ -36,7 +36,6 @@ from src.core.engine            import Engine
 from src.core.channels          import ChannelType, TransitionType, EntryPlayState
 from src.core.resources         import ResourceClass, ResourceUtil
 from src.scheduling.calendar    import AuraCalendarService
-from src.scheduling.fallback    import FallbackManager
 
 
 
@@ -51,6 +50,29 @@ class EntryQueueState(Enum):
 
 
 
+class TimeslotCommand(EngineExecutor):
+    """
+    Command for triggering start of timeslot events.
+    """
+    engine = None
+
+    def __init__(self, engine, timeslot):
+        """
+        Constructor
+
+        Args:
+            timeslot (Timeslot):    The timeslot which is starting at this time
+        """        
+        self.engine = engine
+
+        def do_start_timeslot(timeslot):
+            self.logger.info(SU.cyan(f"=== on_timeslot_start('{timeslot}') ==="))
+            self.engine.event_dispatcher.on_timeslot_start(timeslot)
+                            
+        super().__init__("TIMESLOT", None, timeslot.start_unix, do_start_timeslot, timeslot)
+
+
+
 class AuraScheduler(threading.Thread):
     """
     Aura Scheduler Class
@@ -84,7 +106,7 @@ class AuraScheduler(threading.Thread):
 
 
 
-    def __init__(self, engine):
+    def __init__(self, engine, fallback_manager):
         """
         Constructor
 
@@ -97,7 +119,7 @@ class AuraScheduler(threading.Thread):
         self.logger = logging.getLogger("AuraEngine")
 
         AuraScheduler.init_database()
-        self.fallback = FallbackManager(self)
+        self.fallback = fallback_manager
         self.engine = engine
         self.engine.scheduler = self
         self.is_soundsytem_init = False
@@ -143,29 +165,26 @@ class AuraScheduler(threading.Thread):
             self.clean_timer_queue()
             self.print_timer_queue()
                         
-            # FIXME better location for call
-            if self.engine.event_dispatcher:
-                current_timeslot = self.get_active_timeslot()
-                self.engine.event_dispatcher.on_timeslot(current_timeslot)
-            
+
             EngineExecutor.log_commands()            
             self.exit_event.wait(seconds_to_wait)
 
 
 
-#
-#   PUBLIC METHODS
-#
+    #
+    #   EVENTS
+    #
 
 
     def on_ready(self):
         """
-        Called when the engine is ready.
+        Called when the engine has finished booting and is ready to play.
         """
         self.is_initialized = True
         self.on_scheduler_ready()
 
 
+
     def on_scheduler_ready(self):
         """
         Called when the scheduler is ready.
@@ -181,6 +200,11 @@ class AuraScheduler(threading.Thread):
 
 
 
+    #
+    #   METHODS
+    #
+
+
     def play_active_entry(self):
         """
         Plays the entry scheduled for the very current moment and forwards to the scheduled position in time. 
@@ -194,6 +218,9 @@ class AuraScheduler(threading.Thread):
 
         # 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)
             # Queue the fade-out of the timeslot
             if not active_timeslot.fadeouttimer:
@@ -546,7 +573,10 @@ class AuraScheduler(threading.Thread):
         # Queue the timeslots, their playlists and entries
         if timeslots:
             for next_timeslot in timeslots:
-                # Timeslot any available fallback playlist
+                # 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)
 
                 if next_timeslot.playlist: