tools.py

"""This module provides a collection of convenience functions."""

import datetime
import time
import uuid
import json
import os
from functools import partial
from s3i import Directory
from s3i import Broker
from s3i import ServiceRequest
from s3i import TokenType
from s3i import IdentityProvider
from ml.app_logger import APP_LOGGER
from ml.authentication import GRANT_TYPES

IDENTITY_PROVIDER_URL = "https://idp.s3i.vswf.dev/"


class BColors:
    """colors for the console log"""

    HEADER = "\033[95m"
    OKBLUE = "\033[94m"
    OKGREEN = "\033[92m"
    WARNING = "\033[93m"
    FAIL = "\033[91m"
    ENDC = "\033[0m"
    BOLD = "\033[1m"
    UNDERLINE = "\033[4m"


def make_sub_thing(name, roles, features=[]):
    """
    Creates a dictionary representing this thing in it's current state
    as a subordinate thing. This representation should be used for
    subordinate things in s3i repository entries

    :param name: name of the sub thing
    :type name: str
    :param roles: fml40 roles of the sub thing
    :type roles: str
    :param features: fml40 features of the sub thing
    :type features: list

    :returns: JSON of the sub thing
    :rtype: dict

    """
    sub_thing = {
        "class": "ml40::Thing",
        "name": name,
        "roles": roles,
        "features": features
    }

    return sub_thing


def make_feature_config(class_name, identifier="", name="", subFeatures=""):
    """
    Creates a dictionary representing for its ml/fml40 features

    :param class_name: class name of the fml40 feature
    :type class_name: str
    :param identifier: local id of the fml40 feature
    :type identifier: str
    :param name: name of the fml40 feature
    :type name: str
    :param subFeatures: sub features
    :type subFeatures: dict

    :returns: JSON of the fml40 feature
    :rtype: dict

    """
    config_json = {
        "class": class_name
    }
    if identifier:
        config_json["identifier"] = identifier
    if name:
        config_json["name"] = name
    if subFeatures:
        config_json["subFeatures"] = subFeatures

    return config_json


def make_thing_config(thing_id, name, roles, features=[], config_path=""):
    """
    Creates a configuration file (JSON) for a fml40 thing.

    :param thing_id: identifier of the thing
    :type thing_id: str
    :param name: name of the thing
    :type name: str
    :param roles: fml40 roles of the thing
    :type roles: list
    :param features: fml40 features of the thing
    :type features: list
    :param config_path: path where the configuration file should be located
    :type config_path: str

    :returns: name of the configuration file of the thing
    :rtype: str


    """
    if not config_path:
        config_path = os.path.join("__file__", "configs")

    config_file = {
        "thingId": thing_id,
        "policyId": thing_id,
        "attributes": {
            "class": "ml40::Thing",
            "name": name,
            "roles": roles,
            "features": features,
        }
    }
    file_path = os.path.join(config_path, "{}.json".format(name))
    with open(file_path, 'wb') as file:
        file.write(json.dumps(config_file).encode('utf-8'))
    return "{}.json".format(name)


def load_config(config_filepath):
    """
    Loads a JSON object from a json formatted file found at config_filepath.

    :param config_filepath: Path to json formatted file.
    :type config_filepath: str

    """
    with open(config_filepath) as config_file:
        config = json.load(config_file)
        return config


def find_broker_endpoint(dir_obj, thing_id):
    """
    Finds the S3I-B endpoint of a thing

    :param dir_obj: S³I Directory Object
    :type dir_obj: object
    :param thing_id: identifier of the searched thing

    """
    thing_json = dir_obj.queryThingIDBased(thing_id)
    all_endpoints = thing_json["attributes"].get("allEndpoints", None)
    if all_endpoints:
        for ep in all_endpoints:
            if "s3ib" in ep:
                return ep


def remove_namespace(input_str):
    """
    Removes the specified namespace like ml40 or fml40

    :param input_str: input with namespace
    :type input_str: str
    :returns: output without namespace
    :rtype: str

    """
    return input_str.replace("fml40::", "").replace("ml40::", "")