diff --git a/cellengine/resources/fcs_file.py b/cellengine/resources/fcs_file.py index e46faef2..bceccf07 100644 --- a/cellengine/resources/fcs_file.py +++ b/cellengine/resources/fcs_file.py @@ -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): @@ -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): @@ -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 diff --git a/cellengine/utils/api_client/APIClient.py b/cellengine/utils/api_client/APIClient.py index a9793e3f..2596dfb8 100644 --- a/cellengine/utils/api_client/APIClient.py +++ b/cellengine/utils/api_client/APIClient.py @@ -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}" diff --git a/tests/unit/resources/test_fcsfile.py b/tests/unit/resources/test_fcsfile.py index c61cb4f6..5c80a4c8 100644 --- a/tests/unit/resources/test_fcsfile.py +++ b/tests/unit/resources/test_fcsfile.py @@ -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 @@ -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())