Package ambianic

Expand source code
import os
from argparse import ArgumentParser
from typing import Union

import importlib_metadata as metadata
from dynaconf import Dynaconf, loaders
from dynaconf.utils.boxing import DynaBox

parser = ArgumentParser()
parser.add_argument("-c", "--config", help="Specify config YAML file location")

args, unknown = parser.parse_known_args()

DEFAULT_WORK_DIR: str = "/workspace"
DEFAULT_DATA_DIR: str = "./data"

DEFAULT_CONFIG_FILE: str = args.config or "config.yaml"
DEFAULT_SECRETS_FILE: str = "secrets.yaml"

__CONFIG_FILE: str = None
__SECRETS_FILE: str = None
__version__: str = metadata.version("ambianic-edge")


def get_config_file() -> str:
    return __CONFIG_FILE


def get_secrets_file() -> str:
    if __SECRETS_FILE:
        return __SECRETS_FILE
    return os.path.join(get_work_dir(), DEFAULT_SECRETS_FILE)


def __merge_secrets(config: Union[Dynaconf, DynaBox], src_config: Dynaconf = None):
    if src_config is None:
        src_config = config
    for key, val in config.items():
        if isinstance(val, dict):
            __merge_secrets(val, src_config)
            continue
        # NOTE value must be an exact match to avoid interfering
        # with other templates
        if isinstance(val, str) and (val[0:2] == "${" and val[-1] == "}"):
            ref_key = val[2:-1]
            ref_val = src_config.get(ref_key, None)
            if ref_val is not None:
                config[key] = ref_val
            continue


def __init_config() -> Dynaconf:
    config = Dynaconf(
        settings_files=[get_config_file(), get_secrets_file()],
        # secrets=[],
        merge=True,
        environments=False,
    )
    __merge_secrets(config)
    return config


def load_config(filename: str, clean: bool = False) -> Dynaconf:
    """Loads configuration settings from the given filename.
    If file_to_save is provided then consequent calls to save() will persist settings to the given file path.
    Otherwise save() will persist to the filename path (where the settings are to be loaded from)."""
    if clean:
        config.clean()
    if filename:
        config.load_file(path=[filename, get_secrets_file()])
        __merge_secrets(config)
        global __CONFIG_FILE
        __CONFIG_FILE = filename
    return config


def save_config():
    """Persist configuration settings to disk."""
    # ref: https://dynaconf.readthedocs.io/en/docs_223/guides/advanced_usage.html#exporting
    # ref: https://dynaconf.readthedocs.io/en/docs_223/reference/dynaconf.loaders.html#module-dynaconf.loaders.yaml_loader
    file_to_save = get_config_file()
    data = config.as_dict()
    loaders.write(file_to_save, DynaBox(data).to_dict())


def get_work_dir() -> str:
    """Retrieve the ambianic working directory"""
    env_work_dir = os.environ.get("AMBIANIC_DIR", os.getcwd())
    if not env_work_dir:
        env_work_dir = DEFAULT_WORK_DIR
    return env_work_dir


# initialization
server_instance = None

__CONFIG_FILE = os.path.join(get_work_dir(), DEFAULT_CONFIG_FILE)
config: Dynaconf = __init_config()

Sub-modules

ambianic.device

Base classes for an Ambianic Edge device abstraction

ambianic.logger

Logging functionalities wrapper

ambianic.notification

Utilities to send notifications.

ambianic.pipeline

Main module for Ambianic AI pipelines.

ambianic.server

Main Ambianic server module.

ambianic.util

Service classes for OS interaction and multithreading.

ambianic.webapp

Functions

def get_config_file() ‑> str
Expand source code
def get_config_file() -> str:
    return __CONFIG_FILE
def get_secrets_file() ‑> str
Expand source code
def get_secrets_file() -> str:
    if __SECRETS_FILE:
        return __SECRETS_FILE
    return os.path.join(get_work_dir(), DEFAULT_SECRETS_FILE)
def get_work_dir() ‑> str

Retrieve the ambianic working directory

Expand source code
def get_work_dir() -> str:
    """Retrieve the ambianic working directory"""
    env_work_dir = os.environ.get("AMBIANIC_DIR", os.getcwd())
    if not env_work_dir:
        env_work_dir = DEFAULT_WORK_DIR
    return env_work_dir
def load_config(filename: str, clean: bool = False) ‑> dynaconf.base.LazySettings

Loads configuration settings from the given filename. If file_to_save is provided then consequent calls to save() will persist settings to the given file path. Otherwise save() will persist to the filename path (where the settings are to be loaded from).

Expand source code
def load_config(filename: str, clean: bool = False) -> Dynaconf:
    """Loads configuration settings from the given filename.
    If file_to_save is provided then consequent calls to save() will persist settings to the given file path.
    Otherwise save() will persist to the filename path (where the settings are to be loaded from)."""
    if clean:
        config.clean()
    if filename:
        config.load_file(path=[filename, get_secrets_file()])
        __merge_secrets(config)
        global __CONFIG_FILE
        __CONFIG_FILE = filename
    return config
def save_config()

Persist configuration settings to disk.

Expand source code
def save_config():
    """Persist configuration settings to disk."""
    # ref: https://dynaconf.readthedocs.io/en/docs_223/guides/advanced_usage.html#exporting
    # ref: https://dynaconf.readthedocs.io/en/docs_223/reference/dynaconf.loaders.html#module-dynaconf.loaders.yaml_loader
    file_to_save = get_config_file()
    data = config.as_dict()
    loaders.write(file_to_save, DynaBox(data).to_dict())