Graphical Representation
@@ -140,7 +149,7 @@
diff --git a/templates/predict_pathway.html b/templates/predict_pathway.html
index d0cb1ba6..d3413ca4 100644
--- a/templates/predict_pathway.html
+++ b/templates/predict_pathway.html
@@ -212,7 +212,13 @@
e.preventDefault();
const button = this;
button.disabled = true;
- button.textContent = "Predicting...";
+
+ // Set text depending on mode
+ if (document.getElementById("predict-submit-button").innerText === "Build") {
+ button.textContent = "Building...";
+ } else {
+ button.textContent = "Predicting...";
+ }
// Get SMILES from either input or Ketcher
const smilesInput = document.getElementById("predict-smiles");
diff --git a/tests/test_compound_model.py b/tests/test_compound_model.py
index 83a7113f..575b463b 100644
--- a/tests/test_compound_model.py
+++ b/tests/test_compound_model.py
@@ -1,5 +1,7 @@
from django.conf import settings as s
from django.test import TestCase, override_settings
+
+from epdb.exceptions import InvalidSMILESException
from epdb.logic import PackageManager
from epdb.models import Compound, User, CompoundStructure
@@ -33,13 +35,13 @@ class CompoundTest(TestCase):
self.assertEqual(c.description, "No Desc")
def test_missing_smiles(self):
- with self.assertRaises(ValueError):
+ with self.assertRaises(InvalidSMILESException):
_ = Compound.create(self.package, smiles=None, name="Afoxolaner", description="No Desc")
- with self.assertRaises(ValueError):
+ with self.assertRaises(InvalidSMILESException):
_ = Compound.create(self.package, smiles="", name="Afoxolaner", description="No Desc")
- with self.assertRaises(ValueError):
+ with self.assertRaises(InvalidSMILESException):
_ = Compound.create(self.package, smiles=" ", name="Afoxolaner", description="No Desc")
def test_smiles_are_trimmed(self):
@@ -96,7 +98,7 @@ class CompoundTest(TestCase):
self.assertEqual(len(self.package.compounds), 1)
def test_wrong_smiles(self):
- with self.assertRaises(ValueError):
+ with self.assertRaises(InvalidSMILESException):
_ = Compound.create(
self.package,
smiles="C1C(=NOC1(C2=CC(=CC(=C2)Cl)C(F)(F)F)C(F)(F)F)C3=CC=C(C=CC=CC=C43)C(=O)NCC(=O)NCC(F)(F)F",
diff --git a/tests/test_reaction_model.py b/tests/test_reaction_model.py
index bc501cd5..96b40965 100644
--- a/tests/test_reaction_model.py
+++ b/tests/test_reaction_model.py
@@ -1,6 +1,7 @@
from django.conf import settings as s
from django.test import TestCase, override_settings
+from epdb.exceptions import InvalidSMILESException
from epdb.logic import PackageManager
from epdb.models import Compound, User, Reaction, Rule
@@ -163,7 +164,7 @@ class ReactionTest(TestCase):
self.assertEqual(len(self.package.reactions), 1)
def test_wrong_smiles(self):
- with self.assertRaises(ValueError):
+ with self.assertRaises(InvalidSMILESException):
_ = Reaction.create(
package=self.package,
name="Eawag BBD reaction r0001",
diff --git a/tests/views/test_model_views.py b/tests/views/test_model_views.py
index 12cee202..dabf18c2 100644
--- a/tests/views/test_model_views.py
+++ b/tests/views/test_model_views.py
@@ -46,6 +46,14 @@ class ModelViewTest(TestCase):
)
expected = [
+ {
+ "products": [["CCN(CC)C(=O)C1=CC(C=O)=CC=C1"]],
+ "probability": 0.75,
+ "btrule": {
+ "url": "http://localhost:8000/package/1869d3f0-60bb-41fd-b6f8-afa75ffb09d3/simple-ambit-rule/2f2e0c39-e109-4836-959f-2bda2524f022",
+ "name": "bt0001-3568",
+ },
+ },
{
"products": [["O=C(O)C1=CC(CO)=CC=C1", "CCNCC"]],
"probability": 0.25,
@@ -62,14 +70,6 @@ class ModelViewTest(TestCase):
"name": "bt0243-4301",
},
},
- {
- "products": [["CCN(CC)C(=O)C1=CC(C=O)=CC=C1"]],
- "probability": 0.75,
- "btrule": {
- "url": "http://localhost:8000/package/1869d3f0-60bb-41fd-b6f8-afa75ffb09d3/simple-ambit-rule/2f2e0c39-e109-4836-959f-2bda2524f022",
- "name": "bt0001-3568",
- },
- },
]
actual = response.json()["pred"]
diff --git a/utilities/misc.py b/utilities/misc.py
index 6083c831..e1a84dec 100644
--- a/utilities/misc.py
+++ b/utilities/misc.py
@@ -6,12 +6,15 @@ import logging
import uuid
from collections import defaultdict
from datetime import datetime
-from typing import Any, Dict, List, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
from django.conf import settings as s
from django.db import transaction
+from ninja import Schema
+from pydantic import HttpUrl
from epdb.models import (
+ AdditionalInformation,
Compound,
CompoundStructure,
Edge,
@@ -28,7 +31,6 @@ from epdb.models import (
Rule,
RuleBasedRelativeReasoning,
Scenario,
- SequentialRule,
Setting,
SimpleAmbitRule,
SimpleRDKitRule,
@@ -43,408 +45,235 @@ if TYPE_CHECKING:
from epdb.logic import SPathway
-class PackageExporter:
- def __init__(
- self,
- package: Package,
- include_models: bool = False,
- include_external_identifiers: bool = True,
- ):
- self._raw_package = package
- self.include_modes = include_models
- self.include_external_identifiers = include_external_identifiers
+class LicenseExportSchema(Schema):
+ cc_string: str
+ link: HttpUrl
+ image_link: HttpUrl
- def do_export(self):
- return PackageExporter._export_package_as_json(
- self._raw_package, self.include_modes, self.include_external_identifiers
- )
+
+##############
+# RefSchemas #
+##############
+class RefExportSchema(Schema):
+ uuid: str
+ url: str
@staticmethod
- def _export_package_as_json(
- package: Package, include_models: bool = False, include_external_identifiers: bool = True
- ) -> Dict[str, Any]:
+ def resolve_uuid(obj):
+ value = obj.get("uuid") if isinstance(obj, dict) else obj.uuid
+ return str(value)
+
+
+class RefCompoundExportSchema(RefExportSchema): ...
+
+
+class RefCompoundStructureExportSchema(RefExportSchema): ...
+
+
+class RefReactionExportSchema(RefExportSchema): ...
+
+
+class RefRuleExportSchema(RefExportSchema): ...
+
+
+class RefNodeExportSchema(RefExportSchema): ...
+
+
+class RefEdgeExportSchema(RefExportSchema): ...
+
+
+class RefPathwayExportSchema(RefExportSchema): ...
+
+
+class RefScenarioExportSchema(RefExportSchema): ...
+
+
+############
+# Compound #
+############
+class CompoundExportSchema(RefCompoundExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ default_structure: RefCompoundStructureExportSchema
+ structures: List["CompoundStructureExportSchema"]
+ scenarios: List[RefScenarioExportSchema]
+
+
+class CompoundStructureExportSchema(RefCompoundStructureExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ smiles: str
+ molfile: Optional[str]
+ normalized_structure: bool
+ scenarios: List[RefScenarioExportSchema]
+
+
+############
+# Reaction #
+############
+class ReactionExportSchema(RefReactionExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ educts: List[RefCompoundStructureExportSchema]
+ products: List[RefCompoundStructureExportSchema]
+ rules: List[RefRuleExportSchema]
+ multi_step: bool
+ medline_references: List[str] | None
+ scenarios: List[RefScenarioExportSchema]
+
+
+#########
+# Rules #
+#########
+class RuleExportSchema(RefRuleExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ smirks: str
+ reactant_filter_smarts: Optional[str]
+ product_filter_smarts: Optional[str]
+ scenarios: List[RefScenarioExportSchema]
+
+
+class ParallelRuleExportSchema(RefRuleExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ simple_rules: List[RefRuleExportSchema]
+ scenarios: List[RefScenarioExportSchema]
+
+
+###########################
+# Pathway / Nodes / Edges #
+###########################
+class NodeExportSchema(RefNodeExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ default_node_label: RefCompoundStructureExportSchema
+ node_labels: List[RefCompoundStructureExportSchema]
+ depth: int
+ stereo_removed: bool
+ scenarios: List[RefScenarioExportSchema]
+
+
+class EdgeExportSchema(RefEdgeExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ edge_label: RefReactionExportSchema
+ start_nodes: List[RefNodeExportSchema]
+ end_nodes: List[RefNodeExportSchema]
+ scenarios: List[RefScenarioExportSchema]
+
+
+class PathwayExportSchema(RefPathwayExportSchema):
+ name: str
+ description: str
+ aliases: List[str]
+ predicted: bool
+ nodes: List[NodeExportSchema]
+ edges: List[EdgeExportSchema]
+ scenarios: List[RefScenarioExportSchema]
+
+
+class ScenarioExportSchema(RefScenarioExportSchema):
+ name: str
+ description: str
+ scenario_date: str
+ scenario_type: str
+
+
+class AdditionalInformationExportSchema(RefExportSchema):
+ type: str
+ data: dict
+ scenario: RefScenarioExportSchema | None = None
+ attach_object: RefExportSchema | None = None
+
+ @staticmethod
+ def resolve_attach_object(obj):
+ if isinstance(obj, dict):
+ if obj.get("attach_object") is None:
+ return None
+ return RefExportSchema.model_validate(obj["attach_object"])
+ return obj.content_object
+
+
+###########
+# Package #
+###########
+class PackageExportSchema(Schema):
+ name: str
+ description: str
+ uuid: str
+ reviewed: bool
+ license: LicenseExportSchema | None
+ compounds: List[CompoundExportSchema]
+ reactions: List[ReactionExportSchema]
+ simple_rules: List[RuleExportSchema]
+ composite_rules: List[ParallelRuleExportSchema]
+ pathways: List[PathwayExportSchema]
+ scenarios: List[ScenarioExportSchema]
+ additional_information: List[AdditionalInformationExportSchema]
+
+ @staticmethod
+ def resolve_uuid(obj):
+ value = obj.get("uuid") if isinstance(obj, dict) else obj.uuid
+ return str(value)
+
+ @staticmethod
+ def resolve_simple_rules(obj):
+ if isinstance(obj, dict):
+ result = []
+ for r in obj.get("simple_rules", []):
+ result.append(RuleExportSchema.model_validate(r))
+ return result
+ return SimpleAmbitRule.objects.filter(package=obj)
+
+ @staticmethod
+ def resolve_composite_rules(obj):
+ if isinstance(obj, dict):
+ result = []
+ for r in obj.get("composite_rules", []):
+ result.append(ParallelRuleExportSchema.model_validate(r))
+ return result
+ return ParallelRule.objects.filter(package=obj)
+
+ @staticmethod
+ def resolve_additional_information(obj):
+ if isinstance(obj, dict):
+ result = []
+ for ai in obj.get("additional_information", []):
+ result.append(AdditionalInformationExportSchema.model_validate(ai))
+ return result
+ return AdditionalInformation.objects.filter(package=obj)
+
+
+class PackageExporter:
+ def __init__(self, package: Package):
+ self._raw_package = package
+
+ def do_export(self):
+ return PackageExporter._export_package_as_json(self._raw_package)
+
+ @staticmethod
+ def _export_package_as_json(package: Package) -> Dict[str, Any]:
"""
Dumps a Package and all its related objects as JSON.
Args:
package: The Package instance to dump
- include_models: Whether to include EPModel objects
- include_external_identifiers: Whether to include external identifiers
Returns:
Dict containing the complete package data as JSON-serializable structure
"""
- def serialize_base_object(
- obj, include_aliases: bool = True, include_scenarios: bool = True
- ) -> Dict[str, Any]:
- """Serialize common EnviPathModel fields"""
- base_dict = {
- "uuid": str(obj.uuid),
- "name": obj.name,
- "description": obj.description,
- "url": obj.url,
- "kv": obj.kv,
- }
+ data = PackageExportSchema.from_orm(package)
- # Add aliases if the object has them
- if include_aliases and hasattr(obj, "aliases"):
- base_dict["aliases"] = obj.aliases
-
- # Add scenarios if the object has them
- if include_scenarios and hasattr(obj, "scenarios"):
- base_dict["scenarios"] = [
- {"uuid": str(s.uuid), "url": s.url} for s in obj.scenarios.all()
- ]
-
- return base_dict
-
- def serialize_external_identifiers(obj) -> List[Dict[str, Any]]:
- """Serialize external identifiers for an object"""
- if not include_external_identifiers or not hasattr(obj, "external_identifiers"):
- return []
-
- identifiers = []
- for ext_id in obj.external_identifiers.all():
- identifier_dict = {
- "uuid": str(ext_id.uuid),
- "database": {
- "uuid": str(ext_id.database.uuid),
- "name": ext_id.database.name,
- "base_url": ext_id.database.base_url,
- },
- "identifier_value": ext_id.identifier_value,
- "url": ext_id.url,
- "is_primary": ext_id.is_primary,
- }
- identifiers.append(identifier_dict)
- return identifiers
-
- # Start with the package itself
- result = serialize_base_object(package, include_aliases=True, include_scenarios=True)
- result["reviewed"] = package.reviewed
-
- # # Add license information
- # if package.license:
- # result['license'] = {
- # 'uuid': str(package.license.uuid),
- # 'name': package.license.name,
- # 'link': package.license.link,
- # 'image_link': package.license.image_link
- # }
- # else:
- # result['license'] = None
-
- # Initialize collections
- result.update(
- {
- "compounds": [],
- "structures": [],
- "rules": {"simple_rules": [], "parallel_rules": [], "sequential_rules": []},
- "reactions": [],
- "pathways": [],
- "nodes": [],
- "edges": [],
- "scenarios": [],
- "models": [],
- }
- )
-
- print(f"Exporting package: {package.name}")
-
- # Export compounds
- print("Exporting compounds...")
- for compound in package.compounds.prefetch_related("default_structure").order_by("url"):
- compound_dict = serialize_base_object(
- compound, include_aliases=True, include_scenarios=True
- )
-
- if compound.default_structure:
- compound_dict["default_structure"] = {
- "uuid": str(compound.default_structure.uuid),
- "url": compound.default_structure.url,
- }
- else:
- compound_dict["default_structure"] = None
-
- compound_dict["external_identifiers"] = serialize_external_identifiers(compound)
- result["compounds"].append(compound_dict)
-
- # Export compound structures
- print("Exporting compound structures...")
- compound_structures = (
- CompoundStructure.objects.filter(compound__package=package)
- .select_related("compound")
- .order_by("url")
- )
-
- for structure in compound_structures:
- structure_dict = serialize_base_object(
- structure, include_aliases=True, include_scenarios=True
- )
- structure_dict.update(
- {
- "compound": {
- "uuid": str(structure.compound.uuid),
- "url": structure.compound.url,
- },
- "smiles": structure.smiles,
- "canonical_smiles": structure.canonical_smiles,
- "inchikey": structure.inchikey,
- "normalized_structure": structure.normalized_structure,
- "external_identifiers": serialize_external_identifiers(structure),
- }
- )
- result["structures"].append(structure_dict)
-
- # Export rules
- print("Exporting rules...")
-
- # Simple rules (including SimpleAmbitRule and SimpleRDKitRule)
- for rule in SimpleRule.objects.filter(package=package).order_by("url"):
- rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
-
- # Add specific fields for SimpleAmbitRule
- if isinstance(rule, SimpleAmbitRule):
- rule_dict.update(
- {
- "rule_type": "SimpleAmbitRule",
- "smirks": rule.smirks,
- "reactant_filter_smarts": rule.reactant_filter_smarts or "",
- "product_filter_smarts": rule.product_filter_smarts or "",
- }
- )
- elif isinstance(rule, SimpleRDKitRule):
- rule_dict.update(
- {"rule_type": "SimpleRDKitRule", "reaction_smarts": rule.reaction_smarts}
- )
- else:
- rule_dict["rule_type"] = "SimpleRule"
-
- result["rules"]["simple_rules"].append(rule_dict)
-
- # Parallel rules
- for rule in (
- ParallelRule.objects.filter(package=package)
- .prefetch_related("simple_rules")
- .order_by("url")
- ):
- rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
- rule_dict["rule_type"] = "ParallelRule"
- rule_dict["simple_rules"] = [
- {"uuid": str(sr.uuid), "url": sr.url} for sr in rule.simple_rules.all()
- ]
- result["rules"]["parallel_rules"].append(rule_dict)
-
- # Sequential rules
- for rule in (
- SequentialRule.objects.filter(package=package)
- .prefetch_related("simple_rules")
- .order_by("url")
- ):
- rule_dict = serialize_base_object(rule, include_aliases=True, include_scenarios=True)
- rule_dict["rule_type"] = "SequentialRule"
- rule_dict["simple_rules"] = [
- {
- "uuid": str(sr.uuid),
- "url": sr.url,
- "order_index": sr.sequentialruleordering_set.get(
- sequential_rule=rule
- ).order_index,
- }
- for sr in rule.simple_rules.all()
- ]
- result["rules"]["sequential_rules"].append(rule_dict)
-
- # Export reactions
- print("Exporting reactions...")
- for reaction in package.reactions.prefetch_related("educts", "products", "rules").order_by(
- "url"
- ):
- reaction_dict = serialize_base_object(
- reaction, include_aliases=True, include_scenarios=True
- )
- reaction_dict.update(
- {
- "educts": [{"uuid": str(e.uuid), "url": e.url} for e in reaction.educts.all()],
- "products": [
- {"uuid": str(p.uuid), "url": p.url} for p in reaction.products.all()
- ],
- "rules": [{"uuid": str(r.uuid), "url": r.url} for r in reaction.rules.all()],
- "multi_step": reaction.multi_step,
- "medline_references": reaction.medline_references,
- "external_identifiers": serialize_external_identifiers(reaction),
- }
- )
- result["reactions"].append(reaction_dict)
-
- # Export pathways
- print("Exporting pathways...")
- for pathway in package.pathways.order_by("url"):
- pathway_dict = serialize_base_object(
- pathway, include_aliases=True, include_scenarios=True
- )
-
- # Add setting reference if exists
- if hasattr(pathway, "setting") and pathway.setting:
- pathway_dict["setting"] = {
- "uuid": str(pathway.setting.uuid),
- "url": pathway.setting.url,
- }
- else:
- pathway_dict["setting"] = None
-
- result["pathways"].append(pathway_dict)
-
- # Export nodes
- print("Exporting nodes...")
- pathway_nodes = (
- Node.objects.filter(pathway__package=package)
- .select_related("pathway", "default_node_label")
- .prefetch_related("node_labels", "out_edges")
- .order_by("url")
- )
-
- for node in pathway_nodes:
- node_dict = serialize_base_object(node, include_aliases=True, include_scenarios=True)
- node_dict.update(
- {
- "pathway": {"uuid": str(node.pathway.uuid), "url": node.pathway.url},
- "default_node_label": {
- "uuid": str(node.default_node_label.uuid),
- "url": node.default_node_label.url,
- },
- "node_labels": [
- {"uuid": str(label.uuid), "url": label.url}
- for label in node.node_labels.all()
- ],
- "out_edges": [
- {"uuid": str(edge.uuid), "url": edge.url} for edge in node.out_edges.all()
- ],
- "depth": node.depth,
- }
- )
- result["nodes"].append(node_dict)
-
- # Export edges
- print("Exporting edges...")
- pathway_edges = (
- Edge.objects.filter(pathway__package=package)
- .select_related("pathway", "edge_label")
- .prefetch_related("start_nodes", "end_nodes")
- .order_by("url")
- )
-
- for edge in pathway_edges:
- edge_dict = serialize_base_object(edge, include_aliases=True, include_scenarios=True)
- edge_dict.update(
- {
- "pathway": {"uuid": str(edge.pathway.uuid), "url": edge.pathway.url},
- "edge_label": {"uuid": str(edge.edge_label.uuid), "url": edge.edge_label.url},
- "start_nodes": [
- {"uuid": str(node.uuid), "url": node.url} for node in edge.start_nodes.all()
- ],
- "end_nodes": [
- {"uuid": str(node.uuid), "url": node.url} for node in edge.end_nodes.all()
- ],
- }
- )
- result["edges"].append(edge_dict)
-
- # Export scenarios
- print("Exporting scenarios...")
- for scenario in package.scenarios.order_by("url"):
- scenario_dict = serialize_base_object(
- scenario, include_aliases=False, include_scenarios=False
- )
- scenario_dict.update(
- {
- "scenario_date": scenario.scenario_date,
- "scenario_type": scenario.scenario_type,
- "parent": {"uuid": str(scenario.parent.uuid), "url": scenario.parent.url}
- if scenario.parent
- else None,
- "additional_information": scenario.additional_information,
- }
- )
- result["scenarios"].append(scenario_dict)
-
- # Export models
- if include_models:
- print("Exporting models...")
- package_models = (
- package.models.select_related("app_domain")
- .prefetch_related("rule_packages", "data_packages", "eval_packages")
- .order_by("url")
- )
-
- for model in package_models:
- model_dict = serialize_base_object(
- model, include_aliases=True, include_scenarios=False
- )
-
- # Common fields for PackageBasedModel
- if hasattr(model, "rule_packages"):
- model_dict.update(
- {
- "rule_packages": [
- {"uuid": str(p.uuid), "url": p.url}
- for p in model.rule_packages.all()
- ],
- "data_packages": [
- {"uuid": str(p.uuid), "url": p.url}
- for p in model.data_packages.all()
- ],
- "eval_packages": [
- {"uuid": str(p.uuid), "url": p.url}
- for p in model.eval_packages.all()
- ],
- "threshold": model.threshold,
- "eval_results": model.eval_results,
- "model_status": model.model_status,
- }
- )
-
- if model.app_domain:
- model_dict["app_domain"] = {
- "uuid": str(model.app_domain.uuid),
- "url": model.app_domain.url,
- }
- else:
- model_dict["app_domain"] = None
-
- # Specific fields for different model types
- if isinstance(model, RuleBasedRelativeReasoning):
- model_dict.update(
- {
- "model_type": "RuleBasedRelativeReasoning",
- "min_count": model.min_count,
- "max_count": model.max_count,
- }
- )
- elif isinstance(model, MLRelativeReasoning):
- model_dict["model_type"] = "MLRelativeReasoning"
- elif isinstance(model, EnviFormer):
- model_dict["model_type"] = "EnviFormer"
- else:
- model_dict["model_type"] = "EPModel"
-
- result["models"].append(model_dict)
-
- print(f"Export completed for package: {package.name}")
- print(f"- Compounds: {len(result['compounds'])}")
- print(f"- Structures: {len(result['structures'])}")
- print(f"- Simple rules: {len(result['rules']['simple_rules'])}")
- print(f"- Parallel rules: {len(result['rules']['parallel_rules'])}")
- print(f"- Sequential rules: {len(result['rules']['sequential_rules'])}")
- print(f"- Reactions: {len(result['reactions'])}")
- print(f"- Pathways: {len(result['pathways'])}")
- print(f"- Nodes: {len(result['nodes'])}")
- print(f"- Edges: {len(result['edges'])}")
- print(f"- Scenarios: {len(result['scenarios'])}")
- print(f"- Models: {len(result['models'])}")
-
- return result
+ return data.model_dump(mode="json")
class PackageImporter: