Source code for ska_ser_skallop.utils.env

import atexit
import cProfile
import logging
import os
import pstats
from contextlib import contextmanager
from importlib import util
from tempfile import TemporaryDirectory

# from typing import Literal
from typing import Literal, Union

from ska_ser_skallop.utils.singleton import Singleton

logger = logging.getLogger(__name__)


def get_temp_dir(prefix: Union[str, None] = None) -> str:
    """generate a path to a directory deleted at the end of the program.

    :param prefix: a prefix to start the name of the directory with, defaults to None
    :return: string representing the path to the directory
    """
    dir = TemporaryDirectory(prefix=prefix)
    atexit.register(dir.cleanup)
    return dir.name


def build_in_testing() -> bool:
    try:
        tango = util.find_spec("tango")
    except ValueError:
        return False
    test_env = os.getenv("TEST_ENV")
    if test_env == "BUILD_OUT":
        return False
    if test_env == "BUILD_IN":
        return True
    return tango is not None


def i_can_import(module_name: str):
    try:
        module = util.find_spec(module_name)
    except (ModuleNotFoundError, ValueError):
        return False
    return module is not None


def get_tmc_central_node_fqdn() -> str:
    name = os.getenv("CENTRALNODE_FQDN")
    if name:
        return name
    return "ska_mid/tm_central/central_node"


def get_tmc_subarray_node_fqdn() -> str:
    name = os.getenv("SUBARRAY")
    if name:
        return name
    return "ska_mid/tm_subarray_node"


def get_entry_point_from_env() -> Union[str, None]:
    return os.getenv("ENTRY_POINT")


def telescope_type_is_mid() -> bool:
    container = TelescopeTypeContainer()
    return container.is_type_mid()


def telescope_type_is_low() -> bool:
    container = TelescopeTypeContainer()
    return not container.is_type_mid()


def set_telescope_type_from_env() -> Literal["skalow", "skamid"]:
    if telescope_set_as_low_from_env():
        set_telescope_as_low()
        return "skalow"
    set_telescope_as_mid()
    return "skamid"


def set_telescope_as_mid() -> None:
    container = TelescopeTypeContainer()
    container.set_telescope_type_as_mid()


def set_telescope_as_low() -> None:
    container = TelescopeTypeContainer()
    container.set_telescope_type_as_low()


def is_telescope_type_already_set() -> bool:
    container = TelescopeTypeContainer()
    return container.is_type_set


def telescope_set_as_low_from_env() -> bool:
    env_values = {
        "SKA_TELESCOPE": os.getenv("SKA_TELESCOPE"),
        "TEL": os.getenv("TEL"),
    }
    conditions_for_low = [
        env_values["SKA_TELESCOPE"] == "SKA-Low",
        env_values["TEL"] == "low",
    ]
    conditions_for_mid = [
        env_values["SKA_TELESCOPE"] == "SKA-Mid",
        env_values["TEL"] == "mid",
    ]
    if any(conditions_for_mid) and any(conditions_for_low):
        logger.warning(
            "you have an inconsistency in env variables for determining telescope "
            f"type: will default to setting it as Mid: {env_values}"
        )
        return False
    return any(conditions_for_low)


class TelescopeTypeContainer(metaclass=Singleton):
    def __init__(self) -> None:
        if telescope_set_as_low_from_env():
            self._type = "skalow"
        else:
            self._type = "skamid"

    @property
    def is_type_set(self) -> bool:
        return self._type != "NONE"

    def is_type_mid(self) -> bool:
        return self._type == "skamid"

    def is_type_low(self) -> bool:
        return self._type == "skalow"

    def set_telescope_type_as_low(self):
        self._type = "skalow"

    def set_telescope_type_as_mid(self):
        self._type = "skamid"


@contextmanager
def profile_it():
    with cProfile.Profile() as profile:
        yield
        stats = pstats.Stats(profile)
        stats.sort_stats(pstats.SortKey.CUMULATIVE)
        stats.print_stats(0.1)