control.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 socket
import time
import json


from threading                  import Thread, Timer
from datetime                   import datetime, timedelta
from http_parser.http           import HttpStream
from http_parser.reader         import SocketReader

from modules.base.config        import AuraConfig
from modules.base.utils         import SimpleUtil as SU




class EngineControlInterface:
    """
    Provides ability to control the engine in various ways.
    """
    config = None
    logger = None   
    engine = None
    sci = None

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

        Args:
            config (AuraConfig):    Engine configuration
            logger (AuraLogger):    The logger
        """    
        self.config = AuraConfig.config()
        self.logger = logging.getLogger("AuraEngine")  
        self.sci = SocketControlInterface.get_instance(engine)
        self.logger.info(SU.yellow(f"[ECI] Engine Control Interface starting ..."))


    def terminate(self):
        """
        Terminates the instance and all related objects.
        """
        if self.sci: self.sci.terminate()



class SocketControlInterface:
    """
    Network socket server to control a running Engine from Liquidsoap.

    Note this server only allows a single connection at once. This
    service is primarly utilized to store new playlogs.
    """    
    PORT = 1337
    ACTION_ON_METADATA = "on_metadata"

    instance = None
    config = None
    logger = None    
    server = None
    engine = None


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

        Args:
            config (AuraConfig):    Engine configuration
            logger (AuraLogger):    The logger
        """
        if SocketControlInterface.instance:
            raise Exception(SU.red("[ECI] Socket server is already running!"))
        
        SocketControlInterface.instance = self
        self.config = AuraConfig.config()
        self.logger = logging.getLogger("AuraEngine")
        self.engine = engine              
        host = "127.0.0.1"
        thread = Thread(target = self.run, args = (self.logger, host))
        thread.start() 


    @staticmethod
    def get_instance(engine):
        """
        Returns the Singleton.
        """
        if not SocketControlInterface.instance:
            SocketControlInterface.instance = SocketControlInterface(engine)
        return SocketControlInterface.instance


    def attach(self, engine):
        """
        Attaches the engine to pass events to.
        """
        self.engine = engine


    def run(self, logger, host):
        """
        Starts the socket server
        """             
        while(True):       
            try:
                self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.server.bind((host, SocketControlInterface.PORT))
                break
            except OSError as e:
                wait_time = 2
                self.logger.error(SU.red(f"Cannot bind to Socket. Retrying in {wait_time} seconds..."))
                time.sleep(wait_time)

        logger.info(SU.yellow(f'[ECI] Listening at {host}:{SocketControlInterface.PORT}'))
        self.server.listen()    

        while(True):
            (conn, client) = self.server.accept()

            while(True):
                r = SocketReader(conn)
                p = HttpStream(r)
                data = p.body_file().read()                
                logger.debug(SU.yellow(f'[ECI] Received socket data from {str(client)}: {str(data)}'))
                
                try:
                    self.process(logger, json.loads(data))
                    conn.sendall(b'\n[ECI] processing done.\n')
                except Exception as e:
                    logger.error(SU.red(f'[ECI] Error while processing request: {data}'), e)

                conn.close()                    
                break


    def process(self, logger, data):
        """
        Process incoming actions.
        """
        if "action" in data:            
            if data["action"] == SocketControlInterface.ACTION_ON_METADATA:
                meta_data = data["data"]
                meta_data["duration"] = data["track_duration"]
                logger.debug(SU.yellow(f"[ECI] Executing action: "+SocketControlInterface.ACTION_ON_METADATA))
                self.engine.event_dispatcher.on_metadata(data["data"])
                logger.info(SU.yellow(f"[ECI] Event '{SocketControlInterface.ACTION_ON_METADATA}' issued successfully"))
            else:
                logger.error(SU.red("[ECI] Unknown action: " + data["action"]))
        else:
            logger.error(SU.red(f'[ECI] Missing action in request: {data}'))
    


    def terminate(self):
        SocketControlInterface.instance = None
        self.server.close()
        self.logger.info(SU.yellow("[ECI] Shutting down..."))





class EngineExecutor(Timer):
    """
    Base class for timed or threaded execution of Engine commands.

    Primarly used for automation by the scheduler.
    """
    logger = logging.getLogger("AuraEngine")    
    timer_store = {}    
    child_timer = None
    direct_exec = None
    timer_id = None
    timer_type = None
    param = None
    diff = None
    dt = None


    def __init__(self, timer_type="BASE", child_timer=None, due_time=None, func=None, param=None):
        """
        Constructor

        Args:
            timer_type (String):            Prefix used for the `timer_id` to make it unique
            child_timer (EngineExeuctor):   Child action which is bound to this timer
            due_time (Float):               When timer should be executed. For values <= 0 execution happens immediately in a threaded way
            func (function):                The function to be called
            param (object):                 Parameter passt to the function
        """        
        from modules.core.engine import Engine
        now_unix = Engine.engine_time()        
        self.child_timer = child_timer
        self.direct_exec = False
        self.timer_type = timer_type       
        self.timer_id = f"{timer_type}:{func.__name__}:{due_time}"

        if not due_time:
            diff = 0
        else:                
            diff = due_time - now_unix

        self.diff = diff
        self.dt = datetime.now() + timedelta(seconds=diff)
        self.func = func
        self.param = param

        if diff < 0:
            msg = f"Timer '{self.timer_id}' is due in the past. Executing immediately ..."
            self.logger.error(SU.red(msg))
            self.exec_now()
        elif diff == 0:   
            self.logger.info(f"Timer '{self.timer_id}' to be executed immediately")
            self.exec_now()                    
        else:
            self.exec_timed()            
            self.start()

        self.update_store()



    def exec_now(self):
        """
        Immediate execution within a thread. It's not stored in the timer store.
        """
        self.direct_exec = True
        thread = Thread(target = self.func, args = (self.param,))
        thread.start()



    def exec_timed(self):
        """
        Timed execution in a thread.
        """
        def wrapper_func(param=None):

            # Remove from store
            self.logger.info(SU.green(f"Removing old timer with ID: {self.timer_id}"))
            del EngineExecutor.timer_store[self.timer_id]          

            # Call actual function
            if param: self.func(param,) 
            else: self.func()

        Timer.__init__(self, self.diff, wrapper_func, (self.param,))


    def update_store(self):
        """
        Adds the instance to the store and cancels any previously existing commands.
        """
        existing_command = None
        if self.timer_id in EngineExecutor.timer_store:
            existing_command = EngineExecutor.timer_store[self.timer_id]    
        if existing_command:
            self.logger.info(SU.green(f"Cancelling previous timer with ID: {self.timer_id}"))
            existing_command.cancel()
            if existing_command.child_timer:
                self.logger.info(SU.green(f"Cancelling child timer with ID: {existing_command.child_timer.timer_id}"))

        EngineExecutor.timer_store[self.timer_id] = self
        self.logger.info(SU.green(f"Created command timer with ID: {self.timer_id}"))


    def is_alive(self):
        """
        Returns true if the command is still due to be executed.
        """
        if self.direct_exec == True:
            return False
        return super().is_alive()


    def __str__(self):
        """
        String represenation of the timer.
        """
        return f"[{self.timer_id}] exec at {str(self.dt)} (alive: {self.is_alive()})"


    @staticmethod
    def log_commands():
        """
        Prints a list of active timers to the log.
        """
        timers = EngineExecutor.timer_store.values()                
        msg = "\n [ ENGINE COMMAND QUEUE ]\n"

        if not timers:
            msg += "None available!\n"
        else:
            for timer in timers:
                msg += f"      =>   {str(timer)}\n"
                if timer.child_timer:
                    msg += f"            =>   {str(timer.child_timer)}\n"

        EngineExecutor.logger.info(msg + "\n")