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, Lock
from datetime               import datetime, timedelta
from http_parser.http       import HttpStream
from http_parser.reader     import SocketReader

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




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

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

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



    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
    event_dispatcher = None


    def __init__(self, event_dispatcher):
        """
        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.event_dispatcher = event_dispatcher
        host = "127.0.0.1"
        thread = Thread(target = self.run, args = (self.logger, host))
        thread.start()


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



    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.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.

    Primarily used for automations performed by the scheduler.
    """
    _lock = None
    logger = logging.getLogger("AuraEngine")
    initialized = None
    timer_store = {}
    parent_timer = None
    child_timer = None
    direct_exec = None
    timer_id = None
    timer_type = None
    param = None
    diff = None
    dt = None


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

        Args:
            timer_type (String):            Prefix used for the `timer_id` to make it unique
            parent_timer (EngineExeuctor):  Parent action which is a prerequisite for 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
        """
        self.initialized = False
        self._lock = Lock()
        from src.engine import Engine
        now_unix = Engine.engine_time()

        # Init parent-child relation
        self.parent_timer = parent_timer
        if self.parent_timer:
            self.parent_timer.child_timer = self

        # Init meta data
        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

        is_stored = self.update_store()
        if not is_stored:
            self.logger.info(SU.red(f"Timer '{self.timer_id}' omitted because it's already existing but dead"))
        else:
            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()


    def on_ready(self, func):
        """
        Calls the passed function `func` when the timer is ready.
        """
        while self.initialized == False:
            timer.sleep(0.001)
            self.logger.info(SU.orange("Waiting until the EngineExecutor is done with initialization..."))

        if not self.direct_exec: #TODO Evaluate if we should join for direct exec too
            self.join()
        func()


    def wait_for_parent(self):
        """
        Child timers are dependend on their parents. So let's wait until parents are done with their stuff.
        """
        if self.parent_timer:
            while self.parent_timer.is_alive():
                self.logger.info(f"Timer '{self.timer_id}' is waiting for parent timer '{self.parent_timer.timer_id}' to finish")
                time.sleep(0.2)


    def exec_now(self):
        """
        Immediate execution within a thread. It's not stored in the timer store.

        Assigns the `timer_id` as the thread name.
        """
        self.direct_exec = True
        self.wait_for_parent()
        thread = Thread(name=self.timer_id, target=self.func, args=(self.param,))
        thread.start()
        self.initialized = True


    def exec_timed(self):
        """
        Timed execution in a thread.

        Assigns the `timer_id` as the thread name.
        """
        def wrapper_func(param=None):
            self.wait_for_parent()
            if param: self.func(param,)
            else: self.func()
        super().__init__(self.diff, wrapper_func, (self.param,))
        self._name = self.timer_id
        self.initialized = True


    def update_store(self):
        """
        Adds the instance to the store and cancels any previously existing commands.

        If a timer with the given ID is already existing but also already executed,
        then it is not added to the store. In such case the method returns `False`.

        Returns:
            (Boolean):  True if the timer has been added to the store
        """
        with self._lock:
            existing_command = None
            if self.timer_id in EngineExecutor.timer_store:
                existing_command = EngineExecutor.timer_store[self.timer_id]

            if existing_command:

                # Check if existing timer has been executed already -> don't update
                if not existing_command.is_alive():
                    self.logger.debug(f"Existing dead timer with ID: {self.timer_id}. Don't update.")
                    return False

                # Still waiting for execution -> update
                else:
                    self.logger.debug(f"Cancelling existingTimer with ID: {self.timer_id}")
                    existing_command.cancel()
                    if existing_command.child_timer:
                        self.logger.debug(f"Cancelling existingTimer:childTimer with ID: {existing_command.child_timer.timer_id}")

            EngineExecutor.timer_store[self.timer_id] = self
            self.logger.debug(f"Created command timer with ID: {self.timer_id}")
            return True


    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 remove_stale_timers():
        """
        Removes timers from store which have been executed and are older than 3 hours.
        """
        timers = EngineExecutor.timer_store.values()
        del_keys = []

        for timer in timers:
            if timer.dt < datetime.now() - timedelta(hours=3):
                if not timer.child_timer or (timer.child_timer and not timer.child_timer.is_alive()):
                    timer.logger.debug(f"Removing already executed timer with ID: {timer.timer_id}")
                    del_keys.append(timer.timer_id)

        for timer_id in del_keys:
            del EngineExecutor.timer_store[timer_id]


    @staticmethod
    def log_commands():
        """
        Prints a list of recent active and inactive timers.
        """
        msg = SU.blue("\n [ ENGINE COMMAND QUEUE ]\n")
        EngineExecutor.remove_stale_timers()
        timers = EngineExecutor.timer_store.values()

        if not timers:
            msg += "\nNone available!\n"
        else:
            for timer in timers:
                if not timer.parent_timer:
                    line = f"      =>   {str(timer)}\n"#
                    if timer.is_alive():
                        line = SU.green(line)
                    msg += line
                if timer.child_timer:
                    line = f"            =>   {str(timer.child_timer)}\n"
                    if timer.child_timer.is_alive():
                        line = SU.green(line)
                    msg += line

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