diff --git a/.vscode/settings.json b/.vscode/settings.json index e1d7b7b1..4a15ab62 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -3,5 +3,7 @@ "cellengine", "cytometry", "scaleset" - ] + ], + "python.linting.enabled": true, + "python.linting.flake8Enabled": true } \ No newline at end of file diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 00000000..b08142d6 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,22 @@ +## Contributing + +### Running tests + +Unit tests run with `python3 -m pytest`. + +The integration tests need CellEngine credentials to run. In CI, these are +provided by GitHub secrets. Locally, you need to set environment variables: + +```sh +CELLENGINE_USERNAME="..." +CELLENGINE_PASSWORD="..." +# for running tests that require S3: +S3_ACCESS_KEY="..." +S3_SECRET_KEY="..." +# Run the integration tests: +python3 -m pytest ./tests/integration/test_integration.py +``` + +### Conventions + +* Error message text should end with a period. diff --git a/cellengine/payloads/gate.py b/cellengine/payloads/gate.py deleted file mode 100644 index 7532fdff..00000000 --- a/cellengine/payloads/gate.py +++ /dev/null @@ -1,81 +0,0 @@ -import attr -from abc import ABC -from munch import Munch, munchify - -from cellengine.utils.helpers import GetSet - - -@attr.s(repr=False, slots=True) -class _Gate(ABC): - """ - A class containing Gate resource properties, and - an abstract base class for specific gate types. - - Args: (For all gate types, refer to help for each gate for args - specific to that gate.) - - experiment_id: The ID of the experiment to which to add the gate. - Use when calling this as a static method; not needed when calling - from an Experiment object - name: The name of the gate - x_channel: The name of the x channel to which the gate applies. - gid: Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, you must specify the gid of the global tailored gate. - parent_population_id: ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population: Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file: Whether or not this gate is tailored per FCS file. - fcs_file_id: ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file: Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - locked: Prevents modification of the gate via the web interface. - create_population: Automatically create corresponding population. - """ - - def __repr__(self): - if self.name: - name = self.name - else: - name = str(self.names) - return "{}(_id='{}', name='{}')".format(self.type, self._id, name) - - _id = GetSet("_id", read_only=True) - - _properties = attr.ib(default={}) - - experiment_id = GetSet("experimentId", read_only=True) - - fcs_file_id = GetSet("fcsFileId") - - gid = GetSet("gid") - - name = GetSet("name") - - names = GetSet("names") - - parent_population_id = GetSet("parentPopulationId") - - tailored_per_file = GetSet("tailoredPerFile") - - type = GetSet("type", read_only=True) - - x_channel = GetSet("xChannel") - - y_channel = GetSet("yChannel") - - @property - def model(self): - model = self._properties["model"] - if type(model) is not Munch: - self._properties["model"] = munchify(model) - return munchify(model) diff --git a/cellengine/payloads/gate_utils/__init__.py b/cellengine/payloads/gate_utils/__init__.py deleted file mode 100644 index 7e6c5938..00000000 --- a/cellengine/payloads/gate_utils/__init__.py +++ /dev/null @@ -1,9 +0,0 @@ -# flake8: noqa - -from .utils import format_common_gate -from .rectangle_gate import format_rectangle_gate -from .polygon_gate import format_polygon_gate -from .ellipse_gate import format_ellipse_gate -from .range_gate import format_range_gate -from .split_gate import format_split_gate -from .quadrant_gate import format_quadrant_gate diff --git a/cellengine/payloads/gate_utils/ellipse_gate.py b/cellengine/payloads/gate_utils/ellipse_gate.py deleted file mode 100644 index 9c1e5bd0..00000000 --- a/cellengine/payloads/gate_utils/ellipse_gate.py +++ /dev/null @@ -1,105 +0,0 @@ -from typing import List - -from cellengine.payloads.gate_utils import format_common_gate -from cellengine.utils.generate_id import generate_id - - -def format_ellipse_gate( - experiment_id: str, - x_channel: str, - y_channel: str, - name: str, - x: float, - y: float, - angle: float, - major: float, - minor: float, - label: List = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """Formats an ellipse gate for posting to the CellEngine API. - - Args: - x_channel (str): The name of the x channel to which the gate applies. - y_channel (str): The name of the y channel to which the gate applies. - name (str): The name of the gate - x (float): The x centerpoint of the gate. - y (float): The y centerpoint of the gate. - angle (float): The angle of the ellipse in radians. - major (float): The major radius of the ellipse. - minor (float): The minor radius of the ellipse. - label (float, optional): Position of the label. Defaults to the - midpoint of the gate. - gid (str, optional): Group ID of the gate, used for tailoring. If this - is not specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be - specified. - locked (bool, optional): Prevents modification of the gate via the web - interface. - parent_population_id (Optional[str]): ID of the parent population. Use - ``None`` for the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str, optional): Name of the parent population. An - attempt will be made to find the population by name. If zero or - more than one population exists with the name, an error will be - thrown. If specified, do not specify ``parent_population_id``. - tailored_per_file (bool, optional): Whether or not this gate is - tailored per FCS file. fcs_file_id (str, optional): ID of FCS - file, if tailored per file. Use ``None`` for the global gate in a - tailored gate group. If specified, do not specify ``fcs_file``. - fcs_file (str, optional): Name of FCS file, if tailored per file. An - attempt will be made to find the file by name. If zero or more than - one file exists with the name, an error will be thrown. Looking up - files by name is slower than using the ID, as this requires - additional requests to the server. If specified, do not specify - ``fcs_file_id``. - create_population (optional, bool): Automatically create corresponding - population. - - Returns: - EllipseGate: An EllipseGate object. - - Examples: - ```python - cellengine.Gate.create_ellipse_gate(experiment_id, x_channel="FSC-A", - y_channel="FSC-W", name="my gate", x=260000, y=64000, angle=0, - major=120000, minor=70000) - ``` - """ - if label == []: - label = [x, y] - if gid is None: - gid = generate_id() - - model = { - "locked": locked, - "label": label, - "ellipse": {"angle": angle, "major": major, "minor": minor, "center": [x, y]}, - } - - body = { - "experimentId": experiment_id, - "name": name, - "type": "EllipseGate", - "gid": gid, - "xChannel": x_channel, - "yChannel": y_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/polygon_gate.py b/cellengine/payloads/gate_utils/polygon_gate.py deleted file mode 100644 index e9330e83..00000000 --- a/cellengine/payloads/gate_utils/polygon_gate.py +++ /dev/null @@ -1,95 +0,0 @@ -import numpy -from typing import List - -from cellengine.utils.generate_id import generate_id -from cellengine.payloads.gate_utils import format_common_gate - - -def format_polygon_gate( - experiment_id: str, - x_channel: str, - y_channel: str, - name: str, - vertices: List[float], - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """Formats a polygon gate for posting to the CellEngine API. - - Args: - x_channel (str): The name of the x channel to which the gate applies. - y_channel (str): The name of the y channel to which the gate applies. - vertices (list): List of coordinates, like [[x,y], [x,y], ...] - label (str): Position of the label. Defaults to the midpoint of the gate. - name (str): The name of the gate - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - locked (bool): Prevents modification of the gate via the web interface. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - - Returns: - A PolygonGate object. - - Example: - ```python - experiment.create_polygon_gate(x_channel="FSC-A", - y_channel="FSC-W", name="my gate", vertices=[[1,4], [2,5], [3,6]]) - ``` - """ - if label == []: - label = [ - numpy.mean([item[0] for item in vertices]), - numpy.mean([item[1] for item in vertices]), - ] - if gid is None: - gid = generate_id() - - model = { - "locked": locked, - "label": label, - "polygon": {"vertices": vertices}, - } - - body = { - "experimentId": experiment_id, - "name": name, - "type": "PolygonGate", - "gid": gid, - "xChannel": x_channel, - "yChannel": y_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/quadrant_gate.py b/cellengine/payloads/gate_utils/quadrant_gate.py deleted file mode 100644 index d546ab0a..00000000 --- a/cellengine/payloads/gate_utils/quadrant_gate.py +++ /dev/null @@ -1,136 +0,0 @@ -from typing import List -from math import pi - -import cellengine as ce -from cellengine.utils.generate_id import generate_id -from cellengine.payloads.gate_utils import format_common_gate - - -def format_quadrant_gate( - experiment_id: str, - x_channel: str, - y_channel: str, - name: str, - x: float, - y: float, - labels: List[str] = [], - skewable: bool = False, - angles: List[float] = [0, pi / 2, pi, 3 * pi / 2], - gid: str = None, - gids: List[str] = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """Formats a quadrant gate for posting to the CellEngine API. - - Quadrant gates have four sectors (upper-right, upper-left, lower-left, - lower-right), each with a unique gid and name. - - Args: - x_channel (str): The name of the x channel to which the gate applies. - y_channel (str): The name of the y channel to which the gate applies. - name (str): The name of the gate - x (float): The x coordinate of the center point (after the channel's scale has - been applied). - y (float): The y coordinate (after the channel's scale has been applied). - labels (list): Positions of the quadrant labels. A list of four length-2 - vectors in the order: UR, UL, LL, LR. These are set automatically to - the plot corners. - skewable (bool): Whether the quadrant gate is skewable. - angles (list): List of the four angles of the quadrant demarcations - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - gids (list): Group IDs of each sector, assigned to ``model.gids``. - locked (bool): Prevents modification of the gate via the web interface. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - - Returns: - A QuadrantGate object. - - Example: - ```python - cellengine.Gate.create_quadrant_gate(experimentId, x_channel="FSC-A", - y_channel="FSC-W", name="my gate", x=160000, y=200000) - experiment.create_quadrant_gate(x_channel="FSC-A", - y_channel="FSC-W", name="my gate", x=160000, y=200000) - ``` - """ - # set labels based on axis scale - r = ce.APIClient().get_scaleset(experiment_id, as_dict=True) - scale_min = min(x["scale"]["minimum"] for x in r["scales"]) - scale_max = max(x["scale"]["minimum"] for x in r["scales"]) - - if labels == []: - labels = [ - [scale_max, scale_max], # upper right - [scale_min, scale_max], # upper left - [scale_min, scale_min], # lower left - [scale_max, scale_min], # lower right - ] # lower right - - elif len(labels) == 4 and all(len(label) == 2 for label in labels): - pass - else: - raise ValueError("Labels must be a list of four length-2 lists.") - - if gid is None: - gid = generate_id() - if gids is None: - gids = [ - generate_id(), - generate_id(), - generate_id(), - generate_id(), - ] - - names = [name + append for append in [" (UR)", " (UL)", " (LL)", " (LR)"]] - - model = { - "locked": locked, - "labels": labels, - "gids": gids, - "skewable": skewable, - "quadrant": {"x": x, "y": y, "angles": angles}, - } - - body = { - "experimentId": experiment_id, - "names": names, - "type": "QuadrantGate", - "gid": gid, - "xChannel": x_channel, - "yChannel": y_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/range_gate.py b/cellengine/payloads/gate_utils/range_gate.py deleted file mode 100644 index a4e929c7..00000000 --- a/cellengine/payloads/gate_utils/range_gate.py +++ /dev/null @@ -1,93 +0,0 @@ -import numpy -from typing import List - -from cellengine.utils.generate_id import generate_id -from cellengine.payloads.gate_utils import format_common_gate - - -def format_range_gate( - experiment_id: str, - x_channel: str, - name: str, - x1: float, - x2: float, - y: float = 0.5, - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """Formats a range gate for posting to the CellEngine API. - - Args: - x_channel (str): The name of the x channel to which the gate applies. - name (str): The name of the gate - x1 (float): The first x coordinate (after the channel's scale has been applied). - x2 (float): The second x coordinate (after the channel's scale has been - applied). - y (float): Position of the horizontal line between the vertical lines - label (float): Position of the label. Defaults to the midpoint of the gate. - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - locked (bool): Prevents modification of the gate via the web interface. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - - Returns: - A RangeGate object. - - Example: - ```python - experiment.create_range_gate(x_channel="FSC-A", name="my gate", - x1=12.502, x2=95.102) - cellengine.Gate.create_range_gate(experiment_id, - x_channel="FSC-A", name="my gate", - 12.502, 95.102) - ``` - """ - if label == []: - label = [numpy.mean([x1, x2]), y] - if gid is None: - gid = generate_id() - - model = {"locked": locked, "label": label, "range": {"x1": x1, "x2": x2, "y": y}} - - body = { - "experimentId": experiment_id, - "name": name, - "type": "RangeGate", - "gid": gid, - "xChannel": x_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/rectangle_gate.py b/cellengine/payloads/gate_utils/rectangle_gate.py deleted file mode 100644 index 58166d8e..00000000 --- a/cellengine/payloads/gate_utils/rectangle_gate.py +++ /dev/null @@ -1,103 +0,0 @@ -from typing import List -import numpy - -from cellengine.utils.generate_id import generate_id -from cellengine.payloads.gate_utils import format_common_gate - - -def format_rectangle_gate( - experiment_id: str, - x_channel: str, - y_channel: str, - name: str, - x1: float, - x2: float, - y1: float, - y2: float, - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """Formats a rectangle gate for posting to the CellEngine API. - - Args: - name (str): The name of the gate - x_channel (float): The name of the x channel to which the gate applies. - y_channel (float): The name of the y channel to which the gate applies. - x1 (float): The first x coordinate (after the channel's scale has been applied). - x2 (float): The second x coordinate (after the channel's scale - has been applied). - y1 (float): The first y coordinate (after the channel's scale has been applied). - y2 (float): The second y coordinate (after the channel's scale has - been applied). - label (float): Position of the label. Defaults to the midpoint of the gate. - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - locked (bool): Prevents modification of the gate via the web interface. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - - Returns: - A RectangleGate object. - - Example: - ```python - experiment.create_rectangle_gate(x_channel="FSC-A", y_channel="FSC-W", - name="my gate", 12.502, 95.102, 1020, 32021.2) - cellengine.Gate.create_rectangle_gate(experiment_id, x_channel="FSC-A", - y_channel="FSC-W", name="my gate", x1=12.502, x2=95.102, y1=1020, y2=32021.2, - gid=global_gate.gid) - ``` - """ - if label == []: - label = [numpy.mean([x1, x2]), numpy.mean([y1, y2])] - if gid is None: - gid = generate_id() - - model = { - "locked": locked, - "label": label, - "rectangle": {"x1": x1, "x2": x2, "y1": y1, "y2": y2}, - } - - body = { - "experimentId": experiment_id, - "name": name, - "type": "RectangleGate", - "gid": gid, - "xChannel": x_channel, - "yChannel": y_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/split_gate.py b/cellengine/payloads/gate_utils/split_gate.py deleted file mode 100644 index 94f254c1..00000000 --- a/cellengine/payloads/gate_utils/split_gate.py +++ /dev/null @@ -1,120 +0,0 @@ -from typing import List - -import cellengine as ce -from cellengine.utils.generate_id import generate_id -from cellengine.payloads.gate_utils import format_common_gate - - -def format_split_gate( - experiment_id: str, - x_channel: str, - name: str, - x: str, - y: float = 0.5, - labels: List[str] = [], - gid: str = None, - gids: List[str] = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, -): - """ - Formats a split gate for posting to the CellEngine API. - - Split gates have two sectors (right and left), - each with a unique gid and name. - - Args: - x_channel (str): The name of the x channel to which the gate applies. - name (str): The name of the gate. - x (float): The x coordinate of the center point (after the channel's scale has - been applied). - y (float): The relative position from 0 to 1 of the dashed line extending - from the center point. - labels (list): Positions of the quadrant labels. A list of two length-2 lists in - the order: L, R. These are set automatically to the top corners. - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - gids (list): Group IDs of each sector, assigned to model.gids. - locked (bool): Prevents modification of the gate via the web interface. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - - Returns: - A SplitGate object. - - Example: - ```python - cellengine.Gate.create_split_gate(experiment_id, x_channel="FSC-A", - name="my gate", x=144000, y=100000) - experiment.create_split_gate(x_channel="FSC-A", name="my gate", x=144000, - y=100000) - ``` - """ - # set labels based on axis scale - r = ce.APIClient().get_scaleset(experiment_id, as_dict=True) - scale_min = min(x["scale"]["minimum"] for x in r["scales"]) - scale_max = max(x["scale"]["minimum"] for x in r["scales"]) - - if labels == []: - labels = [ - [scale_min + 0.1 * scale_max, 0.916], - [scale_max - 0.1 * scale_max, 0.916], - ] - elif len(labels) == 2 and len(labels[0]) == 2 and len(labels[1]) == 2: - pass - else: - raise ValueError("Labels must be a list of two length-2 lists.") - - if gid is None: - gid = generate_id() - if gids is None: - gids = [generate_id(), generate_id()] - - names = [name + " (L)", name + " (R)"] - - model = { - "locked": locked, - "labels": labels, - "gids": gids, - "split": {"x": x, "y": y}, - } - - body = { - "experimentId": experiment_id, - "names": names, - "type": "SplitGate", - "gid": gid, - "xChannel": x_channel, - "parentPopulationId": parent_population_id, - "model": model, - } - - return format_common_gate( - experiment_id, - body=body, - tailored_per_file=tailored_per_file, - fcs_file_id=fcs_file_id, - fcs_file=fcs_file, - create_population=create_population, - ) diff --git a/cellengine/payloads/gate_utils/utils.py b/cellengine/payloads/gate_utils/utils.py deleted file mode 100644 index 9bca956d..00000000 --- a/cellengine/payloads/gate_utils/utils.py +++ /dev/null @@ -1,51 +0,0 @@ -import cellengine as ce - - -def format_common_gate( - experiment_id, body, tailored_per_file, fcs_file_id, fcs_file, create_population -): - """ - Args: - experiment_id (str): The ID of the experiment to which to add the gate. - Use when calling this as a static method; not needed when calling - from an Experiment object - name (str): The name of the gate - x_channel (str): The name of the x channel to which the gate applies. - gid (str): Group ID of the gate, used for tailoring. If this is not - specified, then a new Group ID will be created. To create a - tailored gate, the gid of the global tailored gate must be specified. - parent_population_id (str): ID of the parent population. Use ``None`` for - the "ungated" population. If specified, do not specify - ``parent_population``. - parent_population (str): Name of the parent population. An attempt will - be made to find the population by name. If zero or more than - one population exists with the name, an error will be thrown. - If specified, do not specify ``parent_population_id``. - tailored_per_file (bool): Whether or not this gate is tailored per FCS file. - fcs_file_id (str): ID of FCS file, if tailored per file. Use ``None`` for - the global gate in a tailored gate group. If specified, do not - specify ``fcs_file``. - fcs_file (str): Name of FCS file, if tailored per file. An attempt will be made - to find the file by name. If zero or more than one file exists with - the name, an error will be thrown. Looking up files by name is - slower than using the ID, as this requires additional requests - to the server. If specified, do not specify ``fcs_file_id``. - create_population (bool): Automatically create corresponding population. - """ - return parse_fcs_file_args( - experiment_id, body, tailored_per_file, fcs_file_id, fcs_file - ) - - -def parse_fcs_file_args(experiment_id, body, tailored_per_file, fcs_file_id, fcs_file): - """Find the fcs file ID if 'tailored_per_file' and either 'fcs_file' or - 'fcs_file_id' are specified. - """ - if fcs_file is not None and fcs_file_id is not None: - raise ValueError("Please specify only 'fcs_file' or 'fcs_file_id'.") - if fcs_file is not None and tailored_per_file is True: # lookup by name - _file = ce.APIClient().get_fcs_file(experiment_id=experiment_id, name=fcs_file) - fcs_file_id = _file._id - body["tailoredPerFile"] = tailored_per_file - body["fcsFileId"] = fcs_file_id - return body diff --git a/cellengine/resources/experiment.py b/cellengine/resources/experiment.py index c8ba3c8d..123eb539 100644 --- a/cellengine/resources/experiment.py +++ b/cellengine/resources/experiment.py @@ -1,24 +1,18 @@ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime -import inspect -from math import pi -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Union, Tuple, overload + +try: + from typing import Literal +except ImportError: + from typing_extensions import Literal -from custom_inherit import doc_inherit from dataclasses_json.cfg import config from marshmallow import fields from pandas.core.frame import DataFrame import cellengine as ce -from cellengine.payloads.gate_utils import ( - format_ellipse_gate, - format_polygon_gate, - format_quadrant_gate, - format_range_gate, - format_rectangle_gate, - format_split_gate, -) from cellengine.resources.attachment import Attachment from cellengine.resources.compensation import Compensation from cellengine.resources.fcs_file import FcsFile @@ -417,7 +411,6 @@ def create_gates(self, gates: List): """Save a collection of gate objects.""" return Gate.bulk_create(self._id, gates) - @doc_inherit(Gate.delete_gates) def delete_gate( self, _id: str = None, gid: str = None, exclude: str = None ) -> None: @@ -428,177 +421,162 @@ def delete_gate( """ return ce.APIClient().delete_gate(self._id, _id, gid, exclude) - @doc_inherit(format_rectangle_gate) + def delete_gates(self, ids: List[str]) -> None: + """Deletes multiple gates provided a list of _ids.""" + ce.APIClient().delete_gates(self._id, ids) + + @overload def create_rectangle_gate( - self, - x_channel: str, - y_channel: str, - name: str, - x1: float, - x2: float, - y1: float, - y2: float, - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: Literal[True], **kwargs + ) -> Tuple[RectangleGate, Population]: + ... + + @overload + def create_rectangle_gate( + self, *args, create_population: Literal[False], **kwargs ) -> RectangleGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_rectangle_gate( - kwargs.pop("self")._id, # type: ignore + ... + + def create_rectangle_gate( + self, *args, create_population: bool = True, **kwargs + ) -> Union[RectangleGate, Tuple[RectangleGate, Population]]: + """Create a RectangleGate. + + Accepts all args and kwargs available for + [`RectangleGate.create()`][cellengine.resources.gate.RectangleGate.create]. + """ + return RectangleGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore - @doc_inherit(format_polygon_gate) + @overload def create_polygon_gate( - self, - x_channel: str, - y_channel: str, - name: str, - vertices: List[float], - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: Literal[True], **kwargs + ) -> Tuple[PolygonGate, Population]: + ... + + @overload + def create_polygon_gate( + self, *args, create_population: Literal[False], **kwargs ) -> PolygonGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_polygon_gate( - kwargs.pop("self")._id, # type: ignore + ... + + def create_polygon_gate( + self, *args, create_population: bool = True, **kwargs + ) -> Union[PolygonGate, Tuple[PolygonGate, Population]]: + """Create a PolygonGate. + + Accepts all args and kwargs available for + [`PolygonGate.create()`][cellengine.resources.gate.PolygonGate.create]. + """ + return PolygonGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore - @doc_inherit(format_ellipse_gate) + @overload def create_ellipse_gate( - self, - x_channel: str, - y_channel: str, - name: str, - x: float, - y: float, - angle: float, - major: float, - minor: float, - label: List = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: Literal[True], **kwargs + ) -> Tuple[EllipseGate, Population]: + ... + + @overload + def create_ellipse_gate( + self, *args, create_population: Literal[False], **kwargs ) -> EllipseGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_ellipse_gate( - kwargs.pop("self")._id, # type: ignore + ... + + def create_ellipse_gate( + self, *args, create_population: bool = True, **kwargs + ) -> Union[EllipseGate, Tuple[EllipseGate, Population]]: + """Create an EllipseGate. + + Accepts all args and kwargs available for + [`EllipseGate.create()`][cellengine.resources.gate.EllipseGate.create]. + """ + return EllipseGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore - @doc_inherit(format_range_gate) + @overload def create_range_gate( - self, - x_channel: str, - name: str, - x1: float, - x2: float, - y: float = 0.5, - label: List[str] = [], - gid: str = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: Literal[True], **kwargs + ) -> Tuple[RangeGate, Population]: + ... + + @overload + def create_range_gate( + self, *args, create_population: Literal[False], **kwargs ) -> RangeGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_range_gate( - kwargs.pop("self")._id, # type: ignore + ... + + def create_range_gate( + self, *args, create_population: bool = True, **kwargs + ) -> Union[RangeGate, Tuple[RangeGate, Population]]: + """Create a RangeGate. + + Accepts all args and kwargs available for + [`RangeGate.create()`][cellengine.resources.gate.RangeGate.create]. + """ + return RangeGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore - @doc_inherit(format_split_gate) def create_split_gate( - self, - x_channel: str, - name: str, - x: str, - y: float = 0.5, - labels: List[str] = [], - gid: str = None, - gids: List[str] = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: bool = True, **kwargs ) -> SplitGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_split_gate( - kwargs.pop("self")._id, # type: ignore + """Create a SplitGate. + + Accepts all args and kwargs available for + [`SplitGate.create()`][cellengine.resources.gate.SplitGate.create]. + """ + return SplitGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore - @doc_inherit(format_quadrant_gate) + @overload def create_quadrant_gate( - self, - x_channel: str, - y_channel: str, - name: str, - x: float, - y: float, - labels: List[str] = [], - skewable: bool = False, - angles: List[float] = [0, pi / 2, pi, 3 * pi / 2], - gid: str = None, - gids: List[str] = None, - locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, - tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + self, *args, create_population: Literal[True], **kwargs + ) -> Tuple[QuadrantGate, List[Population]]: + ... + + @overload + def create_quadrant_gate( + self, *args, create_population: Literal[False], **kwargs ) -> QuadrantGate: - f = inspect.currentframe() - args, _, _, values = inspect.getargvalues(f) # type: ignore - kwargs = {arg: values.get(arg, None) for arg in args} - post_body = format_quadrant_gate( - kwargs.pop("self")._id, # type: ignore + ... + + def create_quadrant_gate( + self, *args, create_population: bool = True, **kwargs + ) -> Union[QuadrantGate, Tuple[QuadrantGate, List[Population]]]: + """Create a QuadrantGate. + + Accepts all args and kwargs available for + [`QuadrantGate.create()`][cellengine.resources.gate.QuadrantGate.create]. + """ + return QuadrantGate.create( + self._id, + *args, + create_population=create_population, # type: ignore pyright limitation **kwargs, ) - return ce.APIClient().post_gate(self._id, post_body) # type: ignore def create_population(self, population: Dict) -> Population: - """Create a complex population + """Create a population. Args: population (dict): The population to create. diff --git a/cellengine/resources/fcs_file.py b/cellengine/resources/fcs_file.py index f059cef6..c7107378 100644 --- a/cellengine/resources/fcs_file.py +++ b/cellengine/resources/fcs_file.py @@ -230,7 +230,8 @@ def plot( ) -> Plot: """Build a plot for an FcsFile. - See [`Plot.get`][cellengine.resources.plot.Plot.get] for more information. + See [`APIClient.get_plot()`][cellengine.APIClient.get_plot] + for more information. """ plot = Plot.get( experiment_id=self.experiment_id, diff --git a/cellengine/resources/gate.py b/cellengine/resources/gate.py index c951f53f..4b9fc4cb 100644 --- a/cellengine/resources/gate.py +++ b/cellengine/resources/gate.py @@ -1,124 +1,122 @@ from __future__ import annotations -import attr import importlib -from typing import Dict, List, Optional from math import pi -from custom_inherit import doc_inherit +from operator import itemgetter +from typing import Any, Dict, List, Optional, Union, Tuple, overload + +try: + from typing import Literal +except ImportError: + from typing_extensions import Literal + +from attr import define, field +from numpy import array, mean, stack import cellengine as ce -from cellengine.utils.helpers import get_args_as_kwargs -from cellengine.payloads.gate import _Gate -from cellengine.payloads.gate_utils import ( - format_rectangle_gate, - format_split_gate, - format_polygon_gate, - format_ellipse_gate, - format_quadrant_gate, - format_range_gate, +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 +from cellengine.utils.helpers import ( + get_args_as_kwargs, + normalize, + remove_keys_with_none_values, ) +import sys -@attr.s(repr=False) -class Gate(_Gate): - def __init__(cls, *args, **kwargs): - if cls is Gate: - raise TypeError( - "The Gate base class may not be directly \ - instantiated. Use the .create() classmethod." - ) - return object.__new__(cls, *args, **kwargs) +if sys.version_info[:2] >= (3, 8): + from collections.abc import Mapping +else: + from collections import Mapping # type: ignore - @classmethod - def get( - cls, experiment_id: str, _id: Optional[str] = None, name: Optional[str] = None - ) -> Gate: - """Get a specific gate.""" - kwargs = {"name": name} if name else {"_id": _id} - gate = ce.APIClient().get_gate(experiment_id, **kwargs) - return gate - def delete(self): - ce.APIClient().delete_gate(self.experiment_id, self._id) +def exception_handler(func): + def inner_function(*args, **kwargs): + try: + return func(*args, **kwargs) + except Exception as err: + if type(err) is ValueError: + raise err + raise RuntimeError(f"Incorrect arguments passed to {func.__name__}.") - def update(self): - """Save changes to this Gate to CellEngine.""" - props = ce.APIClient().update_entity( - self.experiment_id, self._id, "gates", body=self._properties - ) - self._properties.update(props) + return inner_function - def post(self): - res = ce.APIClient().post_gate( - self.experiment_id, self._properties, as_dict=True - ) - props, _ = self._separate_gate_and_population(res) - self._properties.update(props) - @classmethod - def bulk_create(cls, experiment_id, gates: List) -> List[Gate]: - if type(gates[0]) is dict: - pass - elif str(gates[0].__module__) == "cellengine.resources.gate": - gates = [gate._properties for gate in gates] - return ce.APIClient().post_gate(experiment_id, gates, create_population=False) +def deep_update(source, overrides): + """ + Update a nested dictionary or similar mapping. + Modify `source` in place. + """ + for key, value in overrides.items(): + if isinstance(value, Mapping) and value: + returned = deep_update(source.get(key, {}), value) + source[key] = returned + else: + source[key] = overrides[key] + return source - @classmethod - def factory(cls, gates: Dict) -> List["Gate"]: - if type(gates) is list: - return [cls._build_gate(gate) for gate in gates] + +@define(repr=False) +class Gate: + _id: str = field(on_setattr=readonly) + experiment_id: str = field(on_setattr=readonly) + gid: str = field(on_setattr=readonly) + type: str + x_channel: str + model: Dict + tailored_per_file: bool = False + fcs_file_id: Optional[str] = None + names: Optional[List[str]] = field(default=None) + name: Optional[str] = field(default=None) + y_channel: Optional[str] = field(default=None) + + def __repr__(self): + if self.name: + label, name = ("name", self.name) else: - return cls._build_gate(gates) + label, name = ("names", str(self.names)) + return f"{self.type}(_id='{self._id}', {label}='{name}')" - @classmethod - def _build_gate(cls, gate): - """Get the gate type and return instance of the correct subclass.""" - module = importlib.import_module(__name__) - gate, _ = cls._separate_gate_and_population(gate) - gate_type = getattr(module, gate["type"]) - return gate_type(properties=gate) + @staticmethod + def _format_gate(gate, **kwargs): + module = importlib.import_module("cellengine") + _class = getattr(module, gate["type"]) + return _class._format(**gate) @classmethod - def _separate_gate_and_population(cls, gate): - try: - if "gate" in gate.keys(): - return gate["gate"], [k for k in gate.keys() if k != "gate"] - else: - return gate, None - except KeyError: - raise ValueError("Gate payload format is invalid") + def create_many( + cls, + gates: List[Dict], + ): + experiment_id = set([g["experiment_id"] for g in gates]) + if len(experiment_id) != 1: + raise RuntimeError("Created gates must all be in the same Experiment.") - @staticmethod - def delete_gates( - experiment_id, _id: str = None, gid: str = None, exclude: str = None - ) -> None: - """Deletes a gate or a tailored gate family. + formatted_gates = [cls._format_gate(gate) for gate in gates] + return ce.APIClient().create( + [Gate(id=None, **g) for g in formatted_gates], # type: ignore + create_population=False, + ) - Specify the top-level gid when working with compound gates (specifying - the gid of a sector (i.e. one listed in ``model.gids``) will result in - no gates being deleted). If ``_id`` is specified, only that gate will be - deleted, regardless of the other parameters specified. May be called as - a static method on `cellengine.Gate` or on an `Experiment` instance. + @property + def path(self): + return f"experiments/{self.experiment_id}/gates/{self._id}".rstrip("/None") - Args: - experiment_id (str): ID of experiment. - _id (str): ID of gate. - gid (str): ID of gate family. - exclude (str): Gate ID to exclude from deletion. + @classmethod + def from_dict(cls, data: dict): + return converter.structure(data, cls) - Example: - ```python - cellengine.Gate.delete_gates(experiment_id, gid = ) - # or - experiment.delete_gates(_id = ) - # or - experiment.delete_gates(gid = , exclude = ) - ``` + def to_dict(self): + return converter.unstructure(self) - Returns: - None + def update(self): + """Save changes to this Gate to CellEngine.""" + res = ce.APIClient().update(self) + self.__setstate__(res.__getstate__()) # type: ignore - """ - ce.APIClient().delete_gate(experiment_id, _id, gid, exclude) + def delete(self): + ce.APIClient().delete_gate(self.experiment_id, self._id) @staticmethod def update_gate_family(experiment_id: str, gid: str, body: Dict) -> None: @@ -130,7 +128,7 @@ def update_gate_family(experiment_id: str, gid: str, body: Dict) -> None: Args: experiment_id: ID of experiment gid: ID of gate family to modify - body (dict): camelCase properties to update + body: camelCase properties to update Returns: Raises a warning if no gates are modified, else None @@ -140,18 +138,38 @@ 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_id): - """Tailor this gate to a specific fcs_file.""" - self._properties.update( - ce.APIClient().tailor_to(self.experiment_id, self._id, fcs_file_id) - ) + def tailor_to(self, fcs_file: FcsFile): + self.tailored_per_file = True + self.fcs_file_id = fcs_file._id + self.update() class RectangleGate(Gate): - """Basic concrete class for Rectangle gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + x1: float, + x2: float, + y1: float, + y2: float, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, + ) -> Tuple[RectangleGate, Population]: + ... + @overload @classmethod - @doc_inherit(format_rectangle_gate) def create( cls, experiment_id: str, @@ -162,81 +180,493 @@ def create( x2: float, y1: float, y2: float, - label: List[str] = [], - gid: str = None, + label: List[float] = [], + gid: Optional[str] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, ) -> RectangleGate: - g = format_rectangle_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + ... + + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + x1: float, + x2: float, + y1: float, + y2: float, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: bool = True, + parent_population_id: Optional[str] = None, + ) -> Union[RectangleGate, Tuple[RectangleGate, Population]]: + """Creates a rectangle gate. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + y_channel: The name of the y channel to which the gate applies. + name: The name of the gate + x1: The first x coordinate (after the channel's scale has been + applied). + x2: The second x coordinate (after the channel's scale has been + applied). + y1: The first y coordinate (after the channel's scale has been + applied). + y2: The second y coordinate (after the channel's scale has been + applied). + label: Position of the label. Defaults to the midpoint of the gate. + gid: Group ID of the gate, used for tailoring. If this is not + specified, then a new Group ID will be created. Must be + specified when creating a tailored gate. + locked: Prevents modification of the gate via the web interface. + tailored_per_file: Whether or not this gate is tailored per FCS + file. + fcs_file_id: ID of FCS file, if tailored per file. Use `None` for + the global gate in a tailored gate group. If specified, do not + specify `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. An attempt will be + made to find the file by name. If zero or more than one file + exists with the name, an error will be thrown. Looking up files + by name is slower than using the ID, as this requires additional + requests to the server. If specified, do not specify + `fcs_file_id`. + create_population: If true, a corresponding population will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create this population. + + Returns: + A RectangleGate if `create_population` is False, or a Tuple with the + gate and populations if `create_population` is True. + + Examples: + ```python + gate, pop = experiment.create_rectangle_gate( + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + x1=12.502, + x2=95.102, + y1=1020, + y2=32021.2) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs: Dict[str, Any]): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + x1 = args.get("x1") or args.get("model", {}).get("rectangle", {}).get("x1") + x2 = args.get("x2") or args.get("model", {}).get("rectangle", {}).get("x2") + y1 = args.get("y1") or args.get("model", {}).get("rectangle", {}).get("y1") + y2 = args.get("y2") or args.get("model", {}).get("rectangle", {}).get("y2") + if not x1 or not x2 or not y1 or not y2: + raise ValueError("x1, x2, y1 and y2 must be provided.") + label = args.get("label") or args.get("model", {}).get( + "label", + [ + mean([x1, x2]), + mean([y1, y2]), + ], + ) + + model = { + "locked": args.get("locked") or args.get("model", {}).get("locked", False), + "label": label, + "rectangle": {"x1": x1, "x2": x2, "y1": y1, "y2": y2}, + } + + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", generate_id()), + "model": model, + "name": args.get("name"), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "RectangleGate", + "x_channel": args.get("x_channel"), + "y_channel": args.get("y_channel"), + } class PolygonGate(Gate): - """Basic concrete class for polygon gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + vertices: List[List[float]], + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, # TODO Ungated + ) -> Tuple[PolygonGate, Population]: + ... + @overload @classmethod - @doc_inherit(format_polygon_gate) def create( cls, experiment_id: str, x_channel: str, y_channel: str, name: str, - vertices: List[float], - label: List = [], - gid: str = None, + vertices: List[List[float]], + label: List[float] = [], + gid: Optional[str] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, ) -> PolygonGate: - g = format_polygon_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + ... + + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + vertices: List[List[float]], + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: bool = True, + parent_population_id: Optional[str] = None, + ) -> Union[PolygonGate, Tuple[PolygonGate, Population]]: + """Creates a polygon gate. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + y_channel: The name of the y channel to which the gate applies. + name: The name of the gate + vertices: List of coordinates, like `[[x,y], [x,y], ...]`. + label: Position of the label. Defaults to the midpoint of the gate. + gid: Group ID of the gate, used for tailoring. If this is not + specified, then a new Group ID will be created. To create a + tailored gate, the gid of the global tailored gate must be + specified. + locked: Prevents modification of the gate via the web interface. + tailored_per_file: Whether or not this gate is tailored per FCS + file. + fcs_file_id: ID of FCS file, if tailored per file. Use `None` for + the global gate in a tailored gate group. If specified, do not + specify `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. An attempt will be + made to find the file by name. If zero or more than one file + exists with the name, an error will be thrown. Looking up files + by name is slower than using the ID, as this requires additional + requests to the server. If specified, do not specify + `fcs_file_id`. + create_population: If true, a corresponding population will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create this populations. + + Returns: + A PolygonGate if `create_population` is False, or a Tuple with the + gate and populations if `create_population` is True. + + Examples: + ```python + gate, pop = experiment.create_polygon_gate( + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + vertices=[[1,4], [2,5], [3,6]]) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + vertices = args.get("vertices", []) or args.get("model", {}).get( + "polygon", {} + ).get("vertices", []) + label = args.get("label") or args.get("model", {}).get( + "label", mean(vertices, axis=0).tolist() + ) + + model = { + "locked": args.get("model", {}).get("locked", False), + "label": label, + "polygon": {"vertices": vertices}, + } + + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", generate_id()), + "model": model, + "name": args.get("name"), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "PolygonGate", + "x_channel": args.get("x_channel"), + "y_channel": args.get("y_channel"), + } class EllipseGate(Gate): - """Basic concrete class for ellipse gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + center: List[float], + angle: float, + major: float, + minor: float, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, # TODO Ungated + ) -> Tuple[EllipseGate, Population]: + ... + @overload @classmethod - @doc_inherit(format_ellipse_gate) def create( cls, experiment_id: str, x_channel: str, y_channel: str, name: str, - x: float, - y: float, + center: List[float], angle: float, major: float, minor: float, - label: List[str] = [], - gid: str = None, + label: List[float] = [], + gid: Optional[str] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, ) -> EllipseGate: - g = format_ellipse_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + ... + + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + center: List[float], + angle: float, + major: float, + minor: float, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: bool = True, + parent_population_id: Optional[str] = None, + ) -> Union[EllipseGate, Tuple[EllipseGate, Population]]: + """Creates an ellipse gate. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + y_channel: The name of the y channel to which the gate applies. + name: The name of the gate + center: The x, y centerpoint of the gate. + angle: The angle of the ellipse in radians. + major: The major radius of the ellipse. + minor: The minor radius of the ellipse. + label: Position of the label. Defaults to the midpoint of the gate. + gid: Group ID of the gate, used for tailoring. If this is not + specified, then a new Group ID will be created. To create a + tailored gate, the gid of the global tailored gate must be + specified. + locked: Prevents modification of the gate via the + web interface. + tailored_per_file: Whether or not this gate is tailored per FCS + file. + fcs_file_id: ID of FCS file, if tailored per file. Use `None` for + the global gate in a tailored gate group. If specified, do not + specify `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. An attempt will be + made to find the file by name. If zero or more than one file + exists with the name, an error will be thrown. Looking up files + by name is slower than using the ID, as this requires additional + requests to the server. If specified, do not specify + `fcs_file_id`. + create_population: If true, a corresponding population will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create this populations. + + Returns: + If `create_population` is `True`, a tuple containing an EllipseGate + and a list of two Populations; otherwise, an EllipseGate. + + Examples: + ```python + gate, pop = experiment.create_ellipse_gate( + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + x=260000, + y=64000, + angle=0, + major=120000, + minor=70000) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + angle = args.get("angle") or args.get("model", {}).get("ellipse", {}).get( + "angle" + ) + major = args.get("major") or args.get("model", {}).get("ellipse", {}).get( + "major" + ) + minor = args.get("minor") or args.get("model", {}).get("ellipse", {}).get( + "minor" + ) + x = args.get("x") + y = args.get("y") + if x and y: + center = [x, y] + else: + center = args.get("center") or args.get("model", {}).get("ellipse", {}).get( + "center" + ) + x, y = center # type: ignore + + label = args.get("label") or args.get("model", {}).get("label", [x, y]) + + model = { + "locked": args.get("model", {}).get("locked", False), + "label": label, + "ellipse": { + "angle": angle, + "major": major, + "minor": minor, + "center": center, + }, + } + + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", generate_id()), + "model": model, + "name": args.get("name"), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "EllipseGate", + "x_channel": args.get("x_channel"), + "y_channel": args.get("y_channel"), + } class RangeGate(Gate): - """Basic concrete class for range gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + name: str, + x1: float, + x2: float, + y: float = 0.5, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, + ) -> Tuple[RangeGate, Population]: + ... + @overload @classmethod - @doc_inherit(format_range_gate) def create( cls, experiment_id: str, @@ -245,25 +675,159 @@ def create( x1: float, x2: float, y: float = 0.5, - label: List[str] = [], - gid: str = None, + label: List[float] = [], + gid: Optional[str] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, ) -> RangeGate: - g = format_range_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + ... + + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + name: str, + x1: float, + x2: float, + y: float = 0.5, + label: List[float] = [], + gid: Optional[str] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: bool = True, + parent_population_id: Optional[str] = None, + ) -> Union[RangeGate, Tuple[RangeGate, Population]]: + """Creates a range gate. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + name: The name of the gate + x1: The first x coordinate (after the channel's scale has + been applied). + x2: The second x coordinate (after the channel's scale has been + applied). + y: Position of the horizontal line between the + vertical lines + label: Position of the label. Defaults to + the midpoint of the gate. + gid: Group ID of the gate, used for tailoring. If + this is not specified, then a new Group ID will be created. To + create a tailored gate, the gid of the global tailored gate + must be specified. + locked: Prevents modification of the gate via the + web interface. + tailored_per_file: Whether or not this gate is + tailored per FCS file. + fcs_file_id: ID of FCS + file, if tailored per file. Use `None` for the global gate in + a tailored gate group. If specified, do not specify + `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. + An attempt will be made to find the file by name. If zero or + more than one file exists with the name, an error will be + thrown. Looking up files by name is slower than using the ID, + as this requires additional requests to the server. If + specified, do not specify `fcs_file_id`. + create_population: If true, a corresponding population will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create this populations. + + Returns: + If `create_population` is `True`, a tuple containing a RangeGate + and a list of two Populations; otherwise, a RangeGate. + + Examples: + ```python + gate, pop = experiment.create_range_gate( + x_channel="FSC-A", + name="my gate", + x1=12.502, + x2=95.102) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + x1 = args.get("x1") or args.get("model", {}).get("range", {}).get("x1") + x2 = args.get("x2") or args.get("model", {}).get("range", {}).get("x2") + y = args.get("y") or args.get("model", {}).get("range", {}).get("y", 0.5) + if not x1 or not x2: + raise ValueError("x1 and x2 must be provided.") + + label = args.get("label") or args.get("model", {}).get( + "label", [mean([x1, x2]), y] + ) + + model = { + "locked": args.get("model", {}).get("locked", False), + "label": label, + "range": {"x1": x1, "x2": x2, "y": y}, + } + + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", generate_id()), + "model": model, + "name": args.get("name"), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "RangeGate", + "x_channel": args.get("x_channel"), + } class QuadrantGate(Gate): - """Basic concrete class for quadrant gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + x: float, + y: float, + labels: List[List[float]] = [], + skewable: bool = False, + angles: List[float] = [0, pi / 2, pi, 3 * pi / 2], + gid: Optional[str] = None, + gids: Optional[List[str]] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, # TODO Ungated + ) -> Tuple[QuadrantGate, List[Population]]: + ... + @overload @classmethod - @doc_inherit(format_quadrant_gate) def create( cls, experiment_id: str, @@ -272,28 +836,234 @@ def create( name: str, x: float, y: float, - labels: List[str] = [], + labels: List[List[float]] = [], skewable: bool = False, angles: List[float] = [0, pi / 2, pi, 3 * pi / 2], - gid: str = None, - gids: List[str] = None, + gid: Optional[str] = None, + gids: Optional[List[str]] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, - create_population: bool = True, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, ) -> QuadrantGate: - g = format_quadrant_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + ... + + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + y_channel: str, + name: str, + x: float, + y: float, + labels: List[List[float]] = [], + skewable: bool = False, + angles: List[float] = [0, pi / 2, pi, 3 * pi / 2], + gid: Optional[str] = None, + gids: Optional[List[str]] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: bool = True, + parent_population_id: Optional[str] = None, + ) -> Union[QuadrantGate, Tuple[QuadrantGate, List[Population]]]: + """Creates a quadrant gate. + + Quadrant gates have four sectors (upper-right, upper-left, lower-left, + lower-right), each with a unique gid and name. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + y_channel: The name of the y channel to which the gate applies. + name: The name of the gate + x: The x coordinate of the center point (after the channel's scale + has been applied). + y: The y coordinate (after the channel's scale has been applied). + labels: Positions of the quadrant labels. A list of four length-2 + vectors in the order UR, UL, LL, LR. These are set automatically + to the plot corners. + skewable: Whether the quadrant gate is skewable. + angles: List of the four angles of the quadrant demarcations + gid: Group ID of the gate, used for tailoring. If this is not + specified, then a new Group ID will be created. To create a + tailored gate, the gid of the global tailored gate must be + specified. + gids: Group IDs of each sector, assigned to `model.gids`. + locked: Prevents modification of the gate via the web + interface. + tailored_per_file: Whether or not this gate is + tailored per FCS file. + fcs_file_id: ID of FCS file, if tailored per file. + Use `None` for the global gate in a tailored gate group. If + specified, do not specify `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. + An attempt will be made to find the file by name. If zero or + more than one file exists with the name, an error will be + thrown. Looking up files by name is slower than using the ID, + as this requires additional requests to the server. If + specified, do not specify `fcs_file_id`. + create_population: If true, corresponding populations will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create these populations. + + Returns: + If `create_population` is `True`, a tuple containing the + QuadrantGate and a list of two Populations; otherwise, a + QuadrantGate. + + Examples: + ```python + gate, pops = experiment.create_quadrant_gate( + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + x=160000, + y=200000) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + x = args.get("x") or args.get("model", {}).get("quadrant", {}).get("x") + y = args.get("y") or args.get("model", {}).get("quadrant", {}).get("y") + angles = ( + args.get("model", {}) + .get("quadrant", {}) + .get("angles", [0, pi / 2, pi, 3 * pi / 2]) + ) + labels = args.get("labels") or args.get("model", {}).get("labels", []) + + if labels == []: + labels = cls._nudge_labels( + labels, + args["experiment_id"], + args["x_channel"], + args["y_channel"], + ) + if not (len(labels) == 4 and all(len(label) == 2 for label in labels)): + raise ValueError("Labels must be a list of four length-2 lists.") + + model = { + "locked": args.get("gids") or args.get("model", {}).get("locked", False), + "labels": labels, + "gids": args.get("gids") + or args.get("model", {}).get( + "gids", [generate_id(), generate_id(), generate_id(), generate_id()] + ), + "skewable": args.get("model", {}).get("skewable", False), + "quadrant": { + "x": x, + "y": y, + "angles": angles, + }, + } + + default_gid = generate_id() + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", default_gid), + "model": model, + "names": args.get( + "names", + [ + args.get("name", default_gid) + suffix + for suffix in [" (UR)", " (UL)", " (LL)", " (LR)"] + ], + ), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "QuadrantGate", + "x_channel": args.get("x_channel"), + "y_channel": args.get("y_channel"), + } + + @classmethod + def _nudge_labels( + cls, labels: List, experiment_id: str, x_channel: str, y_channel: str + ) -> List: + # set labels based on axis scale + scaleset = ce.APIClient().get_scaleset(experiment_id) + xmin, xmax = itemgetter("minimum", "maximum")(scaleset.scales[x_channel]) + ymin, ymax = itemgetter("minimum", "maximum")(scaleset.scales[y_channel]) + + # nudge labels in from plot corners by pixels + nudged_labels = array([[290, 290], [0, 290], [0, 0], [290, 0]]) + array( + [[-32, -16], [40, -16], [40, 15], [-32, 15]] + ) + + # scale the nudged px labels to the actual x and y ranges + x_scale = normalize(nudged_labels[:, 0], 0, 290, xmin, xmax) + y_scale = normalize(nudged_labels[:, 1], 0, 290, ymin, ymax) + + labels = stack((x_scale, y_scale)).T.astype(int).tolist() + return labels class SplitGate(Gate): - """Basic concrete class for split gates""" + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + name: str, + x: float, + y: float, + labels: List[List[float]] = [], + gid: Optional[str] = None, + gids: Optional[List[str]] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[True] = True, + parent_population_id: Optional[str] = None, # TODO Ungated + ) -> Tuple[SplitGate, List[Population]]: + ... + + @overload + @classmethod + def create( + cls, + experiment_id: str, + x_channel: str, + name: str, + x: float, + y: float, + labels: List[List[float]] = [], + gid: Optional[str] = None, + gids: Optional[List[str]] = None, + locked: bool = False, + tailored_per_file: bool = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, + create_population: Literal[False] = False, + ) -> SplitGate: + ... @classmethod - @doc_inherit(format_split_gate) def create( cls, experiment_id: str, @@ -301,16 +1071,147 @@ def create( name: str, x: float, y: float = 0.5, - labels: List[str] = [], - gid: str = None, - gids: List[str] = None, + labels: List[List[float]] = [], + gid: Optional[str] = None, + gids: Optional[List[str]] = None, locked: bool = False, - parent_population_id: str = None, - parent_population: str = None, tailored_per_file: bool = False, - fcs_file_id: str = None, - fcs_file: str = None, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, create_population: bool = True, - ) -> SplitGate: - g = format_split_gate(**get_args_as_kwargs(cls, locals())) - return cls(g) + parent_population_id: Optional[str] = None, + ) -> Union[SplitGate, Tuple[SplitGate, List[Population]]]: + """ + Creates a split gate. + + Split gates have two sectors (right and left), each with a unique gid + and name. + + Args: + experiment_id: The ID of the experiment. + x_channel: The name of the x channel to which the gate applies. + name: The name of the gate. + x: The x coordinate of the center point (after the + channel's scale has been applied). + y: The relative position from 0 to 1 of the horizontal dashed line + extending from the center point. + labels: Positions of the quadrant labels. A list of two length-2 + lists in the order: L, R. These are set automatically to the top + corners. + gid: Group ID of the gate, used for tailoring. If this is not + specified, then a new Group ID will be created. To create a + tailored gate, the gid of the global tailored gate must be + specified. + gids: Group IDs of each sector, assigned to `model.gids`. + locked: Prevents modification of the gate via the web interface. + tailored_per_file: Whether or not this gate is tailored per FCS + file. + fcs_file_id: ID of FCS file, if tailored per file. Use `None` for + the global gate in a tailored gate group. If specified, do not + specify `fcs_file`. + fcs_file: Name of FCS file, if tailored per file. An attempt will + be made to find the file by name. If zero or more than one file + exists with the name, an error will be thrown. Looking up files + by name is slower than using the ID, as this requires additional + requests to the server. If specified, do not specify + `fcs_file_id`. + create_population: If true, corresponding populations will be + created and returned in a tuple with the gate. + parent_population_id: Use with `create_population` to specify the + population below which to create these populations. + + Returns: + A SplitGate if `create_population` is False, or a Tuple with the + gate and populations if `create_population` is True. + + Examples: + ```python + # With automatic creation of the corresponding populations: + gate, pops = experiment.create_split_gate( + experiment_id, + x_channel="FSC-A", + name="my gate", + x=144000, y=0.5, + parent_population_id="...") + # Without + gate = experiment.create_split_gate( + experiment_id, + x_channel="FSC-A", + name="my gate", + x=144000, y=0.5, + create_population=False) + ``` + """ + kwargs = get_args_as_kwargs(cls, locals()) + params = { + k: kwargs.pop(k) for k in ["create_population", "parent_population_id"] + } + + gate = cls._format(**kwargs) + return ce.APIClient().create(Gate(id=None, **gate), **params) # type: ignore + + @classmethod + @exception_handler + def _format(cls, **kwargs): + """Get relevant kwargs and shape into a gate model""" + + args = remove_keys_with_none_values(kwargs) + + x = args.get("x") or args.get("model", {}).get("split", {}).get("x") + y = args.get("y") or args.get("model", {}).get("split", {}).get("y", 0.5) + + labels = args.get("labels") or args.get("model", {}).get("labels", []) + + labels = args.get("labels") or args.get("model", {}).get("labels", []) + + if labels == []: + labels = cls._generate_labels( + labels, args["experiment_id"], args["x_channel"] + ) + if not len(labels) == 2 and len(labels[0]) == 2 and len(labels[1]) == 2: + raise ValueError("Labels must be a list of two length-2 lists.") + + model = { + "locked": args.get("locked") or args.get("model", {}).get("locked", False), + "labels": labels, + "gids": args.get("gids") + or args.get("model", {}).get("gids", [generate_id(), generate_id()]), + "split": { + "x": x, + "y": y, + }, + } + + default_gid = generate_id() + return { + "experiment_id": args["experiment_id"], + "fcs_file_id": parse_fcs_file_args( + args.get("experiment_id"), + args.get("tailored_per_file", False), + args.get("fcs_file_id"), + args.get("fcs_file"), + ), + "gid": args.get("gid", default_gid), + "model": model, + "names": args.get( + "names", + [args.get("name", default_gid) + suffix for suffix in [" (L)", " (R)"]], + ), + "tailored_per_file": args.get("tailored_per_file", False), + "type": "SplitGate", + "x_channel": args.get("x_channel"), + } + + @classmethod + def _generate_labels(cls, labels: List, experiment_id: str, x_channel: str): + # set labels based on axis scale + scaleset = ce.APIClient().get_scaleset(experiment_id) + scale_min, scale_max = itemgetter("minimum", "maximum")( + scaleset.scales[x_channel] + ) + + labels = [ + [scale_min + 0.1 * scale_max, 0.916], + [scale_max - 0.1 * scale_max, 0.916], + ] + return labels diff --git a/cellengine/resources/scaleset.py b/cellengine/resources/scaleset.py index 72789974..f719c1e9 100644 --- a/cellengine/resources/scaleset.py +++ b/cellengine/resources/scaleset.py @@ -1,7 +1,11 @@ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, Union, overload -from typing_extensions import Literal + +try: + from typing import Literal +except ImportError: + from typing_extensions import Literal from dataclasses_json.cfg import config from pandas import DataFrame diff --git a/cellengine/utils/__init__.py b/cellengine/utils/__init__.py index ee08d206..31ece59d 100644 --- a/cellengine/utils/__init__.py +++ b/cellengine/utils/__init__.py @@ -2,3 +2,6 @@ from .readonly import readonly from .converter import converter +from .generate_id import generate_id +from .helpers import is_valid_id +from .parse_fcs_file_args import parse_fcs_file_args diff --git a/cellengine/utils/api_client/APIClient.py b/cellengine/utils/api_client/APIClient.py index 80666c08..ed1d9b27 100644 --- a/cellengine/utils/api_client/APIClient.py +++ b/cellengine/utils/api_client/APIClient.py @@ -1,6 +1,7 @@ from __future__ import annotations from functools import lru_cache from getpass import getpass +import importlib import json import os from typing import Any, Dict, List, Optional, Tuple, TypeVar, Union, overload @@ -18,7 +19,15 @@ from ...resources.compensation import Compensation from ...resources.experiment import Experiment from ...resources.fcs_file import FcsFile -from ...resources.gate import Gate +from ...resources.gate import ( + EllipseGate, + Gate, + PolygonGate, + QuadrantGate, + RangeGate, + RectangleGate, + SplitGate, +) from ...resources.plot import Plot from ...resources.population import Population from ...resources.scaleset import ScaleSet @@ -36,6 +45,17 @@ ScaleSet, ) +_Gate = TypeVar( + "_Gate", + Gate, + RectangleGate, + EllipseGate, + PolygonGate, + RangeGate, + QuadrantGate, + SplitGate, +) + def create_many(client: APIClient, entities: List[Gate], **kwargs) -> List[Gate]: body = [client.unstructure_and_clean(e) for e in entities] @@ -47,7 +67,7 @@ def create_many(client: APIClient, entities: List[Gate], **kwargs) -> List[Gate] List[_class.pop()], # type: ignore set(paths).pop(), list(payload), - kwargs=kwargs, + **kwargs, ) @@ -140,7 +160,7 @@ def _handle_response(self, response): response = [response] return response[0] - def _handle_list(self, response: List) -> RuntimeError: + def _handle_list(self, response: List) -> None: if len(response) == 0: raise RuntimeError("Resource with the name '{}' does not exist.") elif len(response) > 1: @@ -161,9 +181,18 @@ def post_and_structure( path: str, body: Union[List[Dict[Any, Any]], Dict[Any, Any]], **kwargs, - ) -> Union[CE, List[CE]]: + ) -> Union[CE, Gate, List[Gate], Tuple[Gate, Union[Population, List[Population]]]]: res = self._post(f"{self.base_url}/{path}", json=body, params=kwargs) - return converter.structure(res, _class) + if _class is List[Gate]: + return [self._parse_gate_population(gate)[0] for gate in res] + if _class is Gate: + gate, population = self._parse_gate_population(res) + if population: + return gate, population + else: + return gate + else: + return converter.structure(res, _class) def unstructure_and_clean(self, entity) -> Tuple[type, str, Dict[Any, Any]]: (cls, path, body) = ( @@ -186,21 +215,22 @@ def create(self, entity: Experiment, **kwargs) -> Experiment: ... @overload def create(self, entity: FcsFile, **kwargs) -> FcsFile: ... @overload - def create(self, entity: Gate, **kwargs) -> Gate: ... + def create(self, entity: Gate, **kwargs) -> _Gate: ... @overload def create(self, entity: Population, **kwargs) -> Population: ... @overload def create(self, entity: ScaleSet, **kwargs) -> ScaleSet: ... @overload - def create(self, entity: List[Gate], **kwargs) -> List[Gate]: ... + def create(self, entity: List[Gate], **kwargs) -> List[_Gate]: ... # fmt: on def create(self, entity, **kwargs): """Create a local entity on CellEngine.""" + # TODO expose create_many since typings are a mess otherwise. if isinstance(entity, list): return create_many(self, entity, **kwargs) body = self.unstructure_and_clean(entity) - return self.post_and_structure(*body) + return self.post_and_structure(*body, **kwargs) def update(self, entity, params: Dict = None): path = self._get_path(entity) @@ -454,14 +484,57 @@ def get_gates(self, experiment_id, as_dict=False) -> List[Gate]: gates = self._get(f"{self.base_url}/experiments/{experiment_id}/gates") if as_dict: return gates - return [Gate.factory(gate) for gate in gates] - - def get_gate(self, experiment_id: str, _id, as_dict=False) -> Gate: + structured_gates = [] + for gate in gates: + structured_gates.append( + self._parse_gate_population(gate)[0] + ) # return only Gate + return structured_gates + + def get_gate(self, experiment_id: str, _id: str, as_dict: bool = False) -> Gate: """Gates cannot be retrieved by name.""" gate = self._get(f"{self.base_url}/experiments/{experiment_id}/gates/{_id}") if as_dict: return gate - return Gate.factory(gate) + return self._parse_gate_population(gate)[0] # return only Gate + + def post_gates( + self, + experiment_id: str, + body: Union[Dict[str, Any], List[Dict[str, Any]]], + params: Dict = {}, + as_dict: bool = False, + ) -> List[Gate]: + gates = self._post( + f"{self.base_url}/experiments/{experiment_id}/gates", + json=body, + params=params, + ) + + if as_dict: + return gates + structured_gates = [] + for gate in gates: + structured_gates.append( + self._parse_gate_population(gate)[0] + ) # return only Gate + return structured_gates + + def post_gate( + self, + experiment_id: str, + body: Union[Dict[str, Any], List[Dict[str, Any]]], + params: Dict = {}, + as_dict: bool = False, + ) -> Gate: + gate = self._post( + f"{self.base_url}/experiments/{experiment_id}/gates", + json=body, + params=params, + ) + if as_dict: + return gate + return self._parse_gate_population(gate)[0] # return only Gate def delete_gate( self, experiment_id: str, _id: str = None, gid: str = None, exclude: str = None @@ -477,7 +550,7 @@ def delete_gate( Args: experiment_id: ID of experiment. _id: ID of the gate to delete. - gid: ID of gate family to delte. + gid: ID of gate family to delete. exclude: Gate ID to exclude from deletion. Example: @@ -500,19 +573,29 @@ def delete_gate( raise ValueError("Either _id or gid must be specified.") self._delete(url) - def post_gate( - self, experiment_id, gate: Dict, create_population=True, as_dict=False - ) -> Gate: - res = self._post( - f"{self.base_url}/experiments/{experiment_id}/gates", - json=gate, - params={"createPopulation": create_population}, + def delete_gates(self, experiment_id: str, ids: List[str]): + url = f"{self.base_url}/experiments/{experiment_id}/gates/" + [self._delete(url + _id) for _id in ids] + + def _parse_gate_population( + self, res: Any + ) -> Tuple[Gate, Union[Population, List[Population], None]]: + keys = res.keys() + if "population" in keys: + gate = res["gate"] + pop = converter.structure(res["population"], Population) + elif "populations" in keys: + gate = res["gate"] + pop = converter.structure(res["populations"], List[Population]) + else: + gate = res + pop = None + module = importlib.import_module("cellengine") + gate_subclass = getattr(module, gate["type"]) + return ( + converter.structure(gate, gate_subclass), + pop, ) - if as_dict: - return res - if type(res) is list: - return [Gate.factory(r) for r in res] - return Gate.factory(res) def update_gate_family(self, experiment_id, gid, body: dict = None) -> dict: return self._patch( @@ -522,10 +605,10 @@ def update_gate_family(self, experiment_id, gid, body: dict = None) -> dict: 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) - gate._properties["tailoredPerFile"] = True - gate._properties["fcsFileId"] = fcs_file_id - return self.update_entity(experiment_id, gate_id, "gates", gate._properties) + 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 get_plot( self, diff --git a/cellengine/utils/api_client/BaseAPIClient.py b/cellengine/utils/api_client/BaseAPIClient.py index fb5914fe..ab02dbc2 100644 --- a/cellengine/utils/api_client/BaseAPIClient.py +++ b/cellengine/utils/api_client/BaseAPIClient.py @@ -13,8 +13,12 @@ def prepare_params(params: Dict) -> Dict: - """Converts Booleans to lower-case strings. (Requests yields upper-case.)""" - return {k: str(v).lower() if type(v) == bool else v for k, v in params.items()} + """Converts Boolean values to lower-case strings (whereas `requests` yields + upper-case) and keys to camelCase.""" + return { + to_camel_case(k): str(v).lower() if type(v) == bool else v + for k, v in params.items() + } class BaseAPIClient(metaclass=AbstractSingleton): diff --git a/cellengine/utils/helpers.py b/cellengine/utils/helpers.py index 7880d928..06742d78 100644 --- a/cellengine/utils/helpers.py +++ b/cellengine/utils/helpers.py @@ -1,11 +1,13 @@ -import re from datetime import datetime -from typing import Any, Dict, List, Union +import re +from typing import Any, Dict, List, TypeVar, Union +import numpy.typing as npt ID_REGEX = re.compile(r"^[a-f0-9]{24}$", re.I) first_cap_re = re.compile("(.)([A-Z][a-z]+)") all_cap_re = re.compile("([a-z0-9])([A-Z])") +camel_re = re.compile("((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))") def is_valid_id(_id: str) -> bool: @@ -26,6 +28,10 @@ def to_camel_case(snake_str: str) -> str: return components[0] + "".join(x.title() for x in components[1:]) +def to_snake_case(camel_str: str) -> str: + return camel_re.sub(r"_\1", camel_str).lower() + + def alter_keys(payload: Union[Dict[Any, Any], List[Dict[Any, Any]]], func): """Apply `func` to alter the keys of a dict or list of dicts.""" empty = {} @@ -44,10 +50,15 @@ def alter_keys(payload: Union[Dict[Any, Any], List[Dict[Any, Any]]], func): return payload -def get_args_as_kwargs(cls_context, locals): +def get_args_as_kwargs(context, locals) -> Dict[str, Any]: + """ + Args: + context: `self` or `cls` in a class method + locals: `locals()` + """ # fmt: off - arg_names = cls_context.create.__code__.co_varnames[ - 1:cls_context.create.__code__.co_argcount + arg_names = context.create.__code__.co_varnames[ + 1:context.create.__code__.co_argcount ] # fmt: on return {key: locals[key] for key in arg_names} @@ -120,3 +131,26 @@ def datetime_to_timestamp(value: datetime) -> str: ``datetime`` objects. """ return datetime.strftime(value, "%Y-%m-%dT%H:%M:%S.%fZ") + + +T = TypeVar("T", float, npt.NDArray) + + +def normalize( + value: T, + observed_min: float, + observed_max: float, + min: float, + max: float, +) -> T: + return (max - min) / (observed_max - observed_min) * (value - observed_max) + max + + +def remove_keys_with_none_values(d: Dict[str, Any]) -> Dict[str, Any]: + new_dict = {} + for k, v in d.items(): + if isinstance(v, dict): + v = remove_keys_with_none_values(v) + if v is not None: + new_dict[k] = v + return new_dict diff --git a/cellengine/utils/parse_fcs_file_args.py b/cellengine/utils/parse_fcs_file_args.py new file mode 100644 index 00000000..5aa6e537 --- /dev/null +++ b/cellengine/utils/parse_fcs_file_args.py @@ -0,0 +1,19 @@ +from typing import Optional +import cellengine as ce + + +def parse_fcs_file_args( + experiment_id: Optional[str] = None, + tailored_per_file: Optional[bool] = False, + fcs_file_id: Optional[str] = None, + fcs_file: Optional[str] = None, +) -> str: + """Finds the fcs file ID if 'tailored_per_file' is True and either + 'fcs_file' or 'fcs_file_id' are specified. + """ + if fcs_file is not None and fcs_file_id is not None: + raise ValueError("Please specify only 'fcs_file' or 'fcs_file_id'.") + if fcs_file is not None and tailored_per_file is True: # lookup by name + _file = ce.APIClient().get_fcs_file(experiment_id=experiment_id, name=fcs_file) + fcs_file_id = _file._id + return fcs_file_id diff --git a/setup.py b/setup.py index 3134a1f0..aa9c7db5 100644 --- a/setup.py +++ b/setup.py @@ -41,17 +41,15 @@ platforms="Posix; MacOS X; Windows", install_requires=[ "attrs~=20.2", - "cattrs~=1.8", + "cattrs~=22.1", "fcsparser~=0.2", "dataclasses-json~=0.5", - "munch~=2.5", "numpy~=1.17", "pandas~=1.1", "python-dateutil~=2.8", "requests~=2.22", "requests-toolbelt~=0.9", "urllib3~=1.25", - "custom_inherit~=2.3", ], extras_require={"interactive": ["Pillow~=9.0"]}, tests_require=["pytest", "pytest-vcr"], diff --git a/tests/fixtures/api-gates.py b/tests/fixtures/api-gates.py index 9e689459..20e26a4d 100644 --- a/tests/fixtures/api-gates.py +++ b/tests/fixtures/api-gates.py @@ -3,35 +3,167 @@ @pytest.fixture(scope="function") def rectangle_gate(): - return specific_gate("RectangleGate") + return gate_types("RectangleGate") @pytest.fixture(scope="function") def ellipse_gate(): - return specific_gate("EllipseGate") + return gate_types("EllipseGate") @pytest.fixture(scope="function") def polygon_gate(): - return specific_gate("PolygonGate") + return gate_types("PolygonGate") @pytest.fixture(scope="function") def range_gate(): - return specific_gate("RangeGate") + return gate_types("RangeGate") @pytest.fixture(scope="function") def quadrant_gate(): - return specific_gate("QuadrantGate") + return gate_types("QuadrantGate") @pytest.fixture(scope="function") def split_gate(): - return specific_gate("SplitGate") + return gate_types("SplitGate") -def specific_gate(gate_type): +@pytest.fixture(scope="function") +def bulk_gate_creation_dict(): + return [ + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5d9401613afd657e233843b4", + "model": { + "ellipse": { + "angle": -0.16875182756633697, + "center": [259441.51377370575, 63059.462213950595], + "major": 113446.7481834943, + "minor": 70116.01916918601, + }, + "label": [263044.8350515464, 66662.79381443298], + "locked": False, + }, + "name": "ellipse-gui", + "names": [], + "tailored_per_file": False, + "type": "EllipseGate", + "x_channel": "FSC-A", + "y_channel": "FSC-H", + }, + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5d9365b5117dfb76dd9ed4b0", + "model": { + "label": [59456.113402061856, 193680.53608247422], + "locked": False, + "polygon": { + "vertices": [ + [59456.113402061856, 184672.1855670103], + [141432.10309278348, 181068.84536082475], + [82877.82474226804, 124316.23711340204], + [109002.0412371134, 63960.28865979381], + [44141.9175257732, 76571.97938144332], + [27926.886597938144, 107200.37113402062], + [10811.0206185567, 143233.77319587627], + [58555.278350515466, 145936.27835051547], + ] + }, + }, + "name": "poly_gate", + "names": [], + "tailored_per_file": False, + "type": "PolygonGate", + "x_channel": "FSC-A", + "y_channel": "FSC-H", + }, + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5db01cb2dd879d32d2ccde05", + "model": { + "gids": [ + "5db01cb2e4eb52e0c1047306", + "5db01cb265909ddcfd6e2807", + "5db01cb2486959d467563e08", + "5db01cb21b8e42bc6499c609", + ], + "labels": [[1, 1], [-200, 1], [-200, -200], [1, -200]], + "locked": False, + "quadrant": { + "angles": [ + 1.5707963267948966, + 3.141592653589793, + 4.71238898038469, + 0, + ], + "x": 160000, + "y": 200000, + }, + "skewable": False, + }, + "names": ["my gate (UR)", "my gate (UL)", "my gate (LL)", "my gate (LR)"], + "tailored_per_file": False, + "type": "QuadrantGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + }, + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5d960ae01070fcef1c1f5a04", + "model": { + "label": [53.802, 0.5], + "locked": False, + "range": {"x1": 12.502, "x2": 95.102, "y": 0.5}, + }, + "name": "my gate", + "names": [], + "tailored_per_file": False, + "type": "RangeGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + }, + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5d8d34993b0bb307a31d9d04", + "model": { + "label": [130000, 145000], + "locked": False, + "rectangle": {"x1": 60000, "x2": 200000, "y1": 75000, "y2": 215000}, + }, + "name": "test_rect_gate", + "names": [], + "tailored_per_file": False, + "type": "RectangleGate", + "x_channel": "FSC-W", + "y_channel": "FSC-A", + }, + { + "experiment_id": "5d38a6f79fae87499999a74b", + "fcs_file_id": None, + "gid": "5db02aff2299543efa9f7e00", + "model": { + "gids": ["5db02aff9375ffe04e55b801", "5db02aff556563a0f01c7a02"], + "labels": [[-199.9, 0.916], [0.9, 0.916]], + "locked": False, + "split": {"x": 160000, "y": 1}, + }, + "names": ["my gate (L)", "my gate (R)"], + "tailored_per_file": False, + "type": "SplitGate", + "x_channel": "FSC-A", + }, + ] + + +def gate_types(gate_type=None): gates = { "EllipseGate": { "_id": "5d9401613afd657e233843b3", @@ -50,7 +182,6 @@ def specific_gate(gate_type): }, "name": "ellipse-gui", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "EllipseGate", "xChannel": "FSC-A", @@ -79,7 +210,6 @@ def specific_gate(gate_type): }, "name": "poly_gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "PolygonGate", "xChannel": "FSC-A", @@ -112,7 +242,6 @@ def specific_gate(gate_type): "skewable": False, }, "names": ["my gate (UR)", "my gate (UL)", "my gate (LL)", "my gate (LR)"], - "parentPopulationId": None, "tailoredPerFile": False, "type": "QuadrantGate", "xChannel": "FSC-A", @@ -130,7 +259,6 @@ def specific_gate(gate_type): }, "name": "my gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RangeGate", "xChannel": "FSC-A", @@ -148,7 +276,6 @@ def specific_gate(gate_type): }, "name": "test_rect_gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RectangleGate", "xChannel": "FSC-W", @@ -166,13 +293,12 @@ def specific_gate(gate_type): "split": {"x": 160000, "y": 1}, }, "names": ["my gate (L)", "my gate (R)"], - "parentPopulationId": None, "tailoredPerFile": False, "type": "SplitGate", "xChannel": "FSC-A", }, } - return gates[gate_type] + return gates[gate_type] if gate_type else gates @pytest.fixture(scope="session") @@ -205,7 +331,6 @@ def gates(): "test gate 2 (LL)", "test gate 2 (LR)", ], - "parentPopulationId": None, "tailoredPerFile": False, "type": "QuadrantGate", "xChannel": "FSC-A", @@ -233,7 +358,6 @@ def gates(): "skewable": False, }, "names": ["subtest (UR)", "subtest (UL)", "subtest (LL)", "subtest (LR)"], - "parentPopulationId": None, "tailoredPerFile": False, "type": "QuadrantGate", "xChannel": "FSC-A", @@ -256,7 +380,6 @@ def gates(): }, "name": "rect", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RectangleGate", "xChannel": "FSC-A", @@ -293,7 +416,6 @@ def gates(): }, }, "names": ["quad (UR)", "quad (UL)", "quad (LL)", "quad (LR)"], - "parentPopulationId": "5d64abe2ca9df61349ed8e89", "tailoredPerFile": False, "type": "QuadrantGate", "xChannel": "FSC-A", @@ -311,7 +433,6 @@ def gates(): }, "name": "my gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RectangleGate", "xChannel": "FSC-A", @@ -329,7 +450,6 @@ def gates(): }, "name": "my gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RectangleGate", "xChannel": "FSC-A", @@ -347,7 +467,6 @@ def gates(): }, "name": "my gate", "names": [], - "parentPopulationId": None, "tailoredPerFile": False, "type": "RectangleGate", "xChannel": "FSC-A", diff --git a/tests/integration/test_integration.py b/tests/integration/test_integration.py index 84b46d2e..21c91638 100644 --- a/tests/integration/test_integration.py +++ b/tests/integration/test_integration.py @@ -1,8 +1,10 @@ import os import pytest import pandas +import uuid import cellengine +from cellengine.utils.api_client.APIClient import APIClient from cellengine.utils.api_client.APIError import APIError from cellengine.resources.attachment import Attachment from cellengine.resources.compensation import Compensation @@ -25,7 +27,7 @@ def client(): @pytest.fixture(scope="module") -def setup_experiment(request, client): +def setup_experiment(request, client: APIClient): print("Setting up CellEngine experiment for {}".format(__name__)) exp = cellengine.Experiment.create("new_experiment") exp.upload_fcs_file("tests/data/Acea - Novocyte.fcs") @@ -36,7 +38,7 @@ def setup_experiment(request, client): client.delete_experiment(exp._id) -def test_experiment_attachments(setup_experiment, client): +def test_experiment_attachments(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") # POST @@ -61,7 +63,7 @@ def test_experiment_attachments(setup_experiment, client): assert len(experiment.attachments) == 0 -def test_fcs_file_events(setup_experiment, client): +def test_fcs_file_events(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") file = experiment.fcs_files[0] @@ -76,7 +78,7 @@ def test_fcs_file_events(setup_experiment, client): assert len(limited_events) == len(file.events) -def test_apply_compensations(setup_experiment, client): +def test_apply_compensations(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") # POST @@ -100,15 +102,15 @@ def test_apply_compensations(setup_experiment, client): assert type(comp.dataframe) == pandas.DataFrame # with inplace=True it should save results to the target FcsFile - another_events_df = comp.apply(file1, inplace=True, preSubsampleN=10) - file1.events == another_events_df + comp.apply(file1, inplace=True, preSubsampleN=10) + # TODO assert # DELETE experiment.compensations[0].delete() assert experiment.compensations == [] -def test_apply_file_internal_compensation(setup_experiment, client): +def test_apply_file_internal_compensation(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") file = experiment.fcs_files[0] @@ -122,19 +124,20 @@ def test_apply_file_internal_compensation(setup_experiment, client): assert all([c in events_df.columns for c in comp.channels]) -def test_experiment(setup_experiment, client): +def test_experiment(setup_experiment, client: APIClient): + experiment_name = uuid.uuid4().hex + # POST - exp = cellengine.Experiment.create("new_experiment_2") + exp = cellengine.Experiment.create(experiment_name) exp.upload_fcs_file("tests/data/Acea - Novocyte.fcs") # GET experiments = client.get_experiments() - # assert len(experiments) == 2 assert all([type(exp) is Experiment for exp in experiments]) - exp2 = client.get_experiment(name="new_experiment_2") + exp2 = client.get_experiment(name=experiment_name) # UPDATE - clone = exp2.clone(name="clone") + clone = exp2.clone() clone.name = "edited_experiment" clone.update() assert clone.name == "edited_experiment" @@ -177,53 +180,54 @@ def test_experiment_fcs_files(setup_experiment, client): client.get_fcs_file(experiment._id, file._id) -def test_experiment_gates(setup_experiment, client): +def test_experiment_gates(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") fcs_file = experiment.fcs_files[0] # CREATE split_gate = SplitGate.create( - experiment._id, fcs_file.channels[0], "split_gate", 2300000, 250000 + experiment._id, + fcs_file.channels[0], + "split_gate", + 2300000, + 250000, + create_population=False, ) - split_gate.post() range_gate = RangeGate.create( experiment._id, fcs_file.channels[0], "range_gate", 2100000, 2500000, + create_population=False, ) - range_gate.post() # UPDATE - range_gate.tailor_to(fcs_file._id) - range_gate.update() + range_gate.tailor_to(fcs_file) assert range_gate.tailored_per_file is True assert range_gate.fcs_file_id == fcs_file._id - population = experiment.populations[0] Gate.update_gate_family( experiment._id, split_gate.gid, - body={"name": "new split gate name", "parentPopulationId": population._id}, + body={"name": "new split gate name"}, ) - assert experiment.gates[0].parent_population_id == population._id assert experiment.gates[0].name == "new split gate name" # DELETE range_gate.delete() assert len(experiment.gates) == 1 - Gate.delete_gates(experiment._id, gid=split_gate.gid) + experiment.delete_gate(gid=split_gate.gid) assert experiment.gates == [] -def test_experiment_populations(setup_experiment, client): +def test_experiment_populations(setup_experiment, client: APIClient): experiment = client.get_experiment(name="new_experiment") fcs_file = experiment.fcs_files[0] # GET - quad_gate = QuadrantGate.create( + quad_gate, quad_pops = QuadrantGate.create( experiment._id, fcs_file.channels[0], fcs_file.channels[1], @@ -231,20 +235,16 @@ def test_experiment_populations(setup_experiment, client): 2300000, 250000, ) - quad_gate.post() - split_gate = SplitGate.create( - experiment._id, fcs_file.channels[0], "split_gate", 2300000, 250000 + split_gate, split_pops = SplitGate.create( + experiment._id, fcs_file.channels[0], "split gate", 2300000, 250000 ) - split_gate.post() - assert "split_gate (L)", "split_gate (R)" in [ - p.name for p in experiment.populations - ] + assert ["split gate (L)", "split gate (R)"] == [p.name for p in split_pops] # CREATE complex_payload = ( ComplexPopulationBuilder("complex pop") - .Or([quad_gate.model.gids[0], quad_gate.model.gids[2]]) + .Or([quad_gate.model["gids"][0], quad_gate.model["gids"][2]]) .build() ) client.post_population(experiment._id, complex_payload) @@ -260,7 +260,12 @@ def test_experiment_populations(setup_experiment, client): assert "complex pop" not in [p.name for p in experiment.populations] -def test_create_new_fcsfile_from_s3(setup_experiment, client): +def test_create_new_fcsfile_from_s3(setup_experiment, client: APIClient): + if not "S3_ACCESS_KEY" in os.environ: + pytest.skip( + "Skipping S3 tests. Set S3_ACCESS_KEY and S3_SECRET_KEY to run them." + ) + experiment = client.get_experiment(name="new_experiment") s3_dict = { "host": "ce-test-s3-a.s3.us-east-2.amazonaws.com", diff --git a/tests/unit/resources/test_fcs_file_parse_args.py b/tests/unit/resources/test_fcs_file_parse_args.py index 9239a6d6..7a89989c 100644 --- a/tests/unit/resources/test_fcs_file_parse_args.py +++ b/tests/unit/resources/test_fcs_file_parse_args.py @@ -161,7 +161,7 @@ def test_fcs_file_and_fcs_file_id_defined( @responses.activate -def test_tailored_per_file_true(ENDPOINT_BASE, experiment, rectangle_gate): +def test_tailored_per_file_true(client, ENDPOINT_BASE, experiment, rectangle_gate): responses.add( responses.POST, ENDPOINT_BASE + f"/experiments/{EXP_ID}/gates", @@ -177,7 +177,14 @@ def test_tailored_per_file_true(ENDPOINT_BASE, experiment, rectangle_gate): y1=3, y2=4, tailored_per_file=True, + locked=True, ) + + assert json.loads(responses.calls[0].request.body)["model"]["locked"] is True + assert json.loads(responses.calls[0].request.body)["model"]["rectangle"]["x1"] == 1 + assert json.loads(responses.calls[0].request.body)["model"]["rectangle"]["x2"] == 2 + assert json.loads(responses.calls[0].request.body)["model"]["rectangle"]["y1"] == 3 + assert json.loads(responses.calls[0].request.body)["model"]["rectangle"]["y2"] == 4 assert json.loads(responses.calls[0].request.body)["tailoredPerFile"] is True diff --git a/tests/unit/resources/test_gates.py b/tests/unit/resources/test_gates.py index b328983b..313d6df8 100644 --- a/tests/unit/resources/test_gates.py +++ b/tests/unit/resources/test_gates.py @@ -1,13 +1,20 @@ import json + +from cattrs.errors import ClassValidationError import pytest import responses from cellengine.resources.gate import ( + EllipseGate, Gate, + PolygonGate, + QuadrantGate, + RangeGate, RectangleGate, - EllipseGate, + SplitGate, ) -from cellengine.utils.api_client.APIError import APIError +from cellengine.utils.helpers import is_valid_id +from tests.unit.resources.test_population import population_tester EXP_ID = "5d38a6f79fae87499999a74b" @@ -23,13 +30,12 @@ def gate_tester(instance): assert hasattr(instance, "y_channel") assert hasattr(instance, "tailored_per_file") assert hasattr(instance, "fcs_file_id") - assert hasattr(instance, "parent_population_id") assert hasattr(instance, "model") def test_init_gate(ENDPOINT_BASE, client, polygon_gate): """Test instantiating a gate object a correct dict of properties""" - g = Gate.factory(polygon_gate) + g = Gate.from_dict(polygon_gate) gate_tester(g) assert g.experiment_id == EXP_ID assert g.x_channel == "FSC-A" @@ -61,87 +67,277 @@ def test_create_one_gate(ENDPOINT_BASE, client, rectangle_gate): status=201, json=rectangle_gate, ) - g = Gate.factory(rectangle_gate) - g.post() + g = RectangleGate.create( + EXP_ID, + x_channel="FSC-A", + y_channel="FSC-W", + name="my fancy gate", + x1=12.502, + x2=95.102, + y1=1020, + y2=32021.2, + ) gate_tester(g) @responses.activate def test_create_multiple_gates_from_dicts( - ENDPOINT_BASE, client, rectangle_gate, polygon_gate + ENDPOINT_BASE, client, rectangle_gate, ellipse_gate ): responses.add( responses.POST, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", status=201, - json=[rectangle_gate, rectangle_gate, polygon_gate], + json=[rectangle_gate, ellipse_gate], ) - new_gates = [rectangle_gate, rectangle_gate, polygon_gate] - gates = Gate.bulk_create(EXP_ID, new_gates) + g1 = { + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + } + g2 = { + "experiment_id": EXP_ID, + "type": "EllipseGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my gate", + "x": 260000, + "y": 64000, + "angle": 0, + "major": 120000, + "minor": 70000, + } + + gates = Gate.create_many([g1, g2]) [gate_tester(gate) for gate in gates] +def test_throws_error_for_missing_keys_in_gate_dict(): + input = {"x_channel": "FSC-A"} + with pytest.raises(ValueError): + RectangleGate._format(**input) + + +test_params = [ + ( + # only required keys + { # input + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + } + }, + }, + { # expected + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "tailored_per_file": False, + "gid": None, + "fcs_file_id": None, + "parent_population_id": None, + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + }, + "label": [53.802, 16520.6], + "locked": False, + }, + }, + ), + ( + # accepts other keys + { # input + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "gid": "my-nice-gid", + "tailored_per_file": True, + "fcs_file_id": "some-file-id", + "parent_population_id": "some-parent-id", + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + }, + "label": [1, 1], # label + "locked": True, + }, + }, + { # expected + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "tailored_per_file": True, + "gid": "my-nice-gid", + "fcs_file_id": "some-file-id", + "parent_population_id": "some-parent-id", + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + }, + "label": [1, 1], + "locked": True, + }, + }, + ), +] + + +def test_formats_gate_dicts_correctly(bulk_gate_creation_dict): + for gate in bulk_gate_creation_dict: + if gate["type"] == "QuadrantGate": + gate["model"]["labels"] = [[1, 2], [3, 4], [5, 6], [7, 8]] + + if gate["type"] == "SplitGate": + gate["model"]["labels"] = [[1, 2], [3, 4]] + + formatted = Gate._format_gate(gate) + + gate_model = gate.pop("model") + formatted_model = formatted.pop("model") + assert all([gate[key] == formatted[key] for key in formatted.keys()]) + assert all( + [gate_model[key] == formatted_model[key] for key in formatted_model.keys()] + ) + + +gate_dicts = [ + ( + # only required keys + [ + { # input + "experiment_id": EXP_ID, + "type": "RectangleGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my fancy gate", + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + } + }, + }, + { + "experiment_id": EXP_ID, + "type": "EllipseGate", + "x_channel": "FSC-A", + "y_channel": "FSC-W", + "name": "my gate", + "model": { + "ellipse": { + "center": [260000, 64000], + "angle": 0, + "major": 120000, + "minor": 70000, + } + }, + }, + ], + [ + { # mock response + "_id": "returned-from-API", + "experimentId": EXP_ID, + "type": "RectangleGate", + "xChannel": "FSC-A", + "yChannel": "FSC-W", + "name": "my fancy gate", + "tailoredPerFile": False, + "gid": None, + "fcsFileId": None, + "model": { + "rectangle": { + "x1": 12.502, + "x2": 95.102, + "y1": 1020, + "y2": 32021.2, + }, + "label": [53.802, 16520.6], + "locked": False, + }, + }, + { + "_id": "returned-from-API", + "experimentId": EXP_ID, + "type": "EllipseGate", + "xChannel": "FSC-A", + "yChannel": "FSC-W", + "name": "my fancy gate", + "tailored_per_file": False, + "gid": None, + "fcsFileId": None, + "model": { + "ellipse": { + "center": [260000, 64000], + "angle": 0, + "major": 120000, + "minor": 70000, + }, + "label": [53.802, 16520.6], + "locked": False, + }, + }, + ], + ), +] + + @responses.activate -def test_create_multiple_gates_from_gate_objects( - ENDPOINT_BASE, client, rectangle_gate, polygon_gate -): +@pytest.mark.parametrize("input, response", gate_dicts) +def test_creates_multiple_gates_from_dicts(client, ENDPOINT_BASE, input, response): responses.add( responses.POST, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", status=201, - json=[rectangle_gate, rectangle_gate, polygon_gate], - ) - g1 = RectangleGate.create( - EXP_ID, - x_channel="FSC-A", - y_channel="FSC-W", - name="my fancy gate", - x1=12.502, - x2=95.102, - y1=1020, - y2=32021.2, - ) - g2 = RectangleGate.create( - EXP_ID, - x_channel="FSC-A", - y_channel="FSC-W", - name="my other gate", - x1=12.502, - x2=95.102, - y1=1020, - y2=32021.2, - ) - g3 = EllipseGate.create( - experiment_id=EXP_ID, - x_channel="FSC-A", - y_channel="FSC-W", - name="my gate", - x=260000, - y=64000, - angle=0, - major=120000, - minor=70000, + json=response, ) - gates = Gate.bulk_create(EXP_ID, [g1, g2, g3]) - [gate_tester(gate) for gate in gates] - # gates = Gate.factory([rectangle_gate, rectangle_gate]) - # gate_tester(gates[0]) - # gate_tester(gates[1]) - # Gate.post(gates) + gates = Gate.create_many(input) + assert isinstance(gates[0], RectangleGate) + assert isinstance(gates[1], EllipseGate) + [gate_tester(gate) for gate in gates] @pytest.fixture def bad_gate(): bad_gate = { - "__v": 0, "experimentId": "5d38a6f79fae87499999a74b", - # "name": "my gate", - "type": "PolygonGate", + "name": "my gate", + # "type": "PolygonGate", # "gid": "5dc6e4514855ff5d3d041d03", - "xChannel": "FSC-A", - "yChannel": "FSC-W", - "parentPopulationId": None, + # "xChannel": "FSC-A", + # "yChannel": "FSC-W", "model": { "polygon": {"vertices": [[1, 4], [2, 5], [3, 6]]}, "label": [2, 5], @@ -155,17 +351,10 @@ def bad_gate(): return bad_gate -@responses.activate -def test_create_gate_with_bad_params(ENDPOINT_BASE, client, bad_gate): - responses.add( - responses.POST, - f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", - status=400, - json={"error": '"gid" is required.'}, - ) - with pytest.raises(APIError): - g = Gate.factory(bad_gate) - g.post() +def test_create_gate_with_bad_params(client, bad_gate): + with pytest.raises(ClassValidationError): + g = Gate.from_dict(bad_gate) + g.update() @responses.activate @@ -174,7 +363,7 @@ def test_update_gate(ENDPOINT_BASE, client, experiment, rectangle_gate): that the correct response is made; this should be done with an integration test. """ - gate = Gate.factory(rectangle_gate) + gate = Gate.from_dict(rectangle_gate) # patch the mocked response with the correct values response = rectangle_gate.copy() response.update({"name": "newname"}) @@ -187,11 +376,11 @@ def test_update_gate(ENDPOINT_BASE, client, experiment, rectangle_gate): gate.update() gate_tester(gate) - assert json.loads(responses.calls[0].request.body) == gate._properties + assert json.loads(responses.calls[0].request.body)["name"] == "newname" # type: ignore @responses.activate -def test_update_gate_family(ENDPOINT_BASE, experiment, rectangle_gate): +def test_update_gate_family(client, ENDPOINT_BASE, experiment, rectangle_gate): gid = rectangle_gate["gid"] responses.add( responses.PATCH, @@ -214,8 +403,17 @@ def test_should_delete_gate(ENDPOINT_BASE, client, rectangle_gate): responses.DELETE, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates/{rectangle_gate['_id']}", ) - g = Gate.factory(rectangle_gate) - g.post() + g = RectangleGate.create( + EXP_ID, + x_channel="FSC-A", + y_channel="FSC-W", + name="my fancy gate", + x1=12.502, + x2=95.102, + y1=1020, + y2=32021.2, + create_population=False, + ) gate_tester(g) g.delete() @@ -243,7 +441,7 @@ def test_should_delete_gates( responses.DELETE, f"{ENDPOINT_BASE}" + url, ) - Gate.delete_gates(experiment._id, **args) + client.delete_gate(experiment._id, **args) @responses.activate @@ -260,7 +458,8 @@ def test_create_rectangle_gate(ENDPOINT_BASE, client, experiment, rectangle_gate status=201, json=rectangle_gate, ) - rectangle_gate = experiment.create_rectangle_gate( + rectangle_gate = RectangleGate.create( + experiment_id=EXP_ID, x_channel="FSC-A", y_channel="FSC-W", name="my gate", @@ -268,13 +467,62 @@ def test_create_rectangle_gate(ENDPOINT_BASE, client, experiment, rectangle_gate x2=200000, y1=75000, y2=215000, + create_population=False, ) - rectangle_gate.post() gate_tester(rectangle_gate) - assert rectangle_gate.model.rectangle.x1 == 60000 - assert rectangle_gate.model.rectangle.x2 == 200000 - assert rectangle_gate.model.rectangle.y1 == 75000 - assert rectangle_gate.model.rectangle.y2 == 215000 + m = rectangle_gate.model["rectangle"] + assert m["x1"] == 60000 + assert m["x2"] == 200000 + assert m["y1"] == 75000 + assert m["y2"] == 215000 + + +@responses.activate +def test_create_rectangle_gate_without_create_population( + ENDPOINT_BASE, client, experiment, rectangle_gate +): + """Regression test for #118.""" + responses.add( + responses.POST, + f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", + status=201, + json=rectangle_gate, + ) + rectangle_gate = experiment.create_rectangle_gate( + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + x1=60000, + x2=200000, + y1=75000, + y2=215000, + create_population=False, + ) + assert "createPopulation=false" in responses.calls[0].request.url # type: ignore + + +@responses.activate +def test_create_rectangle_gate_creates_gate_and_population( + ENDPOINT_BASE, client, rectangle_gate, populations +): + responses.add( + responses.POST, + f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", + status=201, + json={"population": populations[0], "gate": rectangle_gate}, + ) + gate, population = RectangleGate.create( + EXP_ID, + x_channel="FSC-A", + y_channel="FSC-W", + name="my gate", + x1=60000, + x2=200000, + y1=75000, + y2=215000, + ) + gate_tester(gate) + population_tester(population) @responses.activate @@ -285,22 +533,23 @@ def test_create_ellipse_gate(ENDPOINT_BASE, client, experiment, ellipse_gate): status=201, json=ellipse_gate, ) - ellipse_gate = experiment.create_ellipse_gate( + ellipse_gate = EllipseGate.create( + experiment_id=EXP_ID, x_channel="FSC-A", y_channel="FSC-W", name="my gate", - x=259441.51377370575, - y=63059.462213950595, + center=[259441.51377370575, 63059.462213950595], angle=-0.16875182756633697, major=113446.7481834943, minor=70116.01916918601, + create_population=False, ) - ellipse_gate.post() gate_tester(ellipse_gate) - assert ellipse_gate.model.ellipse.center == [259441.51377370575, 63059.462213950595] - assert ellipse_gate.model.ellipse.major == 113446.7481834943 - assert ellipse_gate.model.ellipse.minor == 70116.01916918601 - assert ellipse_gate.model.ellipse.angle == -0.16875182756633697 + m = ellipse_gate.model["ellipse"] + assert m["center"] == [259441.51377370575, 63059.462213950595] + assert m["major"] == 113446.7481834943 + assert m["minor"] == 70116.01916918601 + assert m["angle"] == -0.16875182756633697 @responses.activate @@ -311,15 +560,16 @@ def test_create_polygon_gate(ENDPOINT_BASE, client, experiment, polygon_gate): status=201, json=polygon_gate, ) - polygon_gate = experiment.create_polygon_gate( + polygon_gate = PolygonGate.create( + experiment_id=EXP_ID, x_channel="FSC-A", y_channel="FSC-W", name="my gate", vertices=[[1, 4], [2, 5], [3, 6]], + create_population=False, ) - polygon_gate.post() gate_tester(polygon_gate) - assert polygon_gate.model.polygon.vertices == [ + assert polygon_gate.model["polygon"]["vertices"] == [ [59456.113402061856, 184672.1855670103], [141432.10309278348, 181068.84536082475], [82877.82474226804, 124316.23711340204], @@ -332,25 +582,32 @@ def test_create_polygon_gate(ENDPOINT_BASE, client, experiment, polygon_gate): @responses.activate -def test_create_range_gate(ENDPOINT_BASE, experiment, range_gate): +def test_create_range_gate(ENDPOINT_BASE, client, range_gate): responses.add( responses.POST, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/gates", status=201, json=range_gate, ) - range_gate = experiment.create_range_gate( - x_channel="FSC-A", name="my gate", x1=12.502, x2=95.102 + range_gate = RangeGate.create( + experiment_id=EXP_ID, + x_channel="FSC-A", + name="my gate", + x1=12.502, + x2=95.102, + create_population=False, ) - range_gate.post() gate_tester(range_gate) - assert range_gate.model.range.x1 == 12.502 - assert range_gate.model.range.x2 == 95.102 - assert range_gate.model.range.y == 0.5 + m = range_gate.model["range"] + assert m["x1"] == 12.502 + assert m["x2"] == 95.102 + assert m["y"] == 0.5 @responses.activate -def test_create_quadrant_gate(ENDPOINT_BASE, experiment, scalesets, quadrant_gate): +def test_create_quadrant_gate( + ENDPOINT_BASE, client, experiment, scalesets, quadrant_gate +): responses.add( responses.GET, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/scalesets", @@ -362,24 +619,26 @@ def test_create_quadrant_gate(ENDPOINT_BASE, experiment, scalesets, quadrant_gat status=201, json=quadrant_gate, ) - quadrant_gate = experiment.create_quadrant_gate( + quadrant_gate = QuadrantGate.create( + experiment_id=EXP_ID, name="test_quadrant_gate", x_channel="FSC-A", y_channel="FSC-W", x=160000, y=200000, + create_population=False, ) - quadrant_gate.post() gate_tester(quadrant_gate) - assert quadrant_gate.model.quadrant.x == 160000 - assert quadrant_gate.model.quadrant.y == 200000 - assert quadrant_gate.model.quadrant.angles == [ + m = quadrant_gate.model + assert m["quadrant"]["x"] == 160000 + assert m["quadrant"]["y"] == 200000 + assert m["quadrant"]["angles"] == [ 1.5707963267948966, 3.141592653589793, 4.71238898038469, 0, ] - assert quadrant_gate.model.gids == [ + assert m["gids"] == [ "5db01cb2e4eb52e0c1047306", "5db01cb265909ddcfd6e2807", "5db01cb2486959d467563e08", @@ -391,11 +650,71 @@ def test_create_quadrant_gate(ENDPOINT_BASE, experiment, scalesets, quadrant_gat "my gate (LL)", "my gate (LR)", ] - assert quadrant_gate.model.labels == [[1, 1], [-200, 1], [-200, -200], [1, -200]] + assert quadrant_gate.model["labels"] == [[1, 1], [-200, 1], [-200, -200], [1, -200]] + + +def test_formats_all_gate_defaults_correctly(): + qg = QuadrantGate._format( + experiment_id=EXP_ID, + name="test_quadrant_gate", + x_channel="FSC-A", + y_channel="FSC-W", + x=160000, + y=200000, + labels=[[1, 1], [-200, 1], [-200, -200], [1, -200]], + ) + assert qg["experiment_id"] == EXP_ID + assert qg["names"] == [ + "test_quadrant_gate (UR)", + "test_quadrant_gate (UL)", + "test_quadrant_gate (LL)", + "test_quadrant_gate (LR)", + ] + assert qg["x_channel"] == "FSC-A" + assert qg["y_channel"] == "FSC-W" + m = qg["model"] + assert m["quadrant"]["x"] == 160000 + assert m["quadrant"]["y"] == 200000 + assert m["quadrant"]["angles"] == [ + 0, + 1.5707963267948966, + 3.141592653589793, + 4.71238898038469, + ] + assert all(is_valid_id(id) for id in m["gids"]) + assert m["labels"] == [[1, 1], [-200, 1], [-200, -200], [1, -200]] + + +@responses.activate +def test_formats_gate_defaults_and_generates_nudged_labels( + ENDPOINT_BASE, client, experiment, scalesets, quadrant_gate +): + # makes a request for the scaleset + responses.add( + responses.GET, + f"{ENDPOINT_BASE}/experiments/{EXP_ID}/scalesets", + json=[scalesets], + ) + + qg = QuadrantGate._format( + experiment_id=EXP_ID, + name="test_quadrant_gate", + x_channel="FSC-A", + y_channel="FSC-W", + x=160000, + y=200000, + ) + p = qg["model"]["labels"] + assert qg["model"]["labels"] == [ + [233217, 247680], + [36158, 247680], + [36158, 13560], + [233217, 13560], + ] @responses.activate -def test_create_split_gate(ENDPOINT_BASE, experiment, scalesets, split_gate): +def test_create_split_gate(client, ENDPOINT_BASE, experiment, scalesets, split_gate): responses.add( responses.GET, f"{ENDPOINT_BASE}/experiments/{EXP_ID}/scalesets", @@ -407,16 +726,21 @@ def test_create_split_gate(ENDPOINT_BASE, experiment, scalesets, split_gate): status=201, json=split_gate, ) - split_gate = experiment.create_split_gate( - x_channel="FSC-A", name="my gate", x=160000, y=100000 + split_gate = SplitGate.create( + experiment_id=EXP_ID, + x_channel="FSC-A", + name="my gate", + x=160000, + y=100000, + create_population=False, ) - split_gate.post() gate_tester(split_gate) - assert split_gate.model.split.x == 160000 - assert split_gate.model.split.y == 1 - assert split_gate.model.gids == [ + m = split_gate.model["split"] + assert m["x"] == 160000 + assert m["y"] == 1 + assert split_gate.model["gids"] == [ "5db02aff9375ffe04e55b801", "5db02aff556563a0f01c7a02", ] assert split_gate.names == ["my gate (L)", "my gate (R)"] - assert split_gate.model.labels == [[-199.9, 0.916], [0.9, 0.916]] + assert split_gate.model["labels"] == [[-199.9, 0.916], [0.9, 0.916]] diff --git a/tests/unit/utils/test_normalize_denormalize.py b/tests/unit/utils/test_normalize_denormalize.py new file mode 100644 index 00000000..a10f16a1 --- /dev/null +++ b/tests/unit/utils/test_normalize_denormalize.py @@ -0,0 +1,65 @@ +from math import isclose +from numpy import array +from cellengine.utils.helpers import normalize + + +def test_normalizes_single_value_to_range(): + # Given + rmin = 1 + rmax = 10 + range = [10, 20, 40] + + # When + x = normalize(15, min(range), max(range), rmin, rmax) + + # Then + assert x <= 10 + assert x >= 1 + + +def test_normalizes_values_to_range(): + # Given + rmin = 1 + rmax = 10 + range = array([10, 40]) + + # When + x, y = normalize(range, range.min(), range.max(), rmin, rmax) + + # Then + assert isclose(x, rmin) + assert isclose(y, rmax) + + +def test_normalizes_negative_values_to_range(): + # Given + vals = array([[262144, 262144], [-200, 262144], [-200, -200], [262144, -200]]) + + # When + x = normalize(vals, vals.min(), vals.max(), 0, 290) + + # Then + assert x.min() == 0 + assert x.max() == 290 + + assert (x == array([[290.0, 290.0], [0.0, 290.0], [0.0, 0.0], [290.0, 0.0]])).all() + + +def test_denormalizes_values_to_original_range(): + # Given + vals = array( + [ + [262144, 262144], + [-200, 262144], + [-200, -200], + [262144, -200], + [261944, 10000], + ] + ) + + # When + normalized = normalize(vals, vals.min(), vals.max(), 0, 290) + denormalized = normalize(normalized, 0, 290, vals.min(), vals.max()) + + # Then + assert (denormalized.astype(int) == vals).all()