refactor(gate.py): Synthesize gates from self

Instead of requesting gates in a loop, per #140 (comment)

Author: gegnew

cellengine/resources/gate.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 import importlib
-from cellengine.utils.types import ApplyTailoringRes
+from cellengine.utils.types import (
+    ApplyTailoringInsert,
+    ApplyTailoringUpdate,
+)
 from collections import defaultdict
 from math import pi
 from operator import itemgetter
@@ -127,26 +130,27 @@ def update_gate_family(experiment_id: str, gid: str, body: Dict) -> None:
         if res["nModified"] < 1:
             raise Warning("No gates updated.")
 
-    def apply_tailoring(self, fcs_file_ids: List[str]):
+    def apply_tailoring(self, fcs_file_ids: List[str]) -> Dict[str, List[Gate]]:
         """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)
+        payload = ce.APIClient().apply_tailoring(self.experiment_id, self, fcs_file_ids)
 
-    def _make_tailored_gates(self, payload: ApplyTailoringRes) -> 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]
+                    [ret[k].append(i["_id"]) for i in v]  # type: ignore
                 else:
-                    [
-                        ret[k].append(
-                            ce.APIClient().get_gate(self.experiment_id, i["_id"])
-                        )
-                        for i in v
-                    ]
+                    [ret[k].append(self._synthesize_gate(i)) for i in v]  # type: ignore
         return dict(ret)
 
+    def _synthesize_gate(
+        self,
+        payload: Union[ApplyTailoringInsert, ApplyTailoringUpdate],
+    ):
+        gate = self.to_dict()
+        gate.update(payload)
+        return Gate.from_dict(gate)
+
 
 class RectangleGate(Gate):
     @classmethod

cellengine/utils/types.py

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

tests/unit/resources/test_gates.py

@@ -1,7 +1,6 @@
 import json
 import pytest
 import responses
-from numpy import array
 
 from cellengine.resources.gate import (
     Gate,
@@ -12,7 +11,7 @@
     RangeGate,
     SplitGate,
 )
-from cellengine.utils.helpers import is_valid_id, normalize
+from cellengine.utils.helpers import is_valid_id
 from tests.unit.resources.test_population import population_tester
 from numpy import mean
 
@@ -731,54 +730,12 @@ def test_create_split_gate(client, ENDPOINT_BASE, experiment, scalesets, split_g
     assert split_gate.model["labels"] == [[-199.9, 0.916], [0.9, 0.916]]
 
 
-@responses.activate
-def test_make_tailored_gates(client, ENDPOINT_BASE, monkeypatch, rectangle_gate):
-    def mock_get_gate(exp_id, _id):
-        return "Gate"
-
-    monkeypatch.setattr(client, "get_gate", mock_get_gate)
-
-    responses.add(
-        responses.POST,
-        f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates",
-        status=201,
-        json=rectangle_gate,
-    )
-    gate = RectangleGate.create(
-        experiment_id=EXP_ID,
-        x_channel="FSC-A",
-        y_channel="FSC-W",
-        name="my gate",
-        x1=60000,
-        x2=200000,
-        y1=75000,
-        y2=215000,
-    )
-
-    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"},
+            {"_id": "updatedGateId1", "fcsFileId": "fileId1"},
+            {"_id": "updatedGateId2", "fcsFileId": "fileId2"},
         ],
         "inserted": [],
         "deleted": [{"_id": "deletedGateId", "fcsFileId": "fileId"}],
@@ -808,7 +765,11 @@ def mock_get_gate(exp_id, _id):
     )
 
     res = gate.apply_tailoring([fcs_files[0]["_id"]])
-    assert res == {
-        "updated": ["Gate", "Gate"],
-        "deleted": ["deletedGateId"],
-    }
+
+    assert res["deleted"] == ["deletedGateId"]
+    assert isinstance(res["updated"], list)
+    assert len(res["updated"]) == 2
+    assert res["updated"][0]._id == "updatedGateId1"
+    assert res["updated"][1]._id == "updatedGateId2"
+    assert res["updated"][0].fcs_file_id == "fileId1"
+    assert res["updated"][1].fcs_file_id == "fileId2"