Skip to content

utils

Attributes

logger = structlog.get_logger() module-attribute

CAMEL_TO_SNAKE_REGEX = re.compile('(?<!^)(?=[A-Z])') module-attribute

WORD_REGEX_PATTERN = re.compile('[^A-Za-z]+') module-attribute

SUBCLASS_TYPE = TypeVar('SUBCLASS_TYPE') module-attribute

Classes

Functions

is_debug() -> bool

Source code in src/kiara/utils/__init__.py 
28
29
30
31
32
33
def is_debug() -> bool:
    debug = os.environ.get("DEBUG", "")
    if debug.lower() == "true":
        return True
    else:
        return False

is_develop() -> bool

Source code in src/kiara/utils/__init__.py 
36
37
38
39
40
41
42
43
44
def is_develop() -> bool:
    develop = os.environ.get("DEVELOP", "")
    if not develop:
        develop = os.environ.get("DEV", "")

    if develop and develop.lower() != "false":
        return True

    return False

is_emscripten() -> bool

Check if we're running in Emscripten / Pyodide environment.

Source code in src/kiara/utils/__init__.py 
47
48
49
50
def is_emscripten() -> bool:
    """Check if we're running in Emscripten / Pyodide environment."""

    return sys.platform == "emscripten"

get_dev_config() -> KiaraDevSettings

Source code in src/kiara/utils/__init__.py 
53
54
55
56
def get_dev_config() -> "KiaraDevSettings":
    from kiara.utils.develop import KIARA_DEV_SETTINGS

    return KIARA_DEV_SETTINGS

is_jupyter() -> bool

Source code in src/kiara/utils/__init__.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
def is_jupyter() -> bool:
    try:
        get_ipython  # type: ignore
    except NameError:
        return False
    ipython = get_ipython()  # type: ignore  # noqa
    shell = ipython.__class__.__name__
    if shell == "TerminalInteractiveShell":
        return False
    elif "google.colab" in str(ipython.__class__) or shell == "ZMQInteractiveShell":
        return True
    else:
        return False

log_exception(exc: Exception)

Source code in src/kiara/utils/__init__.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def log_exception(exc: Exception):
    if is_debug():
        logger.error(exc)

    if is_develop():
        from kiara.utils.develop import DetailLevel

        config = get_dev_config()
        if config.log.exc in [DetailLevel.NONE, "none"]:
            return

        show_locals = config.log.exc in [DetailLevel.FULL, "full"]

        from kiara.interfaces import get_console
        from kiara.utils.develop import log_dev_message

        # exc_info = sys.exc_info()
        exc_info = (type(exc), exc, exc.__traceback__)

        # if not exc_info or not exc_info[0]:
        #     # TODO: create exc_info from exception?
        #     if not is_debug():
        #         logger.error(exc)
        # else:
        console = get_console()
        from rich.traceback import Traceback

        log_dev_message(
            Traceback.from_exception(
                exc_info[0],
                exc_info[1],
                traceback=exc_info[2],
                show_locals=show_locals,
                width=console.width - 4,  # type: ignore
            ),
            title="Exception details",
        )

log_message(msg: str, **data)

Source code in src/kiara/utils/__init__.py 
113
114
115
def log_message(msg: str, **data):
    if is_debug():
        logger.debug(msg, **data)

get_auto_workflow_alias(module_type: str, use_incremental_ids: bool = False) -> str

Return an id for a workflow obj of a provided module class.

If 'use_incremental_ids' is set to True, a unique id is returned.

module_type (str): the name of the module type
use_incremental_ids (bool): whether to return a unique (incremental) id

str: a module id
Source code in src/kiara/utils/__init__.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_auto_workflow_alias(module_type: str, use_incremental_ids: bool = False) -> str:
    """
    Return an id for a workflow obj of a provided module class.

    If 'use_incremental_ids' is set to True, a unique id is returned.

    Args:
    ----
        module_type (str): the name of the module type
        use_incremental_ids (bool): whether to return a unique (incremental) id

    Returns:
    -------
        str: a module id
    """
    if not use_incremental_ids:
        return module_type

    nr = _AUTO_MODULE_ID.setdefault(module_type, 0)
    _AUTO_MODULE_ID[module_type] = nr + 1

    return f"{module_type}_{nr}"

camel_case_to_snake_case(camel_text: str, repl: str = '_') -> str

Source code in src/kiara/utils/__init__.py 
147
148
def camel_case_to_snake_case(camel_text: str, repl: str = "_") -> str:
    return CAMEL_TO_SNAKE_REGEX.sub(repl, camel_text).lower()

to_camel_case(text: str) -> str

Source code in src/kiara/utils/__init__.py 
151
152
153
def to_camel_case(text: str) -> str:
    words = WORD_REGEX_PATTERN.split(text)
    return "".join(w.title() for i, w in enumerate(words))

check_valid_field_names(*field_names) -> List[str]

Check whether the provided field names are all valid.

an iterable of strings with invalid field names
Source code in src/kiara/utils/__init__.py 
172
173
174
175
176
177
178
179
180
def check_valid_field_names(*field_names) -> List[str]:
    """
    Check whether the provided field names are all valid.

    Returns:
    -------
        an iterable of strings with invalid field names
    """
    return [x for x in field_names if x in INVALID_VALUE_NAMES or x.startswith("_")]

find_free_id(stem: str, current_ids: Iterable[str], sep='_') -> str

Find a free var (or other name) based on a stem string, based on a list of provided existing names.

stem (str): the base string to use
current_ids (Iterable[str]): currently existing names
method (str): the method to create new names (allowed: 'count' -- for now)
method_args (dict): prototing_config for the creation method

str: a free name
Source code in src/kiara/utils/__init__.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def find_free_id(
    stem: str,
    current_ids: Iterable[str],
    sep="_",
) -> str:
    """
    Find a free var (or other name) based on a stem string, based on a list of provided existing names.

    Args:
    ----
        stem (str): the base string to use
        current_ids (Iterable[str]): currently existing names
        method (str): the method to create new names (allowed: 'count' -- for now)
        method_args (dict): prototing_config for the creation method

    Returns:
    -------
        str: a free name
    """
    start_count = 1
    if stem not in current_ids:
        return stem

    i = start_count

    # new_name = None
    while True:
        new_name = f"{stem}{sep}{i}"
        if new_name in current_ids:
            i = i + 1
            continue
        break
    return new_name

first_line(text: str)

Source code in src/kiara/utils/__init__.py 
218
219
220
221
222
def first_line(text: str):
    if "\n" in text:
        return text.split("\n")[0].strip()
    else:
        return text