utils

Utility functions and helpers.

phylogenetic_trees

Auxiliary functions for phylogentic trees.

Phylogenetic trees are usually defined as trees where every inner node has at least two children. However, the class Tree does not force this property. Simulated trees are often 'planted', i.e., the root has only a single child, and this edge represents the ancestral lineage.

node_to_str(node: TreeNode) -> str

String representation of a node in a phylogenetic tree.

Parameters:

Name	Type	Description	Default
node	TreeNode	A TreeNode object representing a node in a phylogenetic tree. The node is expected to have attributes 'label', 'reconc', and 'dist'.	required

Returns:

Type	Description
str	String representation of the node including reconc (if defined) and dist.

Source code in asymmetree/utils/phylogenetic_trees.py

def node_to_str(node: TreeNode) -> str:
    """String representation of a node in a phylogenetic tree.

    Args:
        node: A TreeNode object representing a node in a phylogenetic tree. The node is expected to
            have attributes 'label', 'reconc', and 'dist'.

    Returns:
        String representation of the node including reconc (if defined) and dist.
    """
    if hasattr(node, "reconc") and isinstance(node.reconc, (tuple, list)):
        return "{}<{}-{}>:{}".format(node.label, *node.reconc, node.dist)
    elif hasattr(node, "reconc") and node.reconc:
        return "{}<{}>:{}".format(node.label, node.reconc, node.dist)
    else:
        return "{}:{}".format(node.label, node.dist)

sorted_nodes(tree: Tree, oldest_to_youngest: bool = True) -> list[TreeNode]

List of nodes sorted by timestamp.

Return a list of all nodes in the tree sorted by time stamp beginning with the oldest (highest time stamp). Optionally the order can be reversed by setting oldest_to_youngest to False.

Parameters:

Name	Type	Description	Default
tree	Tree	A phylogenetic tree with nodes that have a tstamp attribute.	required
oldest_to_youngest	bool	If True, the nodes are sorted from oldest to youngest, otherwise from youngest to oldest.	True

Returns:

Type	Description
list[TreeNode]	Nodes sorted by their timestamp.

Source code in asymmetree/utils/phylogenetic_trees.py

def sorted_nodes(tree: Tree, oldest_to_youngest: bool = True) -> list[TreeNode]:
    """List of nodes sorted by timestamp.

    Return a list of all nodes in the tree sorted by time stamp beginning with the oldest (highest
    time stamp). Optionally the order can be reversed by setting `oldest_to_youngest` to False.

    Args:
        tree: A phylogenetic tree with nodes that have a `tstamp` attribute.
        oldest_to_youngest: If True, the nodes are sorted from oldest to youngest, otherwise from
            youngest to oldest.

    Returns:
        Nodes sorted by their timestamp.
    """
    return sorted(tree.preorder(), key=lambda node: node.tstamp, reverse=oldest_to_youngest)

sorted_edges(tree: Tree) -> list[tuple[TreeNode, TreeNode]]

List of edges (u,v) sorted by timestamp of u.

Return a list of all edges (u,v) sorted by timestamp of u beginning with the oldest (highest time stamp).

Parameters:

Name	Type	Description	Default
tree	Tree	A phylogenetic tree with nodes that have a tstamp attribute.	required

Returns:

Type	Description
list[tuple[TreeNode, TreeNode]]	Edges sorted by their timestamp.

Source code in asymmetree/utils/phylogenetic_trees.py

def sorted_edges(tree: Tree) -> list[tuple[TreeNode, TreeNode]]:
    """List of edges (u,v) sorted by timestamp of u.

    Return a list of all edges (u,v) sorted by timestamp of u beginning with the oldest (highest
    time stamp).

    Args:
        tree: A phylogenetic tree with nodes that have a `tstamp` attribute.

    Returns:
        Edges sorted by their timestamp.
    """
    return sorted(tree.edges(), key=lambda e: (e[0].tstamp, e[1].tstamp), reverse=True)

distance_from_timing(tree: Tree) -> None

Adjusts all distances according to the time stamp difference.

The distance of an edge (u,v) is set to the absolute difference of the time stamps of u and v, i.e., dist = abs(u.tstamp - v.tstamp).

Parameters:

Name	Type	Description	Default
tree	Tree	A phylogenetic tree with nodes that have a tstamp attribute.	required

Source code in asymmetree/utils/phylogenetic_trees.py

def distance_from_timing(tree: Tree) -> None:
    """Adjusts all distances according to the time stamp difference.

    The distance of an edge (u,v) is set to the absolute difference of the time stamps of u and v,
    i.e., `dist = abs(u.tstamp - v.tstamp)`.

    Args:
        tree: A phylogenetic tree with nodes that have a `tstamp` attribute.
    """
    if tree.root:
        tree.root.dist = 0.0

    for u, v in tree.edges():
        v.dist = abs(u.tstamp - v.tstamp)

delete_and_reconnect(tree: Tree, node: TreeNode, add_distances: bool = True, keep_transferred: bool = True) -> TreeNode | None

Delete a node from the tree and reconnect its parent and children.

Parameters:

Name	Type	Description	Default
tree	Tree	A phylogenetic tree.	required
node	TreeNode	The node in the tree to be deleted.	required
add_distances	bool	When the node v is deleted and its children are reconnected to its parent, add to the dist parameter of the children v.dist, i.e., the distance of v from its parent.	True
keep_transferred	bool	When the edge of the deleted node v from its parent u was a transfer edge, make all edges from u to the children of v transfer edges.	True

Returns:

Type	Description
TreeNode | None	The parent of the node, if it could be deleted, or None, if the node could not be deleted, i.e., it has no parent.

Source code in asymmetree/utils/phylogenetic_trees.py

def delete_and_reconnect(
    tree: Tree,
    node: TreeNode,
    add_distances: bool = True,
    keep_transferred: bool = True,
) -> TreeNode | None:
    """Delete a node from the tree and reconnect its parent and children.

    Args:
        tree: A phylogenetic tree.
        node: The node in the tree to be deleted.
        add_distances: When the node v is deleted and its children are reconnected to its parent,
            add to the `dist` parameter of the children `v.dist`, i.e., the distance of v from its
            parent.
        keep_transferred: When the edge of the deleted node v from its parent u was a transfer edge,
            make all edges from u to the children of v transfer edges.

    Returns:
        The parent of the node, if it could be deleted, or None, if the node could not be deleted,
            i.e., it has no parent.
    """
    if not node.parent:
        return None

    for child in node.children:
        if add_distances and hasattr(node, "dist") and hasattr(child, "dist"):
            child.dist += node.dist
        if keep_transferred and hasattr(node, "transferred") and node.transferred:
            child.transferred = 1

    return tree.delete_and_reconnect(node)

add_planted_root(tree: Tree) -> TreeNode

Add a new root that has the original root as its single child.

A tree is called planted if the root has only a single child. This is often the case for simulated trees where the edge from the planted root to its child represents the ancestral lineage. This function adds a planted root to a tree that does not have one yet.

Parameters:

Name	Type	Description	Default
tree	Tree	A phylogenetic tree.	required

Returns:

Type	Description
TreeNode	The newly added planted root.

Source code in asymmetree/utils/phylogenetic_trees.py

def add_planted_root(tree: Tree) -> TreeNode:
    """Add a new root that has the original root as its single child.

    A tree is called planted if the root has only a single child. This is often the case for
    simulated trees where the edge from the planted root to its child represents the ancestral
    lineage. This function adds a planted root to a tree that does not have one yet.

    Args:
        tree: A phylogenetic tree.

    Returns:
        The newly added planted root.
    """
    old_root = tree.root
    tree.root = TreeNode()

    for k, v in DEFAULT_ATTRIBUTES.items():
        setattr(tree.root, k, v)

    if old_root:
        tree.root.add_child(old_root)

    return tree.root

reconc_sorted_leaves(tree: Tree, return_list: bool = False) -> dict | tuple[dict, list]

Sort leaves by their reconciliation attribute.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with nodes that have a reconc attribute.	required
return_list	bool	If True, also return a list of leaves such that leaves of the same reconciliation appear as a coherent sequence.	False

Returns:

Type	Description
dict | tuple[dict, list]	A dictionary with reconciliations as keys and a list of corresponding nodes as values. If return_list is True, also return a list of leaves such that leaves of the same reconciliation appear as a coherent sequence.

Source code in asymmetree/utils/phylogenetic_trees.py

def reconc_sorted_leaves(tree: Tree, return_list: bool = False) -> dict | tuple[dict, list]:
    """Sort leaves by their reconciliation attribute.

    Args:
        tree: A tree with nodes that have a `reconc` attribute.
        return_list: If True, also return a list of leaves such that leaves of the same
            reconciliation appear as a coherent sequence.

    Returns:
        A dictionary with reconciliations as keys and a list of corresponding nodes as values. If
            `return_list` is True, also return a list of leaves such that leaves of the same
            reconciliation appear as a coherent sequence.
    """
    reconc_dict = {}

    for leaf in tree.leaves():
        if leaf.reconc not in reconc_dict:
            reconc_dict[leaf.reconc] = []
        reconc_dict[leaf.reconc].append(leaf)

    if not return_list:
        return reconc_dict

    leaves = []
    for leaf_list in reconc_dict.values():
        for leaf in leaf_list:
            leaves.append(leaf)

    return reconc_dict, leaves

distance_matrix(tree: Tree, leaf_order: list[TreeNode] | None = None) -> tuple[list[TreeNode], np.ndarray]

Distance matrix on the leaf set of the phylogenetic tree.

Computes a distance matrix on the set of leaves of the tree where each distances is the sum of the distances (dist) on the path connecting the pair of leaves.

Additionally a list of leaves corresponding to the indices in the matrix is returned.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with nodes that have a dist attribute.	required
leaf_order	list[TreeNode] | None	A list of all leaves in the tree defining the indices for the matrix. The default is None, in which case leaves are indexed in sibling order.	None

Returns:

Type	Description
list[TreeNode]	A list of TreeNode objects representing the order for the lines/columns in the distance matrix.
ndarray	An array representing the distance matrix.

Raises:

Type	Description
ValueError	If leaf_order is provided and does not match with the leaves in the tree.

Source code in asymmetree/utils/phylogenetic_trees.py

def distance_matrix(
    tree: Tree,
    leaf_order: list[TreeNode] | None = None,
) -> tuple[list[TreeNode], np.ndarray]:
    """Distance matrix on the leaf set of the phylogenetic tree.

    Computes a distance matrix on the set of leaves of the tree where each distances is the sum of
    the distances (`dist`) on the path connecting the pair of leaves.

    Additionally a list of leaves corresponding to the indices in the matrix is returned.

    Args:
        tree: A tree with nodes that have a `dist` attribute.
        leaf_order: A list of all leaves in the tree defining the indices for the matrix. The
            default is None, in which case leaves are indexed in sibling order.

    Returns:
        A list of TreeNode objects representing the order for the lines/columns in the distance
            matrix.
        An array representing the distance matrix.

    Raises:
        ValueError: If `leaf_order` is provided and does not match with the leaves in the tree.
    """
    distance_dict = distances_from_root(tree)
    leaves = tree.leaf_dict()

    if leaf_order:
        if set(leaf_order) != set(leaves[tree.root]):
            raise ValueError("ordered leaf list does not match with the leaves in the tree")
        L = leaf_order
    else:
        # leaves in sibling order
        L = leaves[tree.root]

    leaf_index = {leaf: i for i, leaf in enumerate(L)}

    D = np.zeros((len(L), len(L)), dtype=np.float)

    for v in tree.preorder():
        if not v.children:
            continue

        for c1, c2 in itertools.combinations(v.children, 2):
            for x in leaves[c1]:
                x_index = leaf_index[x]
                x_dist = distance_dict[x] - distance_dict[v]
                for y in leaves[c2]:
                    y_index = leaf_index[y]
                    y_dist = distance_dict[y] - distance_dict[v]
                    D[x_index, y_index] = x_dist + y_dist
                    D[y_index, x_index] = x_dist + y_dist

    return L, D

distances_from_root(tree: Tree) -> dict[TreeNode, float]

The distances of each node to the root of the tree.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with nodes that have a dist attribute.	required

Returns:

Type	Description
dict[TreeNode, float]	A dictionary where the keys are TreeNode objects and the values are their distances (sum of dist) to the root.

Source code in asymmetree/utils/phylogenetic_trees.py

def distances_from_root(tree: Tree) -> dict[TreeNode, float]:
    """The distances of each node to the root of the tree.

    Args:
        tree: A tree with nodes that have a `dist` attribute.

    Returns:
        A dictionary where the keys are TreeNode objects and the values are their distances (sum of
            `dist`) to the root.
    """
    distance_dict = {}

    for v in tree.preorder():
        if not v.parent:
            distance_dict[v] = 0.0
        else:
            depth = distance_dict[v.parent] + v.dist
            distance_dict[v] = depth

    return distance_dict

topology_only(tree: Tree, inplace: bool = True) -> Tree

Reset reconciliations, distances, time stamps, transfer status, and inner labels.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree.	required
inplace	bool	If True, reset the attributes of this tree instance, otherwise make a copy first and modify the copy.	True

Returns:

Type	Description
Tree	The original or a copy of the tree instance with reset attributes.

Source code in asymmetree/utils/phylogenetic_trees.py

def topology_only(tree: Tree, inplace: bool = True) -> Tree:
    """Reset reconciliations, distances, time stamps, transfer status, and inner labels.

    Args:
        tree: A tree.
        inplace: If True, reset the attributes of this tree instance, otherwise make a copy first
            and modify the copy.

    Returns:
        The original or a copy of the tree instance with reset attributes.
    """
    if not inplace:
        T = tree.copy()
    else:
        T = tree

    for v in T.preorder():
        if v.children:
            v.label = ""
            v.reconc = None
        v.dist = 1.0
        v.tstamp = None
        v.transferred = 0

    return T

count_node_types(tree: Tree) -> dict[str, int]

Count speciations, duplication, losses, HGTs and surviving genes.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with nodes that have an event attribute.	required

Returns:

Type	Description
dict[str, int]	A dictionary with the event counts as values. Keys are 'S' (speciations), 'D' (duplication), 'L' (losses), 'H' (HGTs) and 'extant' (surviving genes).

Source code in asymmetree/utils/phylogenetic_trees.py

def count_node_types(tree: Tree) -> dict[str, int]:
    """Count speciations, duplication, losses, HGTs and surviving genes.

    Args:
        tree: A tree with nodes that have an `event` attribute.

    Returns:
        A dictionary with the event counts as values. Keys are 'S' (speciations), 'D' (duplication),
            'L' (losses), 'H' (HGTs) and 'extant' (surviving genes).
    """
    counts = {"S": 0, "D": 0, "L": 0, "H": 0, "extant": 0}

    for v in tree.preorder():
        if not v.children:
            if v.event == "L":
                counts["L"] += 1
            else:
                counts["extant"] += 1

        elif v.event == "S":
            counts["S"] += 1
        elif v.event == "D":
            counts["D"] += 1
        elif v.event == "H":
            counts["H"] += 1

    return counts

random_colored_tree(n: int, colors: int | list, binary: bool = False, force_all_colors: bool = False) -> Tree

Create a random colored tree.

The number of leaves and the reconciliation labels are specified in the parameters n and colors, respectively. Each non-leaf node in the resulting tree will have at least two children (property of phylogenetic trees). The colors (attribute reconc) are assigned to the leaves at random. If force_all_colors is True, the resulting tree is guaranteed to have at least one leaf of each reconciliation.

Parameters:

Name	Type	Description	Default
n	int	The desired number of leaves.	required
colors	int | list	The list of recociliations, or the desired number in which case the reconciliations {1, ..., colors} are used.	required
binary	bool	If True, forces the tree to be binary (the default is False).	False
force_all_colors	bool	If True, the resulting tree is guaranteed to have at least one leaf of each reconciliation (the default is False).	False

Returns:

Type	Description
Tree	A random tree with n leaves to which the reconc attribute is assigned at random.

Raises:

Type	Description
TypeError	If n is not an integer > 0.
ValueError	If the number of colors is greater than n and force_all_colors is true.

Source code in asymmetree/utils/phylogenetic_trees.py

def random_colored_tree(
    n: int,
    colors: int | list,
    binary: bool = False,
    force_all_colors: bool = False,
) -> Tree:
    """Create a random colored tree.

    The number of leaves and the reconciliation labels are specified in the parameters n and colors,
    respectively. Each non-leaf node in the resulting tree will have at least two children (property
    of phylogenetic trees).
    The colors (attribute `reconc`) are assigned to the leaves at random. If `force_all_colors` is
    True, the resulting tree is guaranteed to have at least one leaf of each reconciliation.

    Args:
        n: The desired number of leaves.
        colors: The list of recociliations, or the desired number in which case the reconciliations
            {1, ..., colors} are used.
        binary: If True, forces the tree to be binary (the default is False).
        force_all_colors: If True, the resulting tree is guaranteed to have at least one leaf of
            each reconciliation (the default is False).

    Returns:
        A random tree with n leaves to which the `reconc` attribute is assigned at random.

    Raises:
        TypeError: If n is not an integer > 0.
        ValueError: If the number of colors is greater than n and `force_all_colors` is true.
    """
    tree = Tree.random_tree(n, binary=binary)

    if isinstance(colors, int):
        colors = [i + 1 for i in range(colors)]
    elif not isinstance(colors, collections.abc.Iterable):
        raise TypeError("'colors' must be of type 'int' or iterable")

    if len(colors) > n and force_all_colors:
        raise ValueError("cannot force all colors since #colors > n")

    leaves = [leaf for leaf in tree.leaves()]

    if force_all_colors:
        # use every color at least once
        permutation = np.random.permutation(len(leaves))
        for i in range(len(leaves)):
            if i < len(colors):
                leaves[permutation[i]].reconc = colors[i]
            else:
                # color the remaining leaves randomly
                leaves[permutation[i]].reconc = random.choice(colors)
    else:
        # assign colors completely randomly
        for leaf in leaves:
            leaf.reconc = random.choice(colors)

    return tree

random_ultrametric_timing(tree: Tree, inplace: bool = False, adjust_distances: bool = False) -> Tree

Generate a random ultrametric timing for the tree.

Parameters:

Name	Type	Description	Default
tree	Tree	The tree for which a random timing shall be generated.	required
inplace	bool	If True, the input tree is modified, otherwise a copy is returned.	False
adjust_distances	bool	If True, also adjust the dist attribute of the tree nodes to match the differences of the tstamp values.	False

Returns:

Type	Description
Tree	A random tree whose nodes have tstamp attributes that represent the generated random ultrametric timing.

Source code in asymmetree/utils/phylogenetic_trees.py

def random_ultrametric_timing(
    tree: Tree, inplace: bool = False, adjust_distances: bool = False
) -> Tree:
    """Generate a random ultrametric timing for the tree.

    Args:
        tree: The tree for which a random timing shall be generated.
        inplace: If True, the input tree is modified, otherwise a copy is returned.
        adjust_distances: If True, also adjust the dist attribute of the tree nodes to match the
            differences of the `tstamp` values.

    Returns:
        A random tree whose nodes have `tstamp` attributes that represent the generated random
            ultrametric timing.
    """
    if not inplace:
        tree = tree.copy()

    for v in tree.preorder():
        if not v.children:
            v.tstamp = 0.0
        elif not v.parent:
            v.tstamp = 1.0
        else:  # random walk to a leaf
            pos = v  # current position
            length = 0  # path length |P|
            while pos.children:
                length += 1
                pos = pos.children[np.random.randint(len(pos.children))]
            v.tstamp = v.parent.tstamp * (1 - 2 * np.random.uniform() / (length + 1))

    if adjust_distances:
        distance_from_timing(tree)

    return tree

phylo_tree_attributes(tree: Tree, inplace: bool = True) -> Tree

Add the attributes for a phylogentic tree if not already set.

Parameters:

Name	Type	Description	Default
tree	Tree	The tree for which attributes shall be added.	required
inplace	bool	If True, the input tree is modified, otherwise a copy is returned.	True

Returns:

Type	Description
Tree	The tree with added attributes.

Source code in asymmetree/utils/phylogenetic_trees.py

def phylo_tree_attributes(tree: Tree, inplace: bool = True) -> Tree:
    """Add the attributes for a phylogentic tree if not already set.

    Args:
        tree: The tree for which attributes shall be added.
        inplace: If True, the input tree is modified, otherwise a copy is returned.

    Returns:
        The tree with added attributes.
    """
    if not inplace:
        tree = tree.copy()

    for v in tree.preorder():
        for key, value in DEFAULT_ATTRIBUTES.items():
            if not hasattr(v, key):
                setattr(tree.root, key, value)

    return tree

assign_missing_labels(tree: Tree) -> None

Assign integer label to nodes with no or missing label.

Also assign the number_of_species attribute (i.e., the number of leaves) to the tree.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with some nodes that have no or missing label.	required

Source code in asymmetree/utils/phylogenetic_trees.py

def assign_missing_labels(tree: Tree) -> None:
    """Assign integer label to nodes with no or missing label.

    Also assign the `number_of_species` attribute (i.e., the number of leaves) to the tree.

    Args:
        tree: A tree with some nodes that have no or missing label.
    """
    tree.number_of_species = 0
    labels = set()

    for v in tree.preorder():
        if not v.children:
            tree.number_of_species += 1

        if hasattr(v, "label") and v.label:
            labels.add(v.label)

    # assign new labels to remaining nodes
    current_label = 0
    for v in tree.preorder():
        if not hasattr(v, "label") or not v.label:
            while current_label in labels:
                current_label += 1
            v.label = current_label
            labels.add(current_label)

reconstruct_reconc_from_graph(tree: Tree, G: nx.Graph | nx.DiGraph) -> None

Reconstruct the reconciliations from a NetworkX Graph.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree for which reconciliations shall be reconstructed.	required
G	Graph | DiGraph	The graph from which labels and reconciliations shall be reconstructed.	required

Source code in asymmetree/utils/phylogenetic_trees.py

def reconstruct_reconc_from_graph(tree: Tree, G: nx.Graph | nx.DiGraph) -> None:
    """Reconstruct the reconciliations from a NetworkX Graph.

    Args:
        tree: A tree for which reconciliations shall be reconstructed.
        G: The graph from which labels and reconciliations shall be reconstructed.
    """
    for v in tree.preorder():
        if hasattr(v, "label") and v.label in G:
            if "reconc" in G.nodes[v.label]:
                if isinstance(G.nodes[v.label]["reconc"], list):
                    v.reconc = tuple(G.nodes[v.label]["reconc"])
                else:
                    v.reconc = G.nodes[v.label]["reconc"]
            elif "color" in G.nodes[v.label]:
                if isinstance(G.nodes[v.label]["color"], list):
                    v.reconc = tuple(G.nodes[v.label]["color"])
                else:
                    v.reconc = G.nodes[v.label]["color"]

reconstruct_timestamps(tree: Tree) -> None

Reconstruct the timestamps.

Make the time stamps matching with the distance attribute. The root obtains time stamp 1.0, and all other nodes obtain smaller time stamps such that the difference to the parent's time stamp is exactly dist.

Parameters:

Name	Type	Description	Default
tree	Tree	A tree with nodes that have a dist attribute.	required

Source code in asymmetree/utils/phylogenetic_trees.py

def reconstruct_timestamps(tree: Tree) -> None:
    """Reconstruct the timestamps.

    Make the time stamps matching with the distance attribute. The root obtains time stamp 1.0, and
    all other nodes obtain smaller time stamps such that the difference to the parent's time stamp
    is exactly `dist`.

    Args:
        tree: A tree with nodes that have a `dist` attribute.
    """
    tree.root.tstamp = 1.0
    for v in tree.preorder():
        if v.parent:
            v.tstamp = v.parent.tstamp - v.dist

delete_losses_and_contract(tree: Tree, inplace: bool = False) -> Tree

Delete all branches leading to loss leaves only.

Nodes that would have only a single child afterwards are suppressed (except possibly the planted root), i.e. their children are recursively reconnected to the parents. Distances are cumulated in this process and the transferred status is kept in the sense that an edge is a transfer edge if at least one edge on the contracted path to which it corresponds was a transfer edge.

Parameters:

Name	Type	Description	Default
tree	Tree	The tree in which loss branches shall be removed.	required
inplace	bool	If True, the tree is directly manipulated. The default is False in which case a copy of the tree is created which gets manipulated while the original tree remains untouched.	False

Returns:

Type	Description
Tree	The tree with all loss branches removed (original instance or a new one depending on the inplace parameter).

Source code in asymmetree/utils/phylogenetic_trees.py

def delete_losses_and_contract(tree: Tree, inplace: bool = False) -> Tree:
    """Delete all branches leading to loss leaves only.

    Nodes that would have only a single child afterwards are suppressed (except possibly the planted
    root), i.e. their children are recursively reconnected to the parents. Distances are cumulated
    in this process and the transferred status is kept in the sense that an edge is a transfer edge
    if at least one edge on the contracted path to which it corresponds was a transfer edge.

    Args:
        tree: The tree in which loss branches shall be removed.
        inplace: If True, the tree is directly manipulated. The default is False in which case a
            copy of the tree is created which gets manipulated while the original tree remains
            untouched.

    Returns:
        The tree with all loss branches removed (original instance or a new one depending on the
            `inplace` parameter).
    """
    if not inplace:
        tree = tree.copy()

    loss_nodes = []
    for node in tree.postorder():
        if not node.children and node.event == "L":
            loss_nodes.append(node)

    # traverse from loss node to root delete if degree <= 1
    for loss_node in loss_nodes:
        current = delete_and_reconnect(tree, loss_node, add_distances=True, keep_transferred=True)

        while len(current.children) < 2 and current.parent:
            current = delete_and_reconnect(tree, current, add_distances=True, keep_transferred=True)

    return tree

remove_planted_root(tree: Tree, inplace: bool = True) -> Tree

Remove the planted root of the tree (if existent).

Parameters:

Name	Type	Description	Default
tree	Tree	The tree in which the planted root shall be removed.	required
inplace	bool	If True, the tree is directly manipulated, otherwise a copy is created and the original tree remains untouched.	True

Returns:

Type	Description
Tree	The tree with the planted root removed (original instance or a new one depending on the inplace parameter).

Source code in asymmetree/utils/phylogenetic_trees.py

def remove_planted_root(tree: Tree, inplace: bool = True) -> Tree:
    """Remove the planted root of the tree (if existent).

    Args:
        tree: The tree in which the planted root shall be removed.
        inplace: If True, the tree is directly manipulated, otherwise a copy is created and the
            original tree remains untouched.

    Returns:
        The tree with the planted root removed (original instance or a new one depending on the
            `inplace` parameter).
    """
    if not inplace:
        tree = tree.copy()

    # delete the root if the tree is planted
    if len(tree.root.children) == 1:
        new_root = tree.root.children[0]
        new_root.detach()
        tree.root = new_root
        new_root.dist = 0.0

    if not tree.root.children and not tree.root.label:
        # no surviving genes --> return empty tree
        tree.root = None

    return tree

to_newick(tree: Tree, label: bool = True, reconc: bool = True, distance: bool = True, label_inner: bool = True, reconc_inner: bool = False) -> str

Return a Newick representation of the tree.

The Newick string can contain the labels, reconciliations and distances of the nodes. The labels and reconciliations of the inner nodes can optionally be included as well. The reconciliations are included in <...> brackets, and the distances are included in :... notation as usual. By default, labels, reconciliations and distances of all nodes are included with the exception of the reconciliations of the inner nodes.

Parameters:

Name	Type	Description	Default
label	bool	If True, the Newick str contains the labels of the nodes.	True
reconc	bool	If True, the Newick str contains the reconciliations of the nodes in <[...]> brackets.	True
distance	bool	If True, the Newick str contains the distances of the nodes in standard :[...] notation.	True
label_inner	bool	If True, the Newick str also contains the labels of the inner nodes.	True
reconc_inner	bool	If True, the Newick str contains the reconciliations of the inner nodes.	False

Returns:

Type	Description
str	A Newick representation of the tree.

Source code in asymmetree/utils/phylogenetic_trees.py

def to_newick(
    tree: Tree,
    label: bool = True,
    reconc: bool = True,
    distance: bool = True,
    label_inner: bool = True,
    reconc_inner: bool = False,
) -> str:
    """Return a Newick representation of the tree.

    The Newick string can contain the labels, reconciliations and distances of the nodes. The labels
    and reconciliations of the inner nodes can optionally be included as well. The reconciliations
    are included in <...> brackets, and the distances are included in :... notation as usual.
    By default, labels, reconciliations and distances of all nodes are included with the exception
    of the reconciliations of the inner nodes.

    Args:
        label: If True, the Newick str contains the labels of the nodes.
        reconc: If True, the Newick str contains the reconciliations of the nodes in <[...]>
            brackets.
        distance: If True, the Newick str contains the distances of the nodes in standard :[...]
            notation.
        label_inner: If True, the Newick str also contains the labels of the inner nodes.
        reconc_inner: If True, the Newick str contains the reconciliations of the inner nodes.

    Returns:
        A Newick representation of the tree.
    """

    def _to_newick(node: TreeNode) -> str:
        if not node.children:
            token = ""
            if label and hasattr(node, "label"):
                token += str(node.label)
            if reconc and hasattr(node, "reconc") and node.reconc:
                token += (
                    "<{}-{}>".format(*node.reconc)
                    if isinstance(node.reconc, (tuple, list))
                    else "<{}>".format(node.reconc)
                )
            if distance and hasattr(node, "dist"):
                token += ":{}".format(node.dist)
            return token
        else:
            s = ""
            for child in node.children:
                s += _to_newick(child) + ","
            token = ""
            if label and hasattr(node, "label") and label_inner:
                token += str(node.label)
            if reconc_inner and hasattr(node, "reconc") and node.reconc:
                token += (
                    "<{}-{}>".format(*node.reconc)
                    if isinstance(node.reconc, (tuple, list))
                    else "<{}>".format(node.reconc)
                )
            if distance and hasattr(node, "dist"):
                token += ":{}".format(node.dist)
            return "({}){}".format(s[:-1], token)

    if tree.root:
        return _to_newick(tree.root) + ";"
    else:
        return ";"

parse_newick(newick: str) -> Tree

Parses trees in Newick format into object of type Tree.

This function parses trees in Newick format into object of type Tree. The Newick string may contain labels, reconciliations and distances of the nodes. The reconciliations are expected to be included in <...> brackets, and the distances are expected to be included in :... notation as usual. The labels and reconciliations of the inner nodes can optionally be included as well. Labels and reconciliations that are integer numbers are converted to int. The reconciliations (if present in <...> in the string) are parsed as strings and need to be converted to integers afterwards if necessary.

NOTE: Do not use this function for serialization and reloading Tree objects. Use the serialize() function instead.

Parameters:

Name	Type	Description	Default
newick	str	A tree in Newick format.	required

Returns:

Type	Description
Tree	The parsed tree.

Raises:

Type	Description
TypeError	If the input is not a string.
ValueError	If the input is not a valid Newick string.

Source code in asymmetree/utils/phylogenetic_trees.py

def parse_newick(newick: str) -> Tree:
    """Parses trees in Newick format into object of type `Tree`.

    This function parses trees in Newick format into object of type `Tree`. The Newick string may
    contain labels, reconciliations and distances of the nodes. The reconciliations are expected to
    be included in <...> brackets, and the distances are expected to be included in :... notation as
    usual. The labels and reconciliations of the inner nodes can optionally be included as well.
    Labels and reconciliations that are integer numbers are converted to int. The reconciliations
    (if present in <...> in the string) are parsed as strings and need to be converted to integers
    afterwards if necessary.

    NOTE: Do not use this function for serialization and reloading `Tree` objects. Use the
    `serialize()` function instead.

    Args:
        newick: A tree in Newick format.

    Returns:
        The parsed tree.

    Raises:
        TypeError: If the input is not a string.
        ValueError: If the input is not a valid Newick string.
    """

    def to_int(item: str) -> int | str:
        """Tries to convert the string into int."""
        return int(item) if item.isdigit() else item

    tree = Tree.parse_newick(newick)

    # regex for notation label<reconc>
    label_col_regex = re.compile(r"'?([a-zA-Z0-9_]*)'?<(.*)>")

    for node in tree.preorder():
        node.event = ""

        if not hasattr(node, "dist"):
            node.dist = 1.0

        label = str(node.label)

        label_col = label_col_regex.match(label)
        if label_col:
            node.label = to_int(label_col.group(1))
            node.reconc = to_int(label_col.group(2))
        else:
            node.reconc = None

        # reconc is a tuple
        if node.reconc and isinstance(node.reconc, str) and node.reconc.find("-") != -1:
            a, b = node.reconc.split("-")
            node.reconc = (to_int(a), to_int(b))

    return tree

distance_calculation

Calculation of sequence distances for various models.

The distance between two sequences is defined as the expected number of substitutions per site that have occurred since the two sequences diverged from their common ancestor. This module provides functions to calculate distances under various substitution models, including the Jukes-Cantor 1969 model, the Kimura 1980 model, and maximum likelihood distances under general substitution models.

References

1. Z. Yang. Computational molecular evolution. Oxford series in ecology and evolution. Oxford University Press, 2006. ISBN 978-0-19-856699-1 978-0-19-856702-8.

maximum_likelihood_distance(seq1: str, seq2: str, subst_model: SubstModel = None, model_type: str | None = None, model_name: str | None = None) -> float

Calculate the maximum likelihood distance between two aligned sequences.

Parameters:

Name	Type	Description	Default
seq1	str	First aligned sequence.	required
seq2	str	Second aligned sequence.	required
subst_model	SubstModel	Substitution model to use for distance calculation. If None, model_type and model_name must be provided to construct a SubstModel instance.	None
model_type	str | None	Type of substitution model (e.g., "n" for nucleotide, "a" for amino acid). Required if subst_model is None.	None
model_name	str | None	Name of substitution model (e.g., "JC69", "K80"). Required if subst_model is None.	None

Returns:

Type	Description
float	Estimated distance between the two sequences under the specified substitution model.

Raises:

Type	Description
ValueError	If subst_model is None and either model_type or model_name is not provided.

Source code in asymmetree/utils/distance_calculation.py

def maximum_likelihood_distance(
    seq1: str,
    seq2: str,
    subst_model: SubstModel = None,
    model_type: str | None = None,
    model_name: str | None = None,
) -> float:
    """Calculate the maximum likelihood distance between two aligned sequences.

    Args:
        seq1: First aligned sequence.
        seq2: Second aligned sequence.
        subst_model: Substitution model to use for distance calculation. If None, model_type and
            model_name must be provided to construct a SubstModel instance.
        model_type: Type of substitution model (e.g., "n" for nucleotide, "a" for amino acid).
            Required if subst_model is None.
        model_name: Name of substitution model (e.g., "JC69", "K80"). Required if subst_model is
            None.

    Returns:
        Estimated distance between the two sequences under the specified substitution model.

    Raises:
        ValueError: If subst_model is None and either model_type or model_name is not provided.
    """
    if subst_model is None:
        if model_type is None or model_name is None:
            raise ValueError("no substitution model specified")

        subst_model = SubstModel(model_type, model_name)

    seqs = _to_indices(seq1, seq2, subst_model)

    x0 = _initial_guess(seqs, subst_model.model_type)

    # distance 0.0 if sequences are equal
    if x0 == 0.0:
        return x0

    opt_result = scipy.optimize.minimize(
        _likelihood,
        x0,
        args=(seqs, subst_model),
        bounds=((MIN_LENGTH, None),),
    )

    if opt_result.success:
        return opt_result.x[0]
    else:
        return float("nan")

p_distance(seq1: str, seq2: str, exclude_gaps: bool = True) -> tuple[float, int]

Calculate the p-distance of two aligned sequences (normalized Hamming distance).

The p-distance is the proportion of sites at which the two sequences differ. It equals the Hamming distance divided by the number of valid columns (i.e., columns without gaps if exclude_gaps is True).

Parameters:

Name	Type	Description	Default
seq1	str	First aligned sequence.	required
seq2	str	Second aligned sequence.	required
exclude_gaps	bool	If True, columns with a gap in one sequence are ignored. Columns with gaps in both sequences are always ignored.	True

Returns:

Type	Description
float	The p-distance.
int	The number of valid columns used in the calculation.

Raises:

Type	Description
ValueError	If the input sequences have different lengths.

Source code in asymmetree/utils/distance_calculation.py

def p_distance(seq1: str, seq2: str, exclude_gaps: bool = True) -> tuple[float, int]:
    """Calculate the p-distance of two aligned sequences (normalized Hamming distance).

    The p-distance is the proportion of sites at which the two sequences differ. It equals the
    Hamming distance divided by the number of valid columns (i.e., columns without gaps if
    exclude_gaps is True).

    Args:
        seq1: First aligned sequence.
        seq2: Second aligned sequence.
        exclude_gaps: If True, columns with a gap in one sequence are ignored. Columns with gaps in
            both sequences are always ignored.

    Returns:
        The p-distance.
        The number of valid columns used in the calculation.

    Raises:
        ValueError: If the input sequences have different lengths.
    """
    if len(seq1) != len(seq2):
        raise ValueError(f"unequal sequence lengths: {len(seq1)} and {len(seq2)}")

    diffs, valid_columns = 0, 0

    for i in range(len(seq1)):
        if seq1[i] != "-" and seq2[i] != "-":
            valid_columns += 1
            if seq1[i] != seq2[i]:
                diffs += 1

        elif not exclude_gaps and (seq1[i] != "-" or seq2[i] != "-"):
            valid_columns += 1
            diffs += 1

    p = diffs / valid_columns if valid_columns > 0 else float("inf")

    return p, valid_columns

JC69_distance(seq1: str, seq2: str, exclude_gaps: bool = True, amino_acid: bool = False, variance: bool = False) -> float | tuple[float, float]

Jukes Cantor 1969 distance.

The Jukes-Cantor 1969 model assumes equal base frequencies and equal substitution rates among all nucleotides.

Parameters:

Name	Type	Description	Default
seq1	str	First aligned sequence.	required
seq2	str	Second aligned sequence.	required
exclude_gaps	bool	If True, columns with a gap in one sequence are ignored. Columns with gaps in both sequences are always ignored.	True
amino_acid	bool	If True, the distance is calculated under the Jukes-Cantor model for amino acids, which assumes equal frequencies and substitution rates among all amino acids.	False
variance	bool	If True, also calculate the variance of the distance estimate.	False

Returns:

Type	Description
float | tuple[float, float]	If variance is False, returns the Jukes-Cantor distance. If variance is True, returns a tuple containing the distance and its variance.

Source code in asymmetree/utils/distance_calculation.py

def JC69_distance(
    seq1: str,
    seq2: str,
    exclude_gaps: bool = True,
    amino_acid: bool = False,
    variance: bool = False,
) -> float | tuple[float, float]:
    """Jukes Cantor 1969 distance.

    The Jukes-Cantor 1969 model assumes equal base frequencies and equal substitution rates among
    all nucleotides.

    Args:
        seq1: First aligned sequence.
        seq2: Second aligned sequence.
        exclude_gaps: If True, columns with a gap in one sequence are ignored. Columns with gaps in
            both sequences are always ignored.
        amino_acid: If True, the distance is calculated under the Jukes-Cantor model for amino
            acids, which assumes equal frequencies and substitution rates among all amino acids.
        variance: If True, also calculate the variance of the distance estimate.

    Returns:
        If `variance` is False, returns the Jukes-Cantor distance. If `variance` is True, returns a
            tuple containing the distance and its variance.
    """
    p, valid_columns = p_distance(seq1, seq2, exclude_gaps=exclude_gaps)

    d = _JC69_transform(p, amino_acid=amino_acid)

    if not variance:
        return d
    else:
        var_d = _JC69_distance_var(p, valid_columns, amino_acid=amino_acid)
        return d, var_d

K80_distance(seq1: str, seq2: str, variance: bool = False) -> tuple[float, float] | tuple[float, float, float]

Kimura 1980 distance and transition/transversion ratio.

The Kimura 1980 model distinguishes between transitions (substitutions between purines or pyrimidines) and transversions (substitutions between a purine and a pyrimidine). It assumes equal base frequencies and equal substitution rates among all nucleotides, but allows for different rates between transitions and transversions.

Parameters:

Name	Type	Description	Default
seq1	str	First aligned sequence.	required
seq2	str	Second aligned sequence.	required
variance	bool	If True, also calculate the variance of the distance estimate.	False

Returns:

Type	Description
tuple[float, float] | tuple[float, float, float]	If variance is False, returns a tuple containing the Kimura distance and the transition / transversion ratio. If variance is True, returns a tuple containing the distance, the transition / transversion ratio, and the variance of the distance.

Source code in asymmetree/utils/distance_calculation.py

def K80_distance(
    seq1: str, seq2: str, variance: bool = False
) -> tuple[float, float] | tuple[float, float, float]:
    """Kimura 1980 distance and transition/transversion ratio.

    The Kimura 1980 model distinguishes between transitions (substitutions between purines or
    pyrimidines) and transversions (substitutions between a purine and a pyrimidine). It assumes
    equal base frequencies and equal substitution rates among all nucleotides, but allows
    for different rates between transitions and transversions.

    Args:
        seq1: First aligned sequence.
        seq2: Second aligned sequence.
        variance: If True, also calculate the variance of the distance estimate.

    Returns:
        If `variance` is False, returns a tuple containing the Kimura distance and the transition /
            transversion ratio. If `variance` is True, returns a tuple containing the distance, the
            transition / transversion ratio, and the variance of the distance.
    """
    S, V, valid_columns = _IV_proportions(seq1, seq2)

    d, kappa = _K80_transform(S, V)

    if not variance:
        return d, kappa
    else:
        var_d = _K80_distance_var(S, V, valid_columns)
        return d, kappa, var_d

sampling

Module for sampling from various distributions.

The Sampler class allows sampling from a variety of distributions, including constant, uniform, discrete uniform, gamma, exponential, zipf, and negative binomial distributions. The class also supports specifying minimum and maximum values for the samples, as well as shifting the distribution along the x-axis.

Sampler

Sampling from a variety of distributions.

Source code in asymmetree/utils/sampling.py

class Sampler:
    """Sampling from a variety of distributions."""

    def __init__(
        self,
        params: int | float | tuple | list,
        minimum: int | float | None = None,
        maximum: int | float | None = None,
        discrete: bool = False,
        shift: int | float = 0,
    ) -> None:
        """Construct a `Sampler` object.

        Args:
            params: The distribution and its parameters, see documentation for available options.
            minimum: Minimum value to be sampled.
            maximum: Maximum value to be sampled.
            discrete: If True, round sampled values in case of a continuous distribution.
            shift: Shift distribution along x-axis. The default is 0, i.e., the distribution is not
                shifted.
        """
        self._params = params
        self._discrete = discrete
        self._min = minimum
        self._max = maximum
        self._shift = shift

        # function to be defined in _check_and_initialize_distribution
        self.draw = None

        self._check_and_initialize_distribution()

    def __call__(self) -> int | float:
        """Draw a value from the specified distribution.

        Returns:
            A value drawn from the specified distribution, which is an `int` if `discrete` is True
                and a `float` otherwise.
        """
        return self.draw()

    def _check_and_initialize_distribution(self):
        """Check if the specified distribution is valid and set the draw function."""
        if self._min is not None and not isinstance(self._min, (int, float)):
            raise ValueError("minimum must be a number")

        if self._max is not None and not isinstance(self._max, (int, float)):
            raise ValueError("maximum must be a number")

        if not isinstance(self._shift, (int, float)):
            raise ValueError("shift value must be a number")

        if isinstance(self._params, (int, float)):
            self._distr = "constant"
            self.draw = self._draw_constant
            self._exp_val = round(self._params) if self._discrete else self._params

        elif isinstance(self._params, (tuple, list)) and len(self._params) >= 2:
            if self._params[0] == "constant":
                val = self._params[1]
                if not isinstance(val, (int, float)):
                    raise ValueError("constant value must be a number")

                self._distr = "constant"
                self.draw = self._draw_constant
                self._exp_val = round(val) if self._discrete else val

            elif self._params[0] == "uniform":
                a = self._params[1]
                if not len(self._params) >= 3:
                    raise ValueError("uniform distr. requires 2 parameters")
                b = self._params[2]

                if not isinstance(a, (int, float)) or not isinstance(b, (int, float)) or b < a:
                    raise ValueError(
                        "parameters a and b for uniform distr.must be numbers and a <= b"
                    )

                self._distr = "uniform"
                self.draw = self._draw_uniform
                self._exp_val = (a + b) / 2
                self._a = a
                self._b = b

            elif self._params[0] == "discrete_uniform":
                a = self._params[1]
                if not len(self._params) >= 3:
                    raise ValueError("discrete uniform distr. requires 2 parameters")
                b = self._params[2]

                if not isinstance(a, int) or not isinstance(b, int) or b < a:
                    raise ValueError(
                        "parameters a and b for discrete uniform distr. must be ints and a <= b"
                    )

                self._distr = "discrete_uniform"
                self.draw = self._draw_discrete_uniform
                self._exp_val = (a + b) / 2
                self._a = a
                self._b = b

            elif self._params[0] == "gamma":
                shape = self._params[1]
                if not len(self._params) >= 3:
                    raise ValueError("gamma distr. requires 2 parameters")
                scale = self._params[2]

                if (
                    not isinstance(shape, (int, float))
                    or shape <= 0.0
                    or not isinstance(scale, (int, float))
                    or scale <= 0.0
                ):
                    raise ValueError(
                        "scale and shape parameters for gamma distribution must be numbers >0.0"
                    )

                shape = float(shape)
                scale = float(scale)

                self._distr = "gamma"
                self.draw = self._draw_gamma
                self._exp_val = shape * scale
                self._shape = shape
                self._scale = scale

            elif self._params[0] == "gamma_mean":
                mean = self._params[1]

                if not isinstance(mean, (int, float)) or mean <= 0.0:
                    raise ValueError("mean of gamma distribution must be a number >0.0")

                mean = float(mean)
                self._distr = "gamma"
                self.draw = self._draw_gamma
                self._exp_val = mean
                self._shape = 1.0
                self._scale = mean / self._shape

            elif self._params[0] == "exponential":
                rate = self._params[1]

                if not isinstance(rate, (int, float)) or rate < 0.0:
                    raise ValueError("rate must be a number >=0.0")

                rate = float(rate)
                self._distr = "exponential"
                self.draw = self._draw_exponential
                self._exp_val = float("inf") if rate == 0.0 else 1 / rate
                self._rate = rate

            elif self._params[0] == "zipf":
                a = self._params[1]

                if not isinstance(a, (int, float)) or a <= 1.0:
                    raise ValueError("parameter a for zipf distr. must be a number >1")

                a = float(a)
                # expected value is infinite for a <= 2
                if a > 2.0:
                    self._exp_val = zeta(a - 1.0) / zeta(a)

                # use some valid value when exp. value is infinite
                elif self._min is not None:
                    self._exp_val = self._min
                else:
                    self._exp_val = 0.0

                self._distr = "zipf"
                self.draw = self._draw_zipf
                self._a = a

            elif self._params[0] == "negative_binomial":
                r = self._params[1]
                if not len(self._params) >= 3:
                    raise ValueError("negative binomial distr. requires 2 parameters")
                q = self._params[2]

                if not isinstance(r, int) or r < 1:
                    raise ValueError("parameter r must be an int and >0")
                if not isinstance(q, (int, float)) or q <= 0 or q >= 1.0:
                    raise ValueError("parameter q must be a number >0 and <1")

                q = float(q)
                self._distr = "negative_binomial"
                self.draw = self._draw_neg_bin
                self._exp_val = r * q / (1 - q)
                self._r = r
                self._q = q

            else:
                raise ValueError(f"length distribution '{self._params[0]}' is not supported")

        else:
            raise ValueError("could not parse distribution")

        self._exp_val += self._shift

        # expected value must be between min. and max.
        if (self._min is not None and self._exp_val < self._min) or (
            self._max is not None and self._exp_val > self._max
        ):
            raise ValueError("expected value must be >= minimum and<= maximum")

    def _draw_constant(self):
        """Draw a value from a constant distribution."""
        return self._exp_val

    def _draw_uniform(self):
        """Draw a value from a uniform distribution."""
        while True:
            x = np.random.uniform(low=self._a, high=self._b) + self._shift

            if self._discrete:
                x = round(x)

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

    def _draw_discrete_uniform(self):
        """Draw a value from a discrete uniform distribution."""
        while True:
            x = np.random.randint(self._a, high=self._b) + self._shift

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

    def _draw_gamma(self):
        """Draw a value from a gamma distribution."""
        while True:
            x = np.random.gamma(self._shape, scale=self._scale) + self._shift

            if self._discrete:
                x = round(x)

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

    def _draw_exponential(self):
        """Draw a value from an exponential distribution."""
        while True:
            if self._rate == 0.0:
                x = float("inf")
            else:
                x = np.random.exponential(scale=1 / self._rate) + self._shift

            if self._discrete:
                x = round(x)

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

    def _draw_zipf(self):
        """Draw a value from a Zipf distribution."""
        while True:
            x = np.random.zipf(self._a) + self._shift

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

    def _draw_neg_bin(self):
        """Draw a value from a negative binomial distribution."""
        while True:
            x = np.random.negative_binomial(self._r, 1 - self._q) + self._shift

            if (self._min is None or x >= self._min) and (self._max is None or x <= self._max):
                return x

__init__(params: int | float | tuple | list, minimum: int | float | None = None, maximum: int | float | None = None, discrete: bool = False, shift: int | float = 0) -> None

Construct a Sampler object.

Parameters:

Name	Type	Description	Default
params	int | float | tuple | list	The distribution and its parameters, see documentation for available options.	required
minimum	int | float | None	Minimum value to be sampled.	None
maximum	int | float | None	Maximum value to be sampled.	None
discrete	bool	If True, round sampled values in case of a continuous distribution.	False
shift	int | float	Shift distribution along x-axis. The default is 0, i.e., the distribution is not shifted.	0

Source code in asymmetree/utils/sampling.py

def __init__(
    self,
    params: int | float | tuple | list,
    minimum: int | float | None = None,
    maximum: int | float | None = None,
    discrete: bool = False,
    shift: int | float = 0,
) -> None:
    """Construct a `Sampler` object.

    Args:
        params: The distribution and its parameters, see documentation for available options.
        minimum: Minimum value to be sampled.
        maximum: Maximum value to be sampled.
        discrete: If True, round sampled values in case of a continuous distribution.
        shift: Shift distribution along x-axis. The default is 0, i.e., the distribution is not
            shifted.
    """
    self._params = params
    self._discrete = discrete
    self._min = minimum
    self._max = maximum
    self._shift = shift

    # function to be defined in _check_and_initialize_distribution
    self.draw = None

    self._check_and_initialize_distribution()

__call__() -> int | float

Draw a value from the specified distribution.

Returns:

Type	Description
int | float	A value drawn from the specified distribution, which is an int if discrete is True and a float otherwise.

Source code in asymmetree/utils/sampling.py

def __call__(self) -> int | float:
    """Draw a value from the specified distribution.

    Returns:
        A value drawn from the specified distribution, which is an `int` if `discrete` is True
            and a `float` otherwise.
    """
    return self.draw()