feat(apiclient, gate): Add apply_tailoring function

Author: gegnew

cellengine/resources/gate.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
+from cellengine.utils.types import TailoringDict
+from collections import defaultdict
 from math import pi
 from typing import Dict, List, Optional
 
 from attr import define, field
 import numpy
 
 import cellengine as ce
-from cellengine.resources.fcs_file import FcsFile
 from cellengine.utils import parse_fcs_file_args
 from cellengine.utils import converter, generate_id, is_valid_id, readonly
 
@@ -89,10 +90,25 @@ def update_family(experiment_id, gid: str, body: Dict):
         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], **kwargs):
+        """Tailor this gate to a specific FCS file or files."""
+        res = ce.APIClient().apply_tailoring(self.experiment_id, self, fcs_file_ids)
+        return self._make_tailored_gates(res)
+
+    def _make_tailored_gates(self, payload: TailoringDict) -> Dict[str, List[Gate]]:
+        ret = defaultdict(list)
+        for k, v in payload.items():
+            if v:
+                if k == "deleted":
+                    [ret[k].append(i["_id"]) for i in v]
+                else:
+                    [
+                        ret[k].append(
+                            ce.APIClient().get_gate(self.experiment_id, i["_id"])
+                        )
+                        for i in v
+                    ]
+        return dict(ret)
 
 
 @define(repr=False)

cellengine/utils/api_client/APIClient.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+from cellengine.utils.types import TailoringDict
 from functools import lru_cache
 from getpass import getpass
 import json
@@ -513,12 +514,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]
+    ) -> TailoringDict:
+        """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,12 @@
+from __future__ import annotations
+
+from typing import Dict, List, TypedDict
+
+TailoringDict = TypedDict(
+    "TailoringDict",
+    {
+        "inserted": List[Dict[str, str]],
+        "updated": List[Dict[str, str]],
+        "deleted": List[Dict[str, str]],
+    },
+)

tests/unit/resources/test_gates.py

@@ -444,3 +444,54 @@ def test_create_split_gate(client, ENDPOINT_BASE, experiment, scalesets, split_g
     ]
     assert split_gate.names == ["my gate (L)", "my gate (R)"]
     assert split_gate.model["labels"] == [[-199.9, 0.916], [0.9, 0.916]]
+
+
+def test_make_tailored_gates(client, monkeypatch, rectangle_gate):
+    def mock_get_gate(exp_id, _id):
+        return "Gate"
+
+    monkeypatch.setattr(client, "get_gate", mock_get_gate)
+
+    gate = Gate.factory(rectangle_gate)
+
+    payload = {
+        "updated": [{"_id": "updatedGateId", "fcsFileId": "fileId"}],
+        "inserted": [],
+        "deleted": [{"_id": "deletedGateId", "fcsFileId": "fileId"}],
+    }
+
+    res = gate._make_tailored_gates(payload)
+    assert res == {
+        "updated": ["Gate"],
+        "deleted": ["deletedGateId"],
+    }
+
+
+@responses.activate
+def test_apply_tailoring(client, monkeypatch, ENDPOINT_BASE, rectangle_gate, fcs_files):
+    def mock_get_gate(exp_id, _id):
+        return "Gate"
+
+    monkeypatch.setattr(client, "get_gate", mock_get_gate)
+
+    payload = {
+        "updated": [
+            {"_id": "updatedGateId", "fcsFileId": "fileId"},
+            {"_id": "updatedGateId", "fcsFileId": "fileId"},
+        ],
+        "inserted": [],
+        "deleted": [{"_id": "deletedGateId", "fcsFileId": "fileId"}],
+    }
+    gate = Gate.factory(rectangle_gate)
+
+    responses.add(
+        responses.POST,
+        f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates/applyTailored?gid={gate.gid}",
+        json=payload,
+    )
+
+    res = gate.apply_tailoring([fcs_files[0]["_id"]])
+    assert res == {
+        "updated": ["Gate", "Gate"],
+        "deleted": ["deletedGateId"],
+    }

tests/unit/utils/test_api_client.py

@@ -291,3 +291,25 @@ def test_should_get_statistics(client, ENDPOINT_BASE, statistics):
         population_ids="some population id",
     )
     assert set(expected_query_body) == set(json.loads(responses.calls[0].request.body))
+
+
+@responses.activate
+def test_apply_tailoring(client, ENDPOINT_BASE, rectangle_gate):
+    gate = Gate.factory(rectangle_gate)
+    payload = {
+        "updated": [{"_id": "updatedGateId", "fcsFileId": "fileId"}],
+        "inserted": [],
+        "deleted": [{"_id": "deletedGateId", "fcsFileId": "fileId"}],
+    }
+    responses.add(
+        responses.POST,
+        f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates/applyTailored?gid={gate.gid}",
+        json=payload,
+    )
+
+    res = client.apply_tailoring(EXP_ID, gate, ["fileID"])
+    assert res == {
+        "updated": [{"_id": "updatedGateId", "fcsFileId": "fileId"}],
+        "inserted": [],
+        "deleted": [{"_id": "deletedGateId", "fcsFileId": "fileId"}],
+    }