7894856040

Committed 14 Feb 2024 12:52AM UTC coverage: 46.404% (-5.7%) from 52.085%

Build # 7894856040

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #38 from js51/SplitP-rewrite

Re-organise modules

Run Details

403 of 880 new or added lines in 12 files covered. (45.8%)

413 of 890 relevant lines covered (46.4%)

1.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

72.04

/splitp/phylogeny.py

from splitp.parsers.newick import newick_to_json, json_to_newick
from networkx import json_graph
from dataclasses import dataclass
from networkx import draw
from splitp.enums import DrawFormat
import splitp.constants as constants
import numpy as np

class Phylogeny:
    __slots__ = (
        "name",
        "networkx_graph",
        "newick_string",
    )

    def __init__(
        self,
        newick_string,
        name=None,
    ):
        """A rooted phylogenetic tree.

        A rooted phylogenetic tree, stored as a wrapped networkx graph.

        Attributes:
            name: A name for the tree
            networkx_graph: the underlying networkx graph
        """

        # Set chosen tree properties
        self.name = name if name else "Unnamed Tree"
        self.newick_string = newick_string

        # Build networkx graph from newick string
        self.networkx_graph = json_graph.tree_graph(
            newick_to_json(self.newick_string, generate_names=True)
        )

    def __str__(self):
        """Return the tree in JSON format"""
        return json_to_newick(
            json_graph.tree_data(self.networkx_graph, self.root(return_index=False))
        )
    
    def unrooted_networkx_graph(self):
        """Return the unrooted version of the tree"""
        # Make a copy of the graph
        unrooted_graph = self.networkx_graph.copy()
        # Get the root node
        root_node = self.root(return_index=False)
        # If the root node has more than 2 children, return the original graph
        if len(list(unrooted_graph.successors(root_node))) > 2:
            return unrooted_graph.to_undirected()
        # Get the root node children
        root_children = list(unrooted_graph.successors(self.root(return_index=False)))
        # Remove the root node
        unrooted_graph.remove_node(self.root(return_index=False))
        # Connect the two root node children
        unrooted_graph.add_edge(root_children[0], root_children[1])
        # Make undirected
        unrooted_graph = unrooted_graph.to_undirected()
        # Return the unrooted graph
        return unrooted_graph
    
    def reassign_transition_matrices(self, transition_matrix):
        """DEPRECATED: Reassign transition matrices to all nodes in the tree"""
        for node in self.networkx_graph.nodes:
            self.networkx_graph.nodes[node]["transition_matrix"] = np.array(transition_matrix)
    
    def get_num_nodes(self):
        return len(self.networkx_graph.nodes)
    
    def node_index(self, node):
        return list(self.networkx_graph.nodes).index(node)

    def __eq__(self, other):
        """Check equality of JSON objects"""
        return self.__str__() == other.__str__()

    def is_root(self, node):
        """Determines whether a node is a root node from its index."""
        if type(node) == type(str()):
            # A node name was supplied
            return self.networkx_graph.in_degree(node) == 0
        else:
            # An index was supplied
            return (
                self.networkx_graph.in_degree(list(self.networkx_graph.nodes)[node])
                == 0
            )

    def root(self, return_index=True):
        """Returns the root node"""
        for i, n in enumerate(self.networkx_graph.nodes):
            if self.is_root(n):
                return i if return_index else n

    def get_parent(self, n):
        """Returns the parent node for a given node"""
        return list(self.nx_graph.predecessors(n))[0]
    
    def nodes(self):
        """Returns a list of all nodes in the tree."""
        return list(self.networkx_graph.nodes)

    def get_taxa(self):
        return [n for n in self.networkx_graph.nodes if self.is_leaf(n)]
    
    def get_num_taxa(self):
        return len(self.get_taxa())

    def is_leaf(self, n_index_or_name):
        """Determines whether a node is a leaf node from it's index."""
        if type(n_index_or_name) == type(str()):
            return self.networkx_graph.out_degree(n_index_or_name) == 0
        else:
            return (
                self.networkx_graph.out_degree(
                    list(self.networkx_graph.nodes)[n_index_or_name]
                )
                == 0
            )

    def get_parent(self, n):
        """Returns the parent node for a given node"""
        return list(self.networkx_graph.predecessors(n))[0]

    def get_descendants(self, n, return_iter=False):
        """Returns a list of children/descendents of a given node"""
        return (
            list(self.networkx_graph.successors(n))
            if not return_iter
            else self.networkx_graph.successors(n)
        )

    def splits(self, include_trivial=False, as_strings=False):
        """Returns set of all splits displayed by the tree."""
        from networkx.algorithms.traversal.depth_first_search import dfs_tree
        all_taxa = [x for x in self.get_taxa()]
        splits = set()
        for node in list(self.nodes()):
            subtree = dfs_tree(self.networkx_graph, node)
            left = tuple(
                    sorted(
                        [
                            leaf
                            for leaf in subtree.nodes
                            if leaf in all_taxa
                        ],
                        key=all_taxa.index,
                    )
                )
            right = tuple(sorted((i for i in all_taxa if i not in left), key=all_taxa.index))
            split = (left, right)
            if all_taxa[0] not in split[0]:
                split = (split[1], split[0])
            if include_trivial or (len(split[0]) > 1 and len(split[1]) > 1):
                splits.add(split)
        for split in splits:
            yield split if not as_strings else self.format_split(split)

    def format_split(self, split):
        if isinstance(split, str):
            return split  # If already a string, just send it back
        if len(split[0]) + len(split[1]) > 35:
            raise ValueError(
                "Cannot produce string format for split with more than 35 taxa."
            )
        if all(len(taxon) == 1 for taxon in self.get_taxa()):
            return f'{"".join(split[0])}|{"".join(split[1])}'

    def draw(self, draw_format=DrawFormat.ASCII):
        try:
            from Bio.Phylo import draw_ascii, draw, read
            from io import StringIO
        except ModuleNotFoundError:
            print(
                "Additional dependencies required to draw tree. Please install BioPython and StringIO."
            )
            return
        tree = read(StringIO(str(self)), "newick")
        if draw_format == DrawFormat.ASCII:
            draw_ascii(tree)
        elif draw_format == DrawFormat.matplotlib:
            draw(tree, branch_labels=lambda c: c.branch_length)
        else:
            raise ValueError("Invalid format.")

1	from splitp.parsers.newick import newick_to_json, json_to_newick	3✔
2	from networkx import json_graph	3✔
3	from dataclasses import dataclass	3✔
4	from networkx import draw	3✔
5	from splitp.enums import DrawFormat	3✔
6	import splitp.constants as constants	3✔
7	import numpy as np	3✔
8
9	class Phylogeny:	3✔
10	__slots__ = (	3✔
11	"name",
12	"networkx_graph",
13	"newick_string",
14	)
15
16	def __init__(	3✔
17	self,
18	newick_string,
19	name=None,
20	):
21	"""A rooted phylogenetic tree.
22
23	A rooted phylogenetic tree, stored as a wrapped networkx graph.
24
25	Attributes:
26	name: A name for the tree
27	networkx_graph: the underlying networkx graph
28	"""
29
30	# Set chosen tree properties
31	self.name = name if name else "Unnamed Tree"	3✔
32	self.newick_string = newick_string	3✔
33
34	# Build networkx graph from newick string
35	self.networkx_graph = json_graph.tree_graph(	3✔
36	newick_to_json(self.newick_string, generate_names=True)
37	)
38
39	def __str__(self):	3✔
40	"""Return the tree in JSON format"""
NEW 41	return json_to_newick(	×
42	json_graph.tree_data(self.networkx_graph, self.root(return_index=False))
43	)
44
45	def unrooted_networkx_graph(self):	3✔
46	"""Return the unrooted version of the tree"""
47	# Make a copy of the graph
48	unrooted_graph = self.networkx_graph.copy()	3✔
49	# Get the root node
50	root_node = self.root(return_index=False)	3✔
51	# If the root node has more than 2 children, return the original graph
52	if len(list(unrooted_graph.successors(root_node))) > 2:	3✔
53	return unrooted_graph.to_undirected()	3✔
54	# Get the root node children
55	root_children = list(unrooted_graph.successors(self.root(return_index=False)))	3✔
56	# Remove the root node
57	unrooted_graph.remove_node(self.root(return_index=False))	3✔
58	# Connect the two root node children
59	unrooted_graph.add_edge(root_children[0], root_children[1])	3✔
60	# Make undirected
61	unrooted_graph = unrooted_graph.to_undirected()	3✔
62	# Return the unrooted graph
63	return unrooted_graph	3✔
64
65	def reassign_transition_matrices(self, transition_matrix):	3✔
66	"""DEPRECATED: Reassign transition matrices to all nodes in the tree"""
NEW 67	for node in self.networkx_graph.nodes:	×
NEW 68	self.networkx_graph.nodes[node]["transition_matrix"] = np.array(transition_matrix)	×
69
70	def get_num_nodes(self):	3✔
NEW 71	return len(self.networkx_graph.nodes)	×
72
73	def node_index(self, node):	3✔
NEW 74	return list(self.networkx_graph.nodes).index(node)	×
75
76	def __eq__(self, other):	3✔
77	"""Check equality of JSON objects"""
NEW 78	return self.__str__() == other.__str__()	×
79
80	def is_root(self, node):	3✔
81	"""Determines whether a node is a root node from its index."""
82	if type(node) == type(str()):	3✔
83	# A node name was supplied
84	return self.networkx_graph.in_degree(node) == 0	3✔
85	else:
86	# An index was supplied
NEW 87	return (	×
88	self.networkx_graph.in_degree(list(self.networkx_graph.nodes)[node])
89	== 0
90	)
91
92	def root(self, return_index=True):	3✔
93	"""Returns the root node"""
94	for i, n in enumerate(self.networkx_graph.nodes):	3✔
95	if self.is_root(n):	3✔
96	return i if return_index else n	3✔
97
98	def get_parent(self, n):	3✔
99	"""Returns the parent node for a given node"""
NEW 100	return list(self.nx_graph.predecessors(n))[0]	×
101
102	def nodes(self):	3✔
103	"""Returns a list of all nodes in the tree."""
104	return list(self.networkx_graph.nodes)	3✔
105
106	def get_taxa(self):	3✔
107	return [n for n in self.networkx_graph.nodes if self.is_leaf(n)]	3✔
108
109	def get_num_taxa(self):	3✔
NEW 110	return len(self.get_taxa())	×
111
112	def is_leaf(self, n_index_or_name):	3✔
113	"""Determines whether a node is a leaf node from it's index."""
114	if type(n_index_or_name) == type(str()):	3✔
115	return self.networkx_graph.out_degree(n_index_or_name) == 0	3✔
116	else:
NEW 117	return (	×
118	self.networkx_graph.out_degree(
119	list(self.networkx_graph.nodes)[n_index_or_name]
120	)
121	== 0
122	)
123
124	def get_parent(self, n):	3✔
125	"""Returns the parent node for a given node"""
NEW 126	return list(self.networkx_graph.predecessors(n))[0]	×
127
128	def get_descendants(self, n, return_iter=False):	3✔
129	"""Returns a list of children/descendents of a given node"""
NEW 130	return (	×
131	list(self.networkx_graph.successors(n))
132	if not return_iter
133	else self.networkx_graph.successors(n)
134	)
135
136	def splits(self, include_trivial=False, as_strings=False):	3✔
137	"""Returns set of all splits displayed by the tree."""
138	from networkx.algorithms.traversal.depth_first_search import dfs_tree	3✔
139	all_taxa = [x for x in self.get_taxa()]	3✔
140	splits = set()	3✔
141	for node in list(self.nodes()):	3✔
142	subtree = dfs_tree(self.networkx_graph, node)	3✔
143	left = tuple(	3✔
144	sorted(
145	[
146	leaf
147	for leaf in subtree.nodes
148	if leaf in all_taxa
149	],
150	key=all_taxa.index,
151	)
152	)
153	right = tuple(sorted((i for i in all_taxa if i not in left), key=all_taxa.index))	3✔
154	split = (left, right)	3✔
155	if all_taxa[0] not in split[0]:	3✔
156	split = (split[1], split[0])	3✔
157	if include_trivial or (len(split[0]) > 1 and len(split[1]) > 1):	3✔
158	splits.add(split)	3✔
159	for split in splits:	3✔
160	yield split if not as_strings else self.format_split(split)	3✔
161
162	def format_split(self, split):	3✔
163	if isinstance(split, str):	3✔
NEW 164	return split # If already a string, just send it back	×
165	if len(split[0]) + len(split[1]) > 35:	3✔
NEW 166	raise ValueError(	×
167	"Cannot produce string format for split with more than 35 taxa."
168	)
169	if all(len(taxon) == 1 for taxon in self.get_taxa()):	3✔
170	return f'{"".join(split[0])}\|{"".join(split[1])}'	3✔
171
172	def draw(self, draw_format=DrawFormat.ASCII):	3✔
NEW 173	try:	×
NEW 174	from Bio.Phylo import draw_ascii, draw, read	×
NEW 175	from io import StringIO	×
NEW 176	except ModuleNotFoundError:	×
NEW 177	print(	×
178	"Additional dependencies required to draw tree. Please install BioPython and StringIO."
179	)
NEW 180	return	×
NEW 181	tree = read(StringIO(str(self)), "newick")	×
NEW 182	if draw_format == DrawFormat.ASCII:	×
NEW 183	draw_ascii(tree)	×
NEW 184	elif draw_format == DrawFormat.matplotlib:	×
NEW 185	draw(tree, branch_labels=lambda c: c.branch_length)	×
186	else:
NEW 187	raise ValueError("Invalid format.")	×

js51 / SplitP / 7894856040

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous