Utility Functions#

Module containing supporting classes and functions used accross the project.

lymph.utils.check_unique_names(graph: dict)[source]#

Check all nodes in graph have unique names and no duplicate connections.

lymph.utils.check_spsn(spsn: list[float])[source]#

Check whether specificity and sensitivity are valid.

lymph.utils.comp_transition_tensor(num_parent: int, num_child: int, is_tumor_spread: bool, is_growth: bool, spread_prob: float, micro_mod: float) ndarray[source]#

Compute the transition factors of the edge.

The returned array is of shape (p,c,c), where p is the number of states of the parent node and c is the number of states of the child node.

Essentially, the tensors computed here contain most of the parametrization of the model. They are used to compute the transition matrix.

This function globally computes and caches the transition tensors, such that we do not need to worry about deleting and recomputing them when the parameters of the edge change.

lymph.utils.clinical(spsn: list) ndarray[source]#

Produce the confusion matrix of a clinical modality.

A clinical modality can by definition not detect microscopic metastases.

lymph.utils.pathological(spsn: list) ndarray[source]#

Produce the confusion matrix of a pathological modality.

A pathological modality can detect microscopic disease, but is unable to differentiante between micro- and macroscopic involvement.

lymph.utils.tile_and_repeat(mat: ndarray, tile: tuple[int, int], repeat: tuple[int, int]) ndarray[source]#

Apply the numpy functions tile and repeat successively to mat.

>>> mat = np.array([[1, 2], [3, 4]])
>>> tile_and_repeat(mat, (2, 2), (2, 2))
array([[1, 1, 2, 2, 1, 1, 2, 2],
       [1, 1, 2, 2, 1, 1, 2, 2],
       [3, 3, 4, 4, 3, 3, 4, 4],
       [3, 3, 4, 4, 3, 3, 4, 4],
       [1, 1, 2, 2, 1, 1, 2, 2],
       [1, 1, 2, 2, 1, 1, 2, 2],
       [3, 3, 4, 4, 3, 3, 4, 4],
       [3, 3, 4, 4, 3, 3, 4, 4]])
>>> tile_and_repeat(
...     mat=np.array([False, True], dtype=bool),
...     tile=(1, 2),
...     repeat=(1, 3),
... )
array([[False, False, False,  True,  True,  True, False, False, False,
         True,  True,  True]])
lymph.utils.get_state_idx_matrix(lnl_idx: int, num_lnls: int, num_states: int) ndarray[source]#

Return the indices for the transition tensor correpsonding to lnl_idx.

>>> get_state_idx_matrix(1, 3, 2)
array([[0, 0, 0, 0, 0, 0, 0, 0],
       [0, 0, 0, 0, 0, 0, 0, 0],
       [1, 1, 1, 1, 1, 1, 1, 1],
       [1, 1, 1, 1, 1, 1, 1, 1],
       [0, 0, 0, 0, 0, 0, 0, 0],
       [0, 0, 0, 0, 0, 0, 0, 0],
       [1, 1, 1, 1, 1, 1, 1, 1],
       [1, 1, 1, 1, 1, 1, 1, 1]])
>>> get_state_idx_matrix(1, 2, 3)
array([[0, 0, 0, 0, 0, 0, 0, 0, 0],
       [1, 1, 1, 1, 1, 1, 1, 1, 1],
       [2, 2, 2, 2, 2, 2, 2, 2, 2],
       [0, 0, 0, 0, 0, 0, 0, 0, 0],
       [1, 1, 1, 1, 1, 1, 1, 1, 1],
       [2, 2, 2, 2, 2, 2, 2, 2, 2],
       [0, 0, 0, 0, 0, 0, 0, 0, 0],
       [1, 1, 1, 1, 1, 1, 1, 1, 1],
       [2, 2, 2, 2, 2, 2, 2, 2, 2]])
lymph.utils.row_wise_kron(a: ndarray, b: ndarray) ndarray[source]#

Compute the kronecker product of two matrices row-wise.

>>> a = np.array([[1, 2], [3, 4]])
>>> b = np.array([[5, 6], [7, 8]])
>>> row_wise_kron(a, b)
array([[ 5.,  6., 10., 12.],
       [21., 24., 28., 32.]])
lymph.utils.early_late_mapping(t_stage: int | str) str[source]#

Map the reported T-category (i.e., 1, 2, 3, 4) to “early” and “late”.

lymph.utils.trigger(func: callable) callable[source]#

Decorator that runs instance’s trigger_callbacks when called.

class lymph.utils.smart_updating_dict_cached_property(func)[source]#

Bases: cached_property

Allows setting/deleting dict-like attrs by updating/clearing them.

lymph.utils.dict_to_func(mapping: dict[Any, Any]) callable[source]#

Transform a dictionary into a function.

>>> char_map = {'a': 1, 'b': 2, 'c': 3}
>>> char_map = dict_to_func(char_map)
>>> char_map('a')
1
lymph.utils.popfirst(seq: Sequence[Any]) tuple[Any, Sequence[Any]][source]#

Return the first element of a sequence and the sequence without it.

If the sequence is empty, the first element will be None and the second just the empty sequence. Example:

>>> popfirst([1, 2, 3])
(1, [2, 3])
>>> popfirst([])
(None, [])
lymph.utils.flatten(mapping, parent_key='', sep='_') dict[source]#

Flatten a nested dictionary.

>>> flatten({"a": {"b": 1, "c": 2}, "d": 3})
{'a_b': 1, 'a_c': 2, 'd': 3}
lymph.utils.unflatten_and_split(mapping: dict, expected_keys: list[str], sep: str = '_') tuple[dict, dict][source]#

Unflatten the part of a dict containing expected_keys and return the rest.

>>> unflatten_and_split({'a_b': 1, 'a_c_x': 2, 'd_y': 3}, expected_keys=['a'])
({'a': {'b': 1, 'c_x': 2}}, {'d_y': 3})
lymph.utils.get_params_from(objects: dict[str, HasGetParams], as_dict: bool = True, as_flat: bool = True) Iterable[float] | dict[str, float][source]#

Get the parameters from each get_params() method of the objects.

lymph.utils.set_params_for(objects: dict[str, HasSetParams], *args: float, **kwargs: float) tuple[float][source]#

Pass arguments to each set_params() method of the objects.

lymph.utils.safe_set_params(model: ModelT, params: Iterable[float] | dict[str, float] | None = None) None[source]#

Set the params of the model.

This infers whether params is a dict or a list and calls the model’s method set_params() accordingly.

lymph.utils.synchronize_params(get_from: dict[str, HasGetParams], set_to: dict[str, HasSetParams]) None[source]#

Get the parameters from one object and set them to another.

lymph.utils.draw_diagnoses(diagnose_times: list[int], state_evolution: ndarray, observation_matrix: ndarray, possible_diagnoses: ndarray, rng: Generator | None = None, seed: int = 42) ndarray[source]#

Given the diagnose_times and a hidden state_evolution, draw diagnoses.

lymph.utils.add_or_mult(llh: float, arr: ndarray, log: bool = True) float[source]#

Add or multiply the log-likelihood with the given array.