"""
===========================
Framework Utility Functions
===========================
Collection of utility functions shared by the ``vivarium`` framework.
"""
import functools
from bdb import BdbQuit
from importlib import import_module
from typing import Any, Callable, List, Tuple, Union
import numpy as np
import pandas as pd
[docs]def from_yearly(value, time_step):
return value * (time_step.total_seconds() / (60 * 60 * 24 * 365.0))
[docs]def to_yearly(value, time_step):
return value / (time_step.total_seconds() / (60 * 60 * 24 * 365.0))
[docs]def rate_to_probability(rate: Union[float, List, Tuple, np.ndarray, pd.Series]) -> np.ndarray:
# encountered underflow from rate > 30k
# for rates greater than 250, exp(-rate) evaluates to 1e-109
# beware machine-specific floating point issues
rate = np.array(rate)
rate[rate > 250] = 250.0
return 1 - np.exp(-rate)
[docs]def probability_to_rate(probability):
return -np.log(1 - probability)
[docs]def collapse_nested_dict(d, prefix=None):
results = []
for k, v in d.items():
cur_prefix = prefix + "." + k if prefix else k
if isinstance(v, dict):
results.extend(collapse_nested_dict(v, prefix=cur_prefix))
else:
results.append((cur_prefix, v))
return results
[docs]def import_by_path(path: str) -> Callable:
"""Import a class or function given it's absolute path.
Parameters
----------
path:
Path to object to import
"""
module_path, _, class_name = path.rpartition(".")
return getattr(import_module(module_path), class_name)
[docs]def handle_exceptions(func: Callable, logger: Any, with_debugger: bool) -> Callable:
"""Drops a user into an interactive debugger if func raises an error."""
@functools.wraps(func)
def wrapped(*args, **kwargs):
try:
return func(*args, **kwargs)
except (BdbQuit, KeyboardInterrupt):
raise
except Exception as e:
logger.exception("Uncaught exception {}".format(e))
if with_debugger:
import pdb
import traceback
traceback.print_exc()
pdb.post_mortem()
else:
raise
return wrapped