Skip to content

bug(fcsfile): fix recursion when file has no spill string #145

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 8, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions cellengine/resources/fcs_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ class FcsFile(DataClassMixin):
sample_name: Optional[str] = field(default=ReadOnly(optional=True)) # type: ignore
size: int = field(default=ReadOnly()) # type: ignore
_spill_string: Optional[str] = field(
metadata=config(field_name="spillString"), default=None
metadata=config(field_name="spillString"),
default=None,
)

def __repr__(self):
Expand Down Expand Up @@ -245,7 +246,11 @@ def plot(

def get_file_internal_compensation(self) -> Compensation:
"""Get the file-internal Compensation."""
return Compensation.from_spill_string(self.spill_string)
if not self.has_file_internal_comp:
raise ValueError(
f"FCS File '{self._id}' does not have an internal compensation."
)
return Compensation.from_spill_string(self.spill_string) # type: ignore

@property
def events(self):
Expand All @@ -271,12 +276,11 @@ def events(self, events):

@property
def spill_string(self):
if self._spill_string:
return self._spill_string
else:
ss = ce.APIClient().get_fcs_file(self.experiment_id, self._id).spill_string
self._spill_string = ss
return ss
if not self._spill_string and self.has_file_internal_comp:
self._spill_string = ce.APIClient().get_fcs_file(
self.experiment_id, self._id, as_dict=True
)["spillString"]
return self._spill_string

def get_events(
self, inplace: bool = False, destination=None, **kwargs
Expand Down
24 changes: 22 additions & 2 deletions cellengine/utils/api_client/APIClient.py
Original file line number Diff line number Diff line change
Expand Up @@ -320,9 +320,29 @@ def get_fcs_files(self, experiment_id, as_dict=False) -> List[FcsFile]:
return fcs_files
return [FcsFile.from_dict(fcs_file) for fcs_file in fcs_files]

# fmt: off
@overload
def get_fcs_file(
self, experiment_id, _id=None, name=None, as_dict=False
) -> FcsFile:
self,
experiment_id: str,
_id: str = None,
name: str = None,
as_dict: bool = True,
) -> Dict[str, Any]: ...

@overload
def get_fcs_file(
self,
experiment_id: str,
_id: str = None,
name: str = None,
as_dict: bool = False,
) -> FcsFile: ...
# fmt: on

def get_fcs_file(
self, experiment_id: str, _id: str = None, name: str = None, as_dict=False
):
_id = _id or self._get_id_by_name(name, "fcsfiles", experiment_id)
fcs_file = self._get(
f"{self.base_url}/experiments/{experiment_id}/fcsfiles/{_id}"
Expand Down
34 changes: 31 additions & 3 deletions tests/unit/resources/test_fcsfile.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
import os
from io import BufferedReader, BytesIO
import json
import os

from fcsparser.api import FCSParser
from pandas.core.frame import DataFrame
import pytest
import responses
from io import BufferedReader, BytesIO

from cellengine.resources.fcs_file import FcsFile
from cellengine.resources.compensation import Compensation
from cellengine.resources.fcs_file import FcsFile
from cellengine.utils.helpers import to_camel_case


Expand Down Expand Up @@ -80,6 +82,32 @@ def test_gets_file_internal_compensation(ENDPOINT_BASE, client, fcs_files, spill
assert type(comp) == Compensation


@responses.activate
def test_throws_correct_error_when_no_file_internal_compensation(
ENDPOINT_BASE, client, fcs_files
):
# Given: An FcsFile with no spill string
expected_response = fcs_files[0].copy()
file_data = fcs_files[0].copy()
file_data["spillString"] = None
file_data["hasFileInternalComp"] = False
file = FcsFile.from_dict(file_data)
responses.add(
responses.GET,
f"{ENDPOINT_BASE}/experiments/{EXP_ID}/fcsfiles/{file._id}",
json=expected_response,
)

# When:
with pytest.raises(ValueError) as err:
comp = file.get_file_internal_compensation()
# Then:
assert (
err.value.args[0]
== f"FCS File '{file._id}' does not have an internal compensation."
)


def test_parse_fcs_file():
events_body = open("tests/data/Acea - Novocyte.fcs", "rb")
parser = FCSParser.from_data(events_body.read())
Expand Down