from typing import Optional
import networkx as nx
import numpy
from matchms import Scores
from .networking_functions import get_top_hits
[docs]class SimilarityNetwork:
"""Create a spectal network from spectrum similarities.
For example
.. testcode::
import numpy as np
from matchms import Spectrum, calculate_scores
from matchms.similarity import ModifiedCosine
from matchms.networking import SimilarityNetwork
spectrum_1 = Spectrum(mz=np.array([100, 150, 200.]),
intensities=np.array([0.7, 0.2, 0.1]),
metadata={"precursor_mz": 100.0,
"test_id": "one"})
spectrum_2 = Spectrum(mz=np.array([104.9, 140, 190.]),
intensities=np.array([0.4, 0.2, 0.1]),
metadata={"precursor_mz": 105.0,
"test_id": "two"})
# Use factory to construct a similarity function
modified_cosine = ModifiedCosine(tolerance=0.2)
spectrums = [spectrum_1, spectrum_2]
scores = calculate_scores(spectrums, spectrums, modified_cosine)
ms_network = SimilarityNetwork(identifier_key="test_id")
ms_network.create_network(scores)
nodes = list(ms_network.graph.nodes())
nodes.sort()
print(nodes)
Should output
.. testoutput::
['one', 'two']
"""
[docs] def __init__(self, identifier_key: str = "spectrum_id",
top_n: int = 20,
max_links: int = 10,
score_cutoff: float = 0.7,
link_method: str = 'single',
keep_unconnected_nodes: bool = True):
"""
Parameters
----------
identifier_key
Metadata key for unique intentifier for each spectrum in scores.
Will also be used for the naming the network nodes. Default is 'spectrum_id'.
top_n
Consider edge between spectrumA and spectrumB if score falls into
top_n for spectrumA or spectrumB (link_method="single"), or into
top_n for spectrumA and spectrumB (link_method="mutual"). From those
potential links, only max_links will be kept, so top_n must be >= max_links.
max_links
Maximum number of links to add per node. Default = 10.
Due to incoming links, total number of links per node can be higher.
The links are populated by looping over the query spectrums.
Important side note: The max_links restriction is strict which means that
if scores around max_links are equal still only max_links will be added
which can results in some random variations (sorting spectra with equal
scores restuls in a random order of such elements).
score_cutoff
Threshold for given similarities. Edges/Links will only be made for
similarities > score_cutoff. Default = 0.7.
link_method
Chose between 'single' and 'mutual'. 'single will add all links based
on individual nodes. 'mutual' will only add links if that link appears
in the given top-n list for both nodes.
keep_unconnected_nodes
If set to True (default) all spectra will be included as nodes even
if they have no connections/edges of other spectra. If set to False
all nodes without connections will be removed.
"""
# pylint: disable=too-many-arguments
self.identifier_key = identifier_key
self.top_n = top_n
self.max_links = max_links
self.score_cutoff = score_cutoff
self.link_method = link_method
self.keep_unconnected_nodes = keep_unconnected_nodes
self.graph: Optional[nx.Graph] = None
"""NetworkX graph. Set after calling create_network()"""
@staticmethod
def _select_edge_score(similars_scores: dict, scores_type: numpy.dtype):
"""Chose one value if score contains multiple values (e.g. "score" and "matches")"""
if len(scores_type) > 1 and "score" in scores_type.names:
return {key: value["score"] for key, value in similars_scores.items()}
if len(scores_type) > 1: # Assume that first entry is desired score
return {key: value[0] for key, value in similars_scores.items()}
return similars_scores
[docs] def create_network(self, scores: Scores):
"""
Function to create network from given top-n similarity values. Expects that
similarities given in scores are from an all-vs-all comparison including all
possible pairs.
Parameters
----------
scores
Matchms Scores object containing all spectrums and pair similarities for
generating a network.
"""
assert self.top_n >= self.max_links, "top_n must be >= max_links"
assert numpy.all(scores.queries == scores.references), \
"Expected symmetric scores object with queries==references"
unique_ids = list({s.get(self.identifier_key) for s in scores.queries})
# Initialize network graph, add nodes
msnet = nx.Graph()
msnet.add_nodes_from(unique_ids)
# Collect location and score of highest scoring candidates for queries and references
similars_idx, similars_scores = get_top_hits(scores, identifier_key=self.identifier_key,
top_n=self.top_n,
search_by="queries",
ignore_diagonal=True)
similars_scores = self._select_edge_score(similars_scores, scores.scores.dtype)
# Add edges based on global threshold (cutoff) for weights
for i, spec in enumerate(scores.queries):
query_id = spec.get(self.identifier_key)
ref_candidates = numpy.array([scores.references[x].get(self.identifier_key)
for x in similars_idx[query_id]])
idx = numpy.where((similars_scores[query_id] >= self.score_cutoff) &
(ref_candidates != query_id))[0][:self.max_links]
if self.link_method == "single":
new_edges = [(query_id, str(ref_candidates[x]),
float(similars_scores[query_id][x])) for x in idx]
elif self.link_method == "mutual":
new_edges = [(query_id, str(ref_candidates[x]),
float(similars_scores[query_id][x]))
for x in idx if i in similars_idx[ref_candidates[x]][:]]
else:
raise ValueError("Link method not kown")
msnet.add_weighted_edges_from(new_edges)
if not self.keep_unconnected_nodes:
msnet.remove_nodes_from(list(nx.isolates(msnet)))
self.graph = msnet
[docs] def export_to_graphml(self, filename: str):
"""Save the network as .graphml file.
Parameters
----------
filename
Specify filename for exporting the graph.
"""
if not self.graph:
raise ValueError("No network found. Make sure to first run .create_network() step")
nx.write_graphml_lxml(self.graph, filename)
