feat: add experiment.import_resources()

Author: zbjornson

cellengine/resources/experiment.py

@@ -9,6 +9,11 @@
 
 from pandas.core.frame import DataFrame
 
+try:
+    from typing import TypedDict, NotRequired
+except ImportError:
+    from typing_extensions import TypedDict, NotRequired
+
 import cellengine as ce
 from cellengine.resources.attachment import Attachment
 from cellengine.resources.compensation import Compensation, UNCOMPENSATED, Compensations
@@ -31,6 +36,18 @@
 )
 
 
+ImportOpts = TypedDict(
+    "ImportOpts",
+    {
+        "populations": NotRequired[bool],
+        "illustrations": NotRequired[Union[bool, List[str]]],
+        "compensations": NotRequired[Union[bool, List[str]]],
+        "savedStatisticExports": NotRequired[Union[bool, List[str]]],
+        "annotations": NotRequired[Union[bool, List[str]]],
+    },
+)
+
+
 class Experiment:
     """The main container for an analysis. Don't construct directly; use
     [`Experiment.create`][cellengine.Experiment.create] or
@@ -374,6 +391,47 @@ def save_revision(self, description: str) -> None:
         self._properties["revisions"] = r.get("revisions")
         self._properties["deepUpdated"] = r.get("deepUpdated")
 
+    def import_resources(
+        self,
+        src_experiment_id: str,
+        what: ImportOpts,
+        channel_map: Optional[Dict[str, str]] = {},
+        dst_population_id: Optional[str] = None,
+    ) -> None:
+        """
+        Imports resources from another experiment.
+
+        Args:
+            src_experiment_id (str): The ID of the source experiment.
+            what (ImportOpts): A dictionary with the following optional keys:
+                - populations: Whether to import populations.
+                - illustrations: Whether to import illustrations (True = all,
+                  False = none), or a list of specific illustration IDs to import.
+                - compensations: Whether to import compensations (True = all,
+                  False = none), or a list of specific compensation IDs to import.
+                - savedStatisticExports: Whether to import saved statistic exports
+                  (True = all, False = none), or a list of specific export IDs to
+                  import.
+                - annotations: Whether to import annotations (True = all,
+                  False = none), or a list of specific annotation names to import.
+            channel_map (Dict[str, str]): A dictionary Object mapping channel
+                names from source experiment to destination experiment for
+                imported gates. Gates using channels not present in the map or
+                with channels set to the value "" will not be imported.
+                Populations are only imported if all of their required gates are
+                imported (i.e. the entire set of parents must be imported). This
+                does not affect compensation import.
+            dst_population_id (str): The ID of the destination parent population.
+                If not provided, the root population is used.
+
+        Use `experiment.gates` and similar to access the imported resources.
+        *Note: If it would be useful for this method to return the imported
+        resources, open a GitHub issue letting us know.*
+        """
+        ce.APIClient().import_experiment_resources(
+            self._id, src_experiment_id, what, channel_map, dst_population_id
+        )
+
     # Attachments
 
     @property

cellengine/utils/api_client/APIClient.py

@@ -24,7 +24,7 @@
 
 from ...resources.attachment import Attachment
 from ...resources.compensation import Compensation, Compensations, UNCOMPENSATED
-from ...resources.experiment import Experiment
+from ...resources.experiment import Experiment, ImportOpts
 from ...resources.fcs_file import FcsFile
 from ...resources.folder import Folder
 from ...resources.gate import (
@@ -358,6 +358,24 @@ def save_experiment_revision(self, _id, description: str) -> Dict:
             json={"description": description},
         )
 
+    def import_experiment_resources(
+        self,
+        experiment_id: str,
+        src_experiment_id: str,
+        what: ImportOpts,
+        channel_map: Optional[Dict[str, str]],
+        dst_population_id: Optional[str],
+    ) -> None:
+        self._post(
+            f"{self.base_url}/api/v1/experiments/{experiment_id}/importResources",
+            json={
+                "srcExperiment": src_experiment_id,
+                "channelMap": channel_map,
+                "dstPopulationId": dst_population_id,
+                "import": what,
+            },
+        )
+
     # ------------------------------- FCS Files --------------------------------
 
     def get_fcs_files(self, experiment_id, as_dict=False) -> List[FcsFile]:

tests/integration/test_experiment.py

@@ -119,6 +119,22 @@ def test_save_revision(blank_experiment):
     assert blank_experiment.deep_updated > preDU
 
 
+def test_experiment_import_resources(full_experiment):
+    dest = Experiment.create("dest")
+    dest.upload_fcs_file("tests/data/Specimen_001_A1_A01_MeOHperm(DL350neg).fcs")
+    dest.import_resources(
+        full_experiment["experiment"]._id,
+        {"populations": True, "compensations": True},
+        {
+            "FSC-A": "FSC-A",
+            "SSC-A": "SSC-A",
+        },
+    )
+    assert len(dest.gates) == 1
+    assert len(dest.populations) == 1
+    assert len(dest.compensations) == 1
+
+
 def test_experiment_upload_fcs_file(blank_experiment: Experiment):
     file = blank_experiment.upload_fcs_file(
         "tests/data/Specimen_001_A1_A01_MeOHperm(DL350neg).fcs"