diff --git a/epdb/legacy_api.py b/epdb/legacy_api.py index a6f58ef2..37b789e8 100644 --- a/epdb/legacy_api.py +++ b/epdb/legacy_api.py @@ -46,6 +46,10 @@ from .models import ( Package = s.GET_PACKAGE_MODEL() +def get_package_for_read(user, package_uuid): + return PackageManager.get_package_by_id(user, package_uuid) + + def get_package_for_write(user, package_uuid): p = PackageManager.get_package_by_id(user, package_uuid) if not PackageManager.writable(user, p): @@ -2291,3 +2295,38 @@ def predict(request, np: Form[NonPersistent]): return 403, { "message": f"Getting Setting with id {np.setting_url} failed due to insufficient rights!" } + + +########## +# Export # +########## +class PackageExportInSchema(Schema): + package_uuid: str + additional_information_types: List[str] | None = None + + +@router.get("/export", response={200: Any, 403: Error}) +def export(request, q: Query[PackageExportInSchema]): + try: + p = get_package_for_read(request.user, q.package_uuid) + + from envipy_additional_information import registry + from utilities.misc import PathwayExporter + + ai_types = [] + if q.additional_information_types is not None: + for ai_type in q.additional_information_types: + if registry.get_model(ai_type) is None: + return 400, { + "message": f"Exporting Package with id {q.package_uuid} failed as {ai_type} is not a valid additional information type!" + } + ai_types.append(ai_type) + + exporter = PathwayExporter(p, add_infs_to_export=ai_types) + res = exporter.do_export() + + return res + except ValueError: + return 403, { + "message": f"Exporting Package with id {q.package_uuid} failed due to insufficient rights!" + } diff --git a/templates/objects/node.html b/templates/objects/node.html index 043342b4..0284f888 100644 --- a/templates/objects/node.html +++ b/templates/objects/node.html @@ -139,7 +139,7 @@ Model - {{ half_lifes.0.model }} + {{ half_lifes.0.model.value }} diff --git a/utilities/misc.py b/utilities/misc.py index 7a9992de..10b6a4b4 100644 --- a/utilities/misc.py +++ b/utilities/misc.py @@ -4,11 +4,13 @@ import hmac import json import logging import uuid +from abc import ABC, abstractmethod from collections import defaultdict from datetime import datetime -from typing import Any, Dict, List, Optional, TYPE_CHECKING +from typing import Any, Dict, List, Optional, TYPE_CHECKING, Type from django.conf import settings as s +from envipy_additional_information import EnviPyModel, UIConfig from ninja import Schema from pydantic import HttpUrl, ValidationError @@ -277,15 +279,20 @@ class PackageExportSchema(Schema): return AdditionalInformation.objects.filter(package=obj) -class PackageExporter: +class Exporter(ABC): def __init__(self, package: Package): self._raw_package = package def do_export(self): - return PackageExporter._export_package_as_json(self._raw_package) + return self._export() - @staticmethod - def _export_package_as_json(package: Package) -> Dict[str, Any]: + @abstractmethod + def _export(self): + pass + + +class PackageExporter(Exporter): + def _export(self) -> Dict[str, Any]: """ Dumps a Package and all its related objects as JSON. @@ -296,11 +303,126 @@ class PackageExporter: Dict containing the complete package data as JSON-serializable structure """ - data = PackageExportSchema.from_orm(package) + data = PackageExportSchema.from_orm(self._raw_package) return data.model_dump(mode="json") +class PathwayExporter(Exporter): + def __init__(self, package: Package, add_infs_to_export: List[str] = []): + super().__init__(package) + self._add_infs_to_export = add_infs_to_export + + def _flatten_additional_information(self, ai: AdditionalInformation) -> dict[str, Any]: + model_cls: Type[EnviPyModel] = type(ai.get()) + + def _flatten(d: dict, parent_key: str = "") -> dict[str, Any]: + items: dict[str, Any] = {} + for key, value in d.items(): + new_key = f"{parent_key}__{key.lower()}" if parent_key else key.lower() + if isinstance(value, dict): + items.update(_flatten(value, new_key)) + else: + items[new_key] = value + return items + + flat = _flatten(ai.data, ai.type) + + ui_class = getattr(model_cls, "UI", None) + if ui_class is None: + return flat + + for f in model_cls.model_fields: + ui_info = getattr(ui_class, f, None) + + if not isinstance(ui_info, UIConfig) or ui_info.unit is None: + continue + + flat[f"{model_cls.__name__}__{f}__unit"] = ui_info.unit + + return flat + + def _export(self): + from io import StringIO + from csv import DictWriter + + rows = [] + + for pw in self._raw_package.pathways.all(): + for n in pw.nodes: + for scen in pw.scenarios.all(): + row = { + "pathway_name": pw.name, + "pathway_id": str(pw.url), + "node_depth": n.depth, + "compound_id": str(n.default_node_label.compound.url), + "pubchem_ID": n.default_node_label.pubchem_compound_id, + "compound_name": n.default_node_label.compound.name, + "compound_smiles": n.default_node_label.smiles, + "scenario_id": str(scen.url), + "scenario_name": scen.name, + "scenario_type": scen.scenario_type, + "scenario_description": scen.description, + } + + if self._add_infs_to_export: + ai_qs = AdditionalInformation.objects.filter( + scenario=scen, type__in=self._add_infs_to_export + ) + else: + ai_qs = AdditionalInformation.objects.filter(scenario=scen) + + for ai in ai_qs: + if ai.type == "ProposedIntermediate" and ai.content_object == n: + row.update({"proposed_intermediate": True}) + elif ai.type == "SpikeCompound": + spike = {"SpikeCompound__url": ai.get().url} + + try: + struc = CompoundStructure.objects.get( + compound__package=self._raw_package, url=ai.get().url + ) + spike["SpikeCompound__smiles"] = struc.smiles + except Exception: + spike["SpikeCompound__smiles"] = None + + row.update(**spike) + else: + row.update(self._flatten_additional_information(ai)) + rows.append(row) + + # Get all header fields + all_header_fields = set() + for row in rows: + all_header_fields.update(row.keys()) + + # Per request the CSV should start with these fields + header = [ + "pathway_name", + "pathway_id", + "node_depth", + "compound_id", + "pubchem_ID", + "compound_name", + "compound_smiles", + "scenario_id", + "scenario_name", + "scenario_type", + "scenario_description", + ] + + # User remaining fields and place them after the predefined values in a sorted manner + remainder = sorted(list(all_header_fields.difference(set(header)))) + header.extend(remainder) + + buffer = StringIO() + writer = DictWriter(buffer, fieldnames=header, delimiter="\t") + writer.writeheader() + writer.writerows(rows) + buffer.seek(0) + return buffer.getvalue() + + class PackageImporter: def __init__(self, package: Dict[str, Any], preserve_uuids: bool = False): self.preserve_uuids = preserve_uuids