feat(apiclient, gate): Add apply_tailoring function

cellengine/resources/experiment.py

@@ -189,17 +189,15 @@ def clone(self, props: Optional[Dict[str, Any]] = None) -> Experiment:
         """
         return ce.APIClient().clone_experiment(self._id, props)
 
-    @property
-    def delete(self):
+    def delete(self) -> None:
         """Marks the experiment as deleted.
 
         Deleted experiments are permanently deleted after approximately
         7 days. Until then, deleted experiments can be recovered.
         """
         self.deleted = datetime.today()
 
-    @property
-    def undelete(self):
+    def undelete(self) -> None:
         """Clear a scheduled deletion."""
         if self.deleted:
             del self.deleted

cellengine/resources/fcs_file.py

@@ -76,7 +76,7 @@ def annotations(self, val):
             self._annotations = val
 
     @property
-    def channels(self) -> List:
+    def channels(self) -> List[str]:
         """Return all channels in the file"""
         return [f["channel"] for f in self.panel]  # type: ignore
 

cellengine/resources/gate.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 import importlib
+from cellengine.utils.types import (
+    ApplyTailoringInsert,
+    ApplyTailoringUpdate,
+)
 from math import pi
 from operator import itemgetter
 from typing import Any, Dict, List, Optional, Union, Tuple, overload
@@ -13,7 +17,6 @@
 from numpy import array, mean, stack
 
 import cellengine as ce
-from cellengine.resources.fcs_file import FcsFile
 from cellengine.resources.population import Population
 from cellengine.utils import parse_fcs_file_args
 from cellengine.utils import converter, generate_id, readonly
@@ -57,6 +60,13 @@ def deep_update(source, overrides):
     return source
 
 
+class ApplyTailoringResult:
+    def __init__(self):
+        self.inserted: List[Gate] = []
+        self.updated: List[Gate] = []
+        self.deleted: List[str] = []
+
+
 @define(repr=False)
 class Gate:
     _id: str = field(on_setattr=readonly)
@@ -138,10 +148,22 @@ def update_gate_family(experiment_id: str, gid: str, body: Dict) -> None:
         if res["nModified"] < 1:
             raise Warning("No gates updated.")
 
-    def tailor_to(self, fcs_file: FcsFile):
-        self.tailored_per_file = True
-        self.fcs_file_id = fcs_file._id
-        self.update()
+    def apply_tailoring(self, fcs_file_ids: List[str]) -> ApplyTailoringResult:
+        """Apply this gate's tailoring (copy its geometry) to other FCS files."""
+        payload = ce.APIClient().apply_tailoring(self.experiment_id, self, fcs_file_ids)
+        ret = ApplyTailoringResult()
+        [ret.inserted.append(self._synthesize_gate(i)) for i in payload["inserted"]]
+        [ret.updated.append(self._synthesize_gate(i)) for i in payload["updated"]]
+        [ret.deleted.append(i["_id"]) for i in payload["deleted"]]
+        return ret
+
+    def _synthesize_gate(
+        self,
+        payload: Union[ApplyTailoringInsert, ApplyTailoringUpdate],
+    ):
+        gate = self.to_dict()
+        gate.update(payload)
+        return Gate.from_dict(gate)
 
 
 class RectangleGate(Gate):

cellengine/utils/api_client/APIClient.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+from cellengine.utils.types import ApplyTailoringRes
 from functools import lru_cache
 from getpass import getpass
 import importlib
@@ -603,12 +604,15 @@ def update_gate_family(self, experiment_id, gid, body: dict = None) -> dict:
             json=body,
         )
 
-    def tailor_to(self, experiment_id, gate_id, fcs_file_id):
-        """Tailor a gate to a file."""
-        gate = self.get_gate(experiment_id, gate_id, as_dict=True)
-        gate["tailoredPerFile"] = True
-        gate["fcsFileId"] = fcs_file_id
-        return self.update_entity(experiment_id, gate_id, "gates", gate)
+    def apply_tailoring(
+        self, experiment_id: str, gate: Gate, fcs_file_ids: List[str]
+    ) -> ApplyTailoringRes:
+        """Tailor a gate to a file or files."""
+        return self._post(
+            f"{self.base_url}/experiments/{experiment_id}/gates/applyTailored",
+            params={"gid": gate.gid},
+            json={"gate": gate.to_dict(), "fcsFileIds": fcs_file_ids},
+        )
 
     def get_plot(
         self,

cellengine/utils/types.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+import sys
+from typing import List
+
+if sys.version_info >= (3, 8):
+    from typing import TypedDict
+else:
+    from typing_extensions import TypedDict
+
+ApplyTailoringInsert = TypedDict("ApplyTailoringInsert", {"_id": str, "fcsFileId": str})
+ApplyTailoringUpdate = TypedDict("ApplyTailoringUpdate", {"_id": str, "fcsFileId": str})
+ApplyTailoringDelete = TypedDict("ApplyTailoringDelete", {"_id": str, "fcsFileId": str})
+ApplyTailoringRes = TypedDict(
+    "ApplyTailoringRes",
+    {
+        "inserted": List[ApplyTailoringInsert],
+        "updated": List[ApplyTailoringUpdate],
+        "deleted": List[ApplyTailoringDelete],
+    },
+)

requirements-dev.txt

@@ -1,4 +1,4 @@
-pytest~=5.3
+pytest~=7.1.3
 pytest-recording~=0.12
 pyright~=0.0.9
 responses~=0.10