forked from enviPath/enviPy
Compare commits
2 Commits
cdd51fc7aa
...
d657c0285a
| Author | SHA1 | Date | |
|---|---|---|---|
| d657c0285a | |||
| f4c198981b |
@ -1805,12 +1805,51 @@ class SPathway(object):
|
|||||||
|
|
||||||
logger.info("Update done!")
|
logger.info("Update done!")
|
||||||
|
|
||||||
|
def compute_bayes_probabilities(self) -> Dict[SEdge, float]:
|
||||||
|
"""
|
||||||
|
Computes Bayes-adjusted probabilities for all edges in the pathway
|
||||||
|
by iterating level by level from depth 0 upwards, keyed on educt depth.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
A dict mapping each SEdge to its Bayes-adjusted probability.
|
||||||
|
"""
|
||||||
|
bayes_probs: Dict[SEdge, float] = {}
|
||||||
|
|
||||||
|
# Group edges by their educt depth
|
||||||
|
edges_by_depth: Dict[int, List[SEdge]] = {}
|
||||||
|
for edge in self.edges:
|
||||||
|
d = edge.educts[0].depth
|
||||||
|
edges_by_depth.setdefault(d, []).append(edge)
|
||||||
|
|
||||||
|
for depth in sorted(edges_by_depth.keys()):
|
||||||
|
for edge in edges_by_depth[depth]:
|
||||||
|
if depth == 0:
|
||||||
|
bayes_probs[edge] = edge.probability
|
||||||
|
else:
|
||||||
|
predecessor_edges = [e for e in self.edges if edge.educts[0] in e.products]
|
||||||
|
|
||||||
|
if not predecessor_edges or not all(
|
||||||
|
e in bayes_probs for e in predecessor_edges
|
||||||
|
):
|
||||||
|
# Predecessor not computed yet (e.g. same-depth product),
|
||||||
|
# fall back to raw probability
|
||||||
|
bayes_probs[edge] = edge.probability
|
||||||
|
else:
|
||||||
|
predecessor_avg = sum(bayes_probs[e] for e in predecessor_edges) / len(
|
||||||
|
predecessor_edges
|
||||||
|
)
|
||||||
|
bayes_probs[edge] = predecessor_avg * edge.probability
|
||||||
|
|
||||||
|
return bayes_probs
|
||||||
|
|
||||||
def to_json(self):
|
def to_json(self):
|
||||||
nodes = []
|
nodes = []
|
||||||
edges = []
|
edges = []
|
||||||
|
|
||||||
idx_lookup = {}
|
idx_lookup = {}
|
||||||
|
|
||||||
|
bayes_probs = self.compute_bayes_probabilities()
|
||||||
|
|
||||||
for i, smiles in enumerate(self.smiles_to_node):
|
for i, smiles in enumerate(self.smiles_to_node):
|
||||||
n = self.smiles_to_node[smiles]
|
n = self.smiles_to_node[smiles]
|
||||||
idx_lookup[smiles] = i
|
idx_lookup[smiles] = i
|
||||||
@ -1831,6 +1870,7 @@ class SPathway(object):
|
|||||||
|
|
||||||
if edge.probability:
|
if edge.probability:
|
||||||
e["probability"] = edge.probability
|
e["probability"] = edge.probability
|
||||||
|
e["multiGenProbability"] = bayes_probs[edge]
|
||||||
|
|
||||||
edges.append(e)
|
edges.append(e)
|
||||||
|
|
||||||
|
|||||||
@ -1918,6 +1918,21 @@ def package_reactions(request, package_uuid):
|
|||||||
reaction_name = request.POST.get("reaction-name")
|
reaction_name = request.POST.get("reaction-name")
|
||||||
reaction_description = request.POST.get("reaction-description")
|
reaction_description = request.POST.get("reaction-description")
|
||||||
reaction_smiles = request.POST.get("reaction-smiles")
|
reaction_smiles = request.POST.get("reaction-smiles")
|
||||||
|
|
||||||
|
if reaction_smiles is None or reaction_smiles.strip() == "":
|
||||||
|
return error(
|
||||||
|
request,
|
||||||
|
"Reaction SMILES is empty / missing",
|
||||||
|
"No reaction SMILES provided. Please provide a SMILES for the reaction.",
|
||||||
|
)
|
||||||
|
|
||||||
|
if not FormatConverter.is_valid_smirks(reaction_smiles):
|
||||||
|
return error(
|
||||||
|
request,
|
||||||
|
"Reaction SMILES is invalid",
|
||||||
|
f"The provided reactions SMILES {reaction_smiles} is invalid",
|
||||||
|
)
|
||||||
|
|
||||||
educts = reaction_smiles.split(">>")[0].split(".")
|
educts = reaction_smiles.split(">>")[0].split(".")
|
||||||
products = reaction_smiles.split(">>")[1].split(".")
|
products = reaction_smiles.split(">>")[1].split(".")
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user