src/aura_engine/app.py

+#!/usr/bin/env python3
+#
+# 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/>.
+
+
+"""
+Entrypoint to run the Engine.
+"""
+
+import logging
+import signal
+import sys
+import threading
+
+import confuse
+
+from aura_engine.base.config import AuraConfig
+from aura_engine.base.logger import AuraLogger
+from aura_engine.engine import Engine
+
+
+class EngineRunner:
+    """
+    EngineRunner is in charge of starting the engine.
+    """
+
+    logger: logging.Logger
+    config: confuse.Configuration
+    engine: Engine
+
+    def __init__(self):
+        """
+        Constructor.
+        """
+        self.config = AuraConfig.instance.config
+        AuraLogger(self.config)
+        self.logger = logging.getLogger("engine")
+        self.engine = Engine()
+
+    def run(self):
+        """
+        Start Engine Core.
+        """
+        self.engine.start()
+
+    def exit_gracefully(self, signum, frame):
+        """
+        Shutdown of the engine. Also terminates the Liquidsoap thread.
+        """
+        for thread in threading.enumerate():
+            self.logger.info(thread.name)
+
+        if self.engine:
+            self.engine.terminate()
+
+        self.logger.info(f"Gracefully terminated Aura Engine! (signum:{signum}, frame:{frame})")
+        sys.exit(0)
+
+
+#
+# START THE ENGINE
+#
+
+
+if __name__ == "__main__":
+    runner = EngineRunner()
+    signal.signal(signal.SIGINT, runner.exit_gracefully)
+    signal.signal(signal.SIGTERM, runner.exit_gracefully)
+    runner.run()

src/aura_engine/base/api.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/>.
+
+
+"""
+Simple API library.
+
+~~~~~~~~~~~~~~~~~~~~~
+
+Simple API is an HTTP library, for implicit error handling. This is meant for situations
+where you are mostly interested in results and keep going independently if there is an
+error or not.
+
+By default the library is meant for digesting REST endpoints.
+
+The responses are wrapped in a dictionary containing additial fields which are used at a
+frequent basis:
+
+```py
+{
+    "response" (Response): Actual Python response object
+    "error" (String): In case an error occurred
+    "exception" (Exception): In case an exception occurred
+    "json" (Dictionary): In case of an json request this is the deserialized data
+}
+```
+
+Basic GET usage:
+
+   >>> from api import SimpleRestApi
+   >>> r = api.get("https://aura.radio/foo.json")
+   >>> r.response.status_code
+   200
+   >>> r.response.content
+   <raw content>
+   >>> r.json
+   <unmarshalled json>
+
+"""
+
+import json
+import logging
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+import requests
+
+from aura_engine.base.lang import DotDict
+from aura_engine.base.utils import SimpleUtil as SU
+
+
+class SimpleRestApi:
+    """
+    Simple wrapper on `requests` to deal with REST APIs.
+
+    Use it for services which do not want to deal with exception
+    handling but with results only.
+
+    SimpleRestApi has implicit logging of invalid states at logs
+    to the `engine` logger by default.
+    """
+
+    CONTENT_JSON = "application/json"
+    default_headers = {"content-type": "application/json"}
+
+    logger = None
+
+    def __init__(self, logger_name="engine"):
+        self.logger = logging.getLogger(logger_name)
+
+    def exception_handler(func):
+        """
+        Decorate functions with `@exception_handler` to handle API exceptions in a simple way.
+
+        Args:
+            func (_type_): The decorated function
+
+        """
+
+        def handle_response(*args, **kwargs) -> dict:
+            """Process the decorator response.
+
+            Returns:
+                dict: {
+                    "response": requests.Response object
+                    "error": String with error message
+                    "exception": The actual exception
+                }
+
+            """
+            fn = func.__name__.upper()
+            msg_template = f"during {fn} at '{args[1]}'"
+            error = None
+            exc = None
+            response_dict = None
+
+            response = requests.Response()
+            response.status_code = 400
+
+            try:
+                response = func(*args, **kwargs)
+                if isinstance(response, dict):
+                    response_dict = response
+                    response = response_dict.response
+                if int(response.status_code) >= 300:
+                    reason = "-"
+                    if hasattr(response, "reason"):
+                        reason = response.reason
+                    error = f"{response.status_code} | Error {msg_template}: {reason}"
+                    args[0].logger.error(SU.red(error))
+            except requests.exceptions.ConnectionError as e:
+                exc = e
+                error = f"Bad Request {msg_template}"
+                args[0].logger.error(SU.red(error))
+            except requests.exceptions.Timeout as e:
+                exc = e
+                error = f"Timeout {msg_template}"
+                args[0].logger.error(SU.red(error))
+            except requests.exceptions.RequestException as e:
+                exc = e
+                error = f"Unknown Error {msg_template}"
+                args[0].logger.error(SU.red(error))
+            except Exception as e:
+                exc = e
+                error = f"Unknown Exception {msg_template}"
+                args[0].logger.error(SU.red(error), e)
+            finally:
+                result_dict = {"response": response, "error": error, "exception": exc}
+                if response_dict:
+                    result_dict = result_dict | response_dict
+                return DotDict(result_dict)
+
+        return handle_response
+
+    def clean_dictionary(self, data: dict) -> dict:
+        """
+        Delete keys with the value `None` in a dictionary, recursively.
+
+        Args:
+            data (dict): The dictionary
+
+        Returns:
+            (dict): The cleaned dictionary
+
+        """
+        data = data.copy()
+        for key, value in list(data.items()):
+            if value is None:
+                del data[key]
+            elif isinstance(value, dict):
+                SimpleRestApi.clean_dictionary(self, value)
+        return data
+
+    def serialize_json(self, data: dict, clean_data=True) -> str:
+        """
+        Marshall a dictionary as JSON String.
+
+        Args:
+            data (dict): Dictionary holding the data
+
+        Returns:
+            str: JSON String
+
+        """
+        if clean_data:
+            data = self.clean_dictionary(data)
+        json_data = json.dumps(data, indent=4, sort_keys=True, default=str)
+        self.logger.info("Built JSON: " + json_data)
+        return json_data
+
+    def deserialize_json(self, response: str) -> str:
+        """
+        Unmarshall a JSON String to a dictionary.
+
+        Args:
+            response (dict): Response object
+
+        Returns:
+            dict: JSON as dictionary
+
+        """
+        json_data = None
+        try:
+            json_data = response.json()
+        except Exception:
+            self.logger.error(f"Invalid JSON: {response.content}")
+            return None
+        return json_data
+
+    @exception_handler
+    def get(
+        self, url: str, headers: dict = None, params: dict = None, timeout: int = 5
+    ) -> requests.Response:
+        """
+        GET from an URL.
+
+        Args:
+            url (str): The URL of the request
+
+        Returns:
+            {
+                "response": requests.Response,
+                "error": str,
+                "exception": Exception
+            }
+
+        """
+        json_data = None
+        if not headers:
+            headers = SimpleRestApi.default_headers
+        response = requests.get(url, headers=headers, params=params, timeout=timeout)
+        if headers.get("content-type") == SimpleRestApi.CONTENT_JSON:
+            json_data = self.deserialize_json(response)
+        return DotDict({"response": response, "json": json_data})
+
+    @exception_handler
+    def post(self, url: str, data: dict, headers: dict = None, timeout: int = 5):
+        """
+        POST to an URL.
+
+        Args:
+            url (str): The URL of the request
+            data (dict): Data payload for request body
+            headers (dict, optional): Optional headers, defaults to `SimpleRestApi.default_headers`
+
+        Returns:
+            {
+                "response": requests.Response,
+                "error": str,
+                "exception": Exception
+            }
+
+        """
+        if not headers:
+            headers = SimpleRestApi.default_headers
+        body: str = self.serialize_json(data)
+        return requests.post(url, data=body, headers=headers, timeout=timeout)
+
+    @exception_handler
+    def put(
+        self, url: str, data: dict, headers: dict = None, timeout: int = 5
+    ) -> requests.Response:
+        """
+        PUT to an URL.
+
+        Args:
+            url (str): The URL of the request
+            data (dict): Data payload for request body
+            headers (dict, optional): Optional headers, defaults to `SimpleRestApi.default_headers`
+
+        Returns:
+            {
+                "response": requests.Response,
+                "error": str,
+                "exception": Exception
+            }
+
+        """
+        if not headers:
+            headers = SimpleRestApi.default_headers
+        body: str = self.serialize_json(data)
+        return requests.put(url, data=body, headers=headers, timeout=timeout)
+
+
+class SimpleCachedRestApi:
+    """
+    Wrapper to cache GET responses based on the simple REST API.
+
+    It uses a network-first strategy:
+
+      1. Query the requested API endpoint
+      2. Store the result in a JSON file
+      3. Return the result as a JSON object
+
+    If the API endpoint is not available at step 1.) the cached JSON from the
+    most recent, previously successful request is returned.
+
+    """
+
+    cache_location: str
+    simple_api: SimpleRestApi
+
+    logger = None
+
+    def __init__(self, simple_api: SimpleRestApi, cache_location: str, logger_name="engine"):
+        if cache_location[-1] != "/":
+            cache_location += "/"
+        cache_location += "api/"
+        os.makedirs(cache_location, exist_ok=True)
+        self.simple_api = simple_api
+        self.cache_location = cache_location
+        self.logger = logging.getLogger(logger_name)
+
+    def get(self, url: str, headers: dict = None, params: dict = None) -> requests.Response:
+        """
+        GET from an URL while also storing the result in the local cache.
+
+        Args:
+            url (str): The URL of the request
+
+        Returns:
+            {
+                "response": requests.Response,
+                "error": str,
+                "exception": Exception
+            }
+
+        """
+        filename = self.build_filename(url)
+        cache_filepath = self.cache_location + filename
+        result = self.simple_api.get(url, headers, params)
+
+        if result and result.json and result.response.status_code == 200:
+            with open(cache_filepath, "w") as file:
+                json.dump(result.json, file)
+                file.close()
+        else:
+            json_data = None
+            try:
+                file = open(cache_filepath, "r")
+                json_data = json.load(file)
+                file.close()
+            except FileNotFoundError:
+                pass
+
+            if json_data:
+                result = {
+                    "response": DotDict({"status_code": 304, "error": "Not Modified"}),
+                    "json": json_data,
+                }
+            else:
+                result = {
+                    "response": DotDict({"status_code": 404, "error": "Not Found in local cache"}),
+                    "json": None,
+                }
+
+        return DotDict(result)
+
+    def build_filename(self, url: str) -> str:
+        """
+        Build a valid file name based on the URI parts of an URL.
+
+        Args:
+            url (str): The URL to build the filename from
+
+        Returns:
+            str: File name representing an URL
+        """
+        parts = urlparse(url)
+        dirs = parts.path.strip("/").split("/")
+        return "-".join(dirs) + ".json"
+
+    def prune_cache_dir(self):
+        """
+        Delete everything in the API cache directory.
+        """
+        [f.unlink() for f in Path(self.cache_location).iterdir() if f.is_file()]
+
+
+class LiquidsoapUtil:
+    """
+    Utilities specific to Liquidsoap.
+    """
+
+    @staticmethod
+    def json_to_dict(data: str) -> dict:
+        """
+        Convert a Liquidsoap JSON String to dictionary.
+        """
+        data = data.replace("+", " ")
+        data = data.replace("-", " ")
+        data = requests.utils.unquote(data)
+        return json.loads(data)
+
+    @staticmethod
+    def annotate_uri(uri: str, annotations: dict) -> str:
+        """
+        Wrap the given URI with the passed annotation dictionary.
+        """
+        metadata = ""
+        for k, v in annotations.items():
+            metadata += f'{k}="{v}",'
+        uri = f"annotate:{metadata[:-1]}:{uri}"
+        return uri

src/aura_engine/base/config.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/>.
+
+
+"""
+Dealing with configuration data.
+"""
+
+import logging
+import os
+import os.path
+import re
+import sys
+from pathlib import Path
+
+import confuse
+import yaml
+
+template = {
+    "general": {
+        "socket_dir": str,
+        "cache_dir": str,
+    },
+    "log": {
+        "directory": str,
+        "level": confuse.OneOf(["debug", "info", "warning", "error", "critical"]),
+    },
+    "monitoring": {
+        "heartbeat": {"host": str, "port": int, "frequency": int},
+    },
+    "api": {
+        "steering": {"status": str, "calendar": str},
+        "tank": {"session": str, "secret": str, "status": str, "playlist": str},
+        "engine": {
+            "number": int,
+            "status": str,
+            "store_playlog": str,
+            "store_clock": str,
+            "store_health": str,
+        },
+    },
+    "scheduler": {
+        "audio": {
+            "source_folder": str,
+            "source_extension": str,
+            "playlist_folder": str,
+            "engine_latency_offset": float,
+        },
+        "fetching_frequency": int,
+        "scheduling_window_start": int,
+        "scheduling_window_end": int,
+        "preload_offset": int,
+        "input_stream": {"buffer": float},
+        "fade_in_time": float,
+        "fade_out_time": float,
+    },
+}
+
+
+class AuraConfig:
+    """
+    Creates config by reading yaml file according to template above.
+    """
+
+    _instance = None
+    config_file_path = ""
+    confuse_config: confuse.Configuration
+    config: confuse.Configuration | None = (
+        None  # TODO points to a validated config (hopefully later)
+    )
+    logger: logging.Logger | None = None
+
+    # FIXME: Class properties are deprecated in Python 3.11
+    # and will not be supported in Python 3.13
+    @classmethod
+    @property
+    def instance(cls):
+        """Create and return singleton instance."""
+        if cls._instance is None:
+            cls._instance = AuraConfig()
+        return cls._instance
+
+    def __init__(self, config_file_path="/etc/aura/engine.yaml"):
+        """
+        Initialize the configuration, defaults to `/etc/aura/engine.yaml`.
+
+        If this file doesn't exist it uses `./config/engine.yaml` from
+        the project directory.
+
+        Args:
+            config_file_path(String): The path to the configuration file `engine.yaml`
+
+        """
+        self.logger = logging.getLogger("engine")
+        config_file = Path(config_file_path)
+        project_root = Path(__file__).parent.parent.parent.parent.absolute()
+
+        if not config_file.is_file():
+            config_file_path = f"{project_root}/config/engine.yaml"
+
+        self.config_file_path = config_file_path
+        print(f"Using configuration at: {config_file_path}")
+
+        envar_matcher = re.compile(r"\$\{([^}^{]+)\}")
+
+        def envar_constructor(loader, node):
+            value = os.path.expandvars(node.value)
+            # workaround not to parse numerics as strings
+            try:
+                value = int(value)
+            except ValueError:
+                pass
+            try:
+                value = float(value)
+            except ValueError:
+                pass
+            return value
+
+        envar_loader = yaml.SafeLoader
+        envar_loader.add_implicit_resolver("!envar", envar_matcher, None)
+        envar_loader.add_constructor("!envar", envar_constructor)
+
+        self.confuse_config = confuse.Configuration("engine", loader=envar_loader)
+        self.confuse_config.set_file(config_file_path)
+        self.load_config()
+
+        # custom overrides and defaults
+        self.confuse_config["install_dir"].set(os.path.realpath(project_root))
+        self.confuse_config["config_dir"].set(os.path.dirname(config_file_path))
+
+        AuraConfig.instance = self
+
+    def init_version(self, version: dict):
+        """
+        Read and set the component version from VERSION file in project root.
+        """
+        self.confuse_config["version_control"].set(version.get("control"))
+        self.confuse_config["version_core"].set(version.get("core"))
+        self.confuse_config["version_liquidsoap"].set(version.get("liquidsoap"))
+
+    def load_config(self):
+        """
+        Set config defaults and load settings from file.
+        """
+        if not os.path.isfile(self.config_file_path):
+            self.logger.critical(self.config_file_path + " not found  :(")
+            sys.exit(1)
+
+        self.config = self.confuse_config.get(template)

src/aura_engine/base/lang.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/>.
+
+
+"""
+A collection of meta-programming and language utilities.
+"""
+
+import inspect
+from functools import wraps
+from multiprocessing import Lock
+
+
+def synchronized(member):
+    """
+    @synchronized decorator.
+
+    Lock a method for synchronized access only. The lock is stored to the function or class
+    instance, depending on what is available.
+    """
+
+    @wraps(member)
+    def wrapper(*args, **kwargs):
+        lock = vars(member).get("_synchronized_lock", None)
+        result = ""
+        try:
+            if lock is None:
+                lock = vars(member).setdefault("_synchronized_lock", Lock())
+            lock.acquire()
+            result = member(*args, **kwargs)
+            lock.release()
+        except Exception as e:
+            lock.release()
+            raise e
+        return result
+
+    return wrapper
+
+
+def private(member):
+    """
+    @private decorator.
+
+    Use this to annotate your methods for private-visibility.
+
+    This is an more expressive alternative to the pythonic underscore visibility.
+    """
+
+    @wraps(member)
+    def wrapper(*args, **kwargs):
+        me = member.__name__
+        stack = inspect.stack()
+        calling_class = stack[1][0].f_locals["self"].__class__.__name__
+        calling_method = stack[1][0].f_code.co_name
+        if calling_method not in dir(args[0]) and calling_method is not me:
+            msg = f'"{me}(..)" called by "{calling_class}.{calling_method}(..)" is private'
+            print(msg)
+            raise Exception(msg)
+        return member(*args, **kwargs)
+
+    return wrapper
+
+
+class DotDict(dict):
+    """
+    Wrap a dictionary with `DotDict()` to allow property access using the dot.notation.
+    """
+
+    __getattr__ = dict.get
+    __setattr__ = dict.__setitem__
+    __delattr__ = dict.__delitem__

src/aura_engine/base/logger.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/>.
+
+
+"""
+Logging all the noise.
+"""
+
+import logging
+
+import confuse
+
+
+class AuraLogger:
+    """
+    Logger for all Aura Engine components.
+
+    The default logger is `AuraEngine`. Other loggers are defined by passing a custom name on
+    instantiation. The logger respects the log-level as defined in the engine's configuration
+    file.
+    """
+
+    config: confuse.Configuration
+    logger: logging.Logger
+
+    def __init__(self, config: confuse.Configuration, name="engine"):
+        """
+        Initialize the logger.
+
+        Args:
+            config (AuraConfig): The configuration file
+            name (String): The name of the logger
+
+        """
+        self.config = config
+        lvl = self.get_log_level()
+        self.create_logger(name, lvl)
+
+    def get_log_level(self):
+        """
+        Retrieve the configured log level (default=INFO).
+        """
+        lvl = self.config.log.level
+        mapping = {
+            "debug": logging.DEBUG,
+            "info": logging.INFO,
+            "warning": logging.WARNING,
+            "error": logging.ERROR,
+            "critical": logging.CRITICAL,
+        }
+
+        log_level = mapping.get(lvl)
+        if not log_level:
+            print("No log level configured. Using INFO.")
+            log_level = logging.INFO
+        print(f"Setting log level {log_level} ({lvl})")
+        return log_level
+
+    def create_logger(self, name, lvl):
+        """
+        Create the logger instance for the given name.
+
+        Args:
+            name (String): The name of the logger
+            lvl (Enum): The logging level
+
+        """
+        self.logger = logging.getLogger(name)
+        self.logger.setLevel(lvl)
+
+        if not self.logger.hasHandlers():
+            # create file handler for logger
+            file_handler = logging.FileHandler(self.config.log.directory + "/" + name + ".log")
+            file_handler.setLevel(lvl)
+
+            # create stream handler for logger
+            stream_handler = logging.StreamHandler()
+            stream_handler.setLevel(lvl)
+
+            # set format of log
+            datepart = "%(asctime)s:%(name)s:%(levelname)s"
+            message = " - %(message)s - "
+            filepart = "[%(filename)s:%(lineno)s-%(funcName)s()]"
+            formatter = logging.Formatter(datepart + message + filepart)
+
+            # set log of handlers
+            file_handler.setFormatter(formatter)
+            stream_handler.setFormatter(formatter)
+
+            # add handlers to the logger
+            self.logger.addHandler(file_handler)
+            self.logger.addHandler(stream_handler)
+
+            self.logger.debug("ADDED HANDLERS")
+        else:
+            self.logger.debug("REUSED LOGGER")

src/aura_engine/base/utils.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/>.
+
+
+"""
+A collection of all kinds of simplifications.
+"""
+
+import datetime
+import json
+import time
+from enum import Enum
+from logging import Logger
+
+
+class SimpleUtil:
+    """
+    A container class for simple utility methods.
+    """
+
+    @staticmethod
+    def string_to_datetime(datetime_str: str):
+        """
+        Convert a ISO 8601 date-time string into `datetime`.
+        """
+        if datetime_str:
+            return datetime.datetime.fromisoformat(datetime_str)
+        return None
+
+    @staticmethod
+    def timestamp_to_datetime(timestamp: float) -> datetime:
+        """
+        Convert a timestamp to datetime.
+
+        Args:
+            timestamp (float): The timestamp to convert.
+
+        Returns:
+            (datetime): The `datetime` object.
+        """
+        if timestamp:
+            return datetime.datetime.fromtimestamp(timestamp)
+        return None
+
+    @staticmethod
+    def fmt_time(timestamp: int):
+        """
+        Format a UNIX timestamp to a String displaying time in the format '%H:%M:%S'.
+
+        Args:
+            (Integer) timestamp: Unix epoch
+
+        Returns:
+            (String): Displaying the time
+
+        """
+        return datetime.datetime.fromtimestamp(timestamp).strftime("%H:%M:%S")
+
+    @staticmethod
+    def round_seconds(dt: datetime) -> datetime:
+        """
+        Rounds date/time to the nearest second.
+
+        Args:
+            dt (datetime): the date/time object to round.
+
+        Returns:
+            datetime: the rounded version.
+
+        """
+        rounded_dt = dt + datetime.timedelta(seconds=0.5)
+        return rounded_dt.replace(microsecond=0)
+
+    @staticmethod
+    def nano_to_seconds(nanoseconds) -> float:
+        """
+        Convert nano-seconds to seconds.
+
+        Args:
+            (Integer) nanoseconds
+
+        Returns:
+            (Float): seconds
+
+        @deprecated since Tank moves to seconds as float.
+        """
+        return float(nanoseconds / 1000000000)
+
+    @staticmethod
+    def timestamp(date_and_time=None):
+        """
+        Transform the given `datetime` into a UNIX epoch timestamp.
+
+        If no parameter is passed, the current timestamp is returned.
+
+        Args:
+            (Datetime) date_and_time: The date and time to transform.
+
+        Returns:
+            (Integer): timestamp in seconds.
+
+        """
+        if not date_and_time:
+            date_and_time = datetime.datetime.now()
+        return time.mktime(date_and_time.timetuple())
+
+    @staticmethod
+    def strike(text):
+        """
+        Create a strikethrough version of the given text.
+
+        Args:
+            (String) text: The text to strike.
+
+        Returns:
+            (String): the striked text.
+
+        """
+        result = ""
+        for c in str(text):
+            result += c + TerminalColors.STRIKE.value
+        return result
+
+    @staticmethod
+    def bold(text):
+        """
+        Create a bold version of the given text.
+        """
+        return TerminalColors.BOLD.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def underline(text):
+        """
+        Create a underlined version of the given text.
+        """
+        return TerminalColors.UNDERLINE.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def blue(text):
+        """
+        Create a blue version of the given text.
+        """
+        return TerminalColors.BLUE.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def red(text):
+        """
+        Create a red version of the given text.
+        """
+        return TerminalColors.RED.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def pink(text):
+        """
+        Create a red version of the given text.
+        """
+        return TerminalColors.PINK.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def yellow(text):
+        """
+        Create a yellow version of the given text.
+        """
+        return TerminalColors.YELLOW.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def green(text):
+        """
+        Create a red version of the given text.
+        """
+        return TerminalColors.GREEN.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def cyan(text):
+        """
+        Create a cyan version of the given text.
+        """
+        return TerminalColors.CYAN.value + text + TerminalColors.ENDC.value
+
+    @staticmethod
+    def log_json(logger: Logger, json_data: dict):
+        """
+        Write formatted JSON to the debug logger.
+
+        Args:
+            logger (Logger): The logger.
+            json_data (dict): The json object.
+        """
+        json_str = json.dumps(json_data, sort_keys=True, indent=2, separators=(",", ": "))
+        logger.debug(SimpleUtil.cyan(json_str))
+
+
+class TerminalColors(Enum):
+    """
+    Colors for formatting terminal output.
+    """
+
+    HEADER = "\033[95m"
+
+    RED = "\033[31m"
+    GREEN = "\033[32m"
+    YELLOW = "\033[33m"
+    BLUE = "\033[34m"
+    PINK = "\033[35m"
+    CYAN = "\033[36m"
+
+    WARNING = "\033[31m"
+    FAIL = "\033[41m"
+
+    BOLD = "\033[1m"
+    UNDERLINE = "\033[4m"
+    STRIKE = "\u0336"
+
+    ENDC = "\033[0m"