forked from enviPath/enviPy
143 lines
4.8 KiB
Python
143 lines
4.8 KiB
Python
from uuid import UUID
|
|
|
|
from epdb.logic import PackageManager
|
|
from epdb.models import Pathway
|
|
|
|
from epapi.v1.errors import EPAPINotFoundError, EPAPIPermissionDeniedError
|
|
|
|
from .dto import (
|
|
PathwayCompoundDTO,
|
|
PathwayEdgeDTO,
|
|
PathwayExportDTO,
|
|
PathwayModelInfoDTO,
|
|
PathwayNodeDTO,
|
|
PathwayScenarioDTO,
|
|
)
|
|
|
|
|
|
def get_pathway_for_iuclid_export(user, pathway_uuid: UUID) -> PathwayExportDTO:
|
|
"""Return pathway data projected into DTOs for the IUCLID export consumer."""
|
|
try:
|
|
pathway = (
|
|
Pathway.objects.select_related("package", "setting", "setting__model")
|
|
.prefetch_related(
|
|
"node_set__default_node_label__compound__external_identifiers__database",
|
|
"node_set__scenarios",
|
|
"edge_set__start_nodes__default_node_label__compound",
|
|
"edge_set__end_nodes__default_node_label__compound",
|
|
)
|
|
.get(uuid=pathway_uuid)
|
|
)
|
|
except Pathway.DoesNotExist:
|
|
raise EPAPINotFoundError(f"Pathway with UUID {pathway_uuid} not found")
|
|
|
|
if not user or user.is_anonymous or not PackageManager.readable(user, pathway.package):
|
|
raise EPAPIPermissionDeniedError("Insufficient permissions to access this pathway.")
|
|
|
|
nodes: list[PathwayNodeDTO] = []
|
|
edges: list[PathwayEdgeDTO] = []
|
|
compounds_by_pk: dict[int, PathwayCompoundDTO] = {}
|
|
root_compound_pks: list[int] = []
|
|
|
|
for node in pathway.node_set.all().order_by("depth", "pk"):
|
|
cs = node.default_node_label
|
|
if cs is None:
|
|
continue
|
|
compound = cs.compound
|
|
|
|
cas_number = None
|
|
ec_number = None
|
|
for ext_id in compound.external_identifiers.all():
|
|
db_name = ext_id.database.name if ext_id.database else None
|
|
if db_name == "CAS" and cas_number is None:
|
|
cas_number = ext_id.identifier_value
|
|
elif db_name == "EC" and ec_number is None:
|
|
ec_number = ext_id.identifier_value
|
|
|
|
ai_for_node = []
|
|
scenario_entries: list[PathwayScenarioDTO] = []
|
|
for scenario in sorted(node.get_scenarios(), key=lambda item: item.pk):
|
|
ai_for_scenario = list(scenario.get_additional_information(direct_only=True))
|
|
ai_for_node.extend(ai_for_scenario)
|
|
scenario_entries.append(
|
|
PathwayScenarioDTO(
|
|
scenario_uuid=scenario.uuid,
|
|
name=scenario.name,
|
|
additional_info=ai_for_scenario,
|
|
)
|
|
)
|
|
|
|
nodes.append(
|
|
PathwayNodeDTO(
|
|
node_uuid=node.uuid,
|
|
compound_pk=compound.pk,
|
|
name=compound.name,
|
|
depth=node.depth,
|
|
smiles=cs.smiles,
|
|
cas_number=cas_number,
|
|
ec_number=ec_number,
|
|
additional_info=ai_for_node,
|
|
scenarios=scenario_entries,
|
|
)
|
|
)
|
|
|
|
if node.depth == 0 and compound.pk not in root_compound_pks:
|
|
root_compound_pks.append(compound.pk)
|
|
|
|
if compound.pk not in compounds_by_pk:
|
|
compounds_by_pk[compound.pk] = PathwayCompoundDTO(
|
|
pk=compound.pk,
|
|
name=compound.name,
|
|
smiles=cs.smiles,
|
|
cas_number=cas_number,
|
|
ec_number=ec_number,
|
|
)
|
|
|
|
for edge in pathway.edge_set.all():
|
|
start_compounds = {
|
|
n.default_node_label.compound.pk
|
|
for n in edge.start_nodes.all()
|
|
if n.default_node_label is not None
|
|
}
|
|
end_compounds = {
|
|
n.default_node_label.compound.pk
|
|
for n in edge.end_nodes.all()
|
|
if n.default_node_label is not None
|
|
}
|
|
|
|
probability = None
|
|
if edge.kv and edge.kv.get("probability") is not None:
|
|
try:
|
|
probability = float(edge.kv.get("probability"))
|
|
except (TypeError, ValueError):
|
|
probability = None
|
|
|
|
edges.append(
|
|
PathwayEdgeDTO(
|
|
edge_uuid=edge.uuid,
|
|
start_compound_pks=sorted(start_compounds),
|
|
end_compound_pks=sorted(end_compounds),
|
|
probability=probability,
|
|
)
|
|
)
|
|
|
|
model_info = None
|
|
if pathway.setting and pathway.setting.model:
|
|
model = pathway.setting.model
|
|
model_info = PathwayModelInfoDTO(
|
|
model_name=model.get_name(),
|
|
model_uuid=model.uuid,
|
|
software_name="enviPath",
|
|
software_version=None,
|
|
)
|
|
|
|
return PathwayExportDTO(
|
|
pathway_uuid=pathway.uuid,
|
|
pathway_name=pathway.get_name(),
|
|
compounds=list(compounds_by_pk.values()),
|
|
nodes=nodes,
|
|
edges=edges,
|
|
root_compound_pks=root_compound_pks,
|
|
model_info=model_info,
|
|
)
|