Merge branch 'main' into keep-legacy-access-point

bluepyemodel/evaluation/fitness_calculator_configuration.py

@@ -19,55 +19,58 @@
 import logging
 from copy import deepcopy
 
+from bluepyopt.ephys.locations import EPhysLocInstantiateException
+
 from bluepyemodel.evaluation.efeature_configuration import EFeatureConfiguration
 from bluepyemodel.evaluation.evaluator import LEGACY_PRE_PROTOCOLS
 from bluepyemodel.evaluation.evaluator import PRE_PROTOCOLS
+from bluepyemodel.evaluation.evaluator import define_location
 from bluepyemodel.evaluation.evaluator import seclist_to_sec
 from bluepyemodel.evaluation.protocol_configuration import ProtocolConfiguration
 from bluepyemodel.tools.utils import are_same_protocol
 
 logger = logging.getLogger(__name__)
 
 
-def _set_morphology_dependent_locations(stimulus, cell):
+def _set_morphology_dependent_locations(recording, cell):
     """Here we deal with morphology dependent locations"""
 
-    def _get_stim(stimulus, sec_id):
-        new_stim = deepcopy(stimulus)
-        stim_split = stimulus["name"].split(".")
-        new_stim["type"] = "nrnseclistcomp"
-        new_stim["name"] = f"{'.'.join(stim_split[:-1])}_{sec_id}.{stim_split[-1]}"
-        new_stim["sec_index"] = sec_id
-        return new_stim
-
-    new_stims = []
-    if stimulus["type"] == "somadistanceapic":
-        new_stims = [deepcopy(stimulus)]
-        new_stims[0]["sec_name"] = seclist_to_sec.get(
-            stimulus["seclist_name"], stimulus["seclist_name"]
+    def _get_rec(recording, sec_id):
+        new_rec = deepcopy(recording)
+        rec_split = recording["name"].split(".")
+        new_rec["type"] = "nrnseclistcomp"
+        new_rec["name"] = f"{'.'.join(rec_split[:-1])}_{sec_id}.{rec_split[-1]}"
+        new_rec["sec_index"] = sec_id
+        return new_rec
+
+    new_recs = []
+    if recording["type"] == "somadistanceapic":
+        new_recs = [deepcopy(recording)]
+        new_recs[0]["sec_name"] = seclist_to_sec.get(
+            recording["seclist_name"], recording["seclist_name"]
         )
 
-    elif stimulus["type"] == "terminal_sections":
+    elif recording["type"] == "terminal_sections":
         # all terminal sections
-        for sec_id, section in enumerate(getattr(cell.icell, stimulus["seclist_name"])):
+        for sec_id, section in enumerate(getattr(cell.icell, recording["seclist_name"])):
             if len(section.subtree()) == 1:
-                new_stims.append(_get_stim(stimulus, sec_id))
+                new_recs.append(_get_rec(recording, sec_id))
 
-    elif stimulus["type"] == "all_sections":
+    elif recording["type"] == "all_sections":
         # all section of given type
-        for sec_id, section in enumerate(getattr(cell.icell, stimulus["seclist_name"])):
-            new_stims.append(_get_stim(stimulus, sec_id))
+        for sec_id, section in enumerate(getattr(cell.icell, recording["seclist_name"])):
+            new_recs.append(_get_rec(recording, sec_id))
 
     else:
-        new_stims = [deepcopy(stimulus)]
+        new_recs = [deepcopy(recording)]
 
-    if len(new_stims) == 0 and stimulus["type"] in [
+    if len(new_recs) == 0 and recording["type"] in [
         "somadistanceapic",
         "terminal_sections",
         "all_sections",
     ]:
-        logger.warning("We could not add a location for %s", stimulus)
-    return new_stims
+        logger.warning("We could not add a location for %s", recording)
+    return new_recs
 
 
 class FitnessCalculatorConfiguration:
@@ -501,22 +504,38 @@ def configure_morphology_dependent_locations(self, _cell, simulator):
 
         # TODO: THE SAME FOR STIMULI
 
+        skipped_recordings = []
         for i, protocol in enumerate(self.protocols):
             recordings = []
             for j, rec in enumerate(protocol.recordings):
                 if rec["type"] != "CompRecording":
                     for _rec in _set_morphology_dependent_locations(rec, cell):
-                        recordings.append(_rec)
+                        try:
+                            location = define_location(_rec)
+                            location.instantiate(sim=simulator, icell=cell.icell)
+                            recordings.append(_rec)
+                        except EPhysLocInstantiateException:
+                            logger.warning(
+                                "Could not find %s, ignoring recording at this location",
+                                location.name,
+                            )
+                            skipped_recordings.append(_rec["name"])
                 else:
                     recordings.append(self.protocols[i].recordings[j])
             self.protocols[i].recordings = recordings
 
-        # if the loc of the recording is of the form axon*.v, we replace * by
-        # all the corresponding int from the created recordings
         to_remove = []
         efeatures = []
         for i, efeature in enumerate(self.efeatures):
             if isinstance(efeature.recording_name, str):
+                # remove efeature associated to skipped recording
+                for skiprec in skipped_recordings:
+                    if f"{efeature.protocol_name}.{efeature.recording_name}" == skiprec:
+                        to_remove.append(i)
+                        logger.warning("Removing %s", efeature.name)
+                        continue
+                # if the loc of the recording is of the form axon*.v, we replace * by
+                # all the corresponding int from the created recordings
                 loc_name, rec_name = efeature.recording_name.split(".")
                 if loc_name[-1] == "*":
                     to_remove.append(i)

tests/unit_tests/test_fitness_calculator_configuration.py

@@ -16,8 +16,15 @@
 
 import pytest
 
+from bluepyemodel.model import model
+
+from bluepyemodel.evaluation.evaluator import get_simulator
 from bluepyemodel.evaluation.fitness_calculator_configuration import FitnessCalculatorConfiguration
 
+from tests.unit_tests.test_local_access_point import api_config
+from tests.unit_tests.test_local_access_point import db
+from tests.unit_tests.test_local_access_point import DATA
+
 
 @pytest.fixture
 def config_dict():
@@ -94,6 +101,86 @@ def config_dict():
     return config_dict
 
 
+@pytest.fixture
+def config_dict_bad_recordings():
+
+    efeatures = [
+        {
+            "efel_feature_name": "maximum_voltage_from_voltagebase",
+            "protocol_name": "Step_150",
+            "recording_name": "soma.v",
+            "efel_settings": {"stim_start": 700, "stim_end": 2700},
+            "mean": -83.1596,
+            "std": 1.0102,
+            "threshold_efeature_std": 0.05
+        },
+        {
+            "efel_feature_name": "maximum_voltage_from_voltagebase",
+            "protocol_name": "Step_150",
+            "recording_name": "dend100.v",
+            "efel_settings": {"stim_start": 700, "stim_end": 2700},
+            "mean": -83.1596,
+            "std": 1.0102,
+            "threshold_efeature_std": 0.05
+        },
+        {
+            "efel_feature_name": "maximum_voltage_from_voltagebase",
+            "protocol_name": "Step_150",
+            "recording_name": "dend10000.v",
+            "efel_settings": {"stim_start": 700, "stim_end": 2700},
+            "mean": -83.1596,
+            "std": 1.0102,
+            "threshold_efeature_std": 0.05
+        },
+    ]
+
+    protocols = [
+        {
+            "name": "Step_150",
+            "stimuli": [{
+                "location": "soma",
+                "delay": 700.0,
+                "amp": None,
+                "thresh_perc": 148.7434,
+                "duration": 2000.0,
+                "totduration": 3000.0,
+                "holding_current": None
+            }],
+            "recordings": [
+                {
+                "type": "CompRecording",
+                "name": "Step_150.soma.v",
+                "location": "soma",
+                "variable": "v"
+                },
+                {
+                "type": "somadistanceapic",
+                "somadistance": 100,
+                "name": "Step_150.dend100.v",
+                "seclist_name": "apical",
+                "variable": "v"
+                },
+                {
+                "type": "somadistanceapic",
+                "somadistance": 10000,
+                "name": "Step_150.dend10000.v",
+                "seclist_name": "apical",
+                "variable": "v"
+                }
+            ],
+            "validation": False
+        }
+    ]
+
+    config_dict = {
+        "efeatures": efeatures,
+        "protocols": protocols,
+        "name_rmp_protocol": "IV_-40",
+        "name_rin_protocol": "IV_0",
+    }
+
+    return config_dict
+
 @pytest.fixture
 def config_dict_from_bpe2():
 
@@ -198,3 +285,38 @@ def test_init_from_bpe2(config_dict, config_dict_from_bpe2):
         "ohmic_input_resistance_vb_ssse"
     ]:
         assert next((f for f in config.efeatures if f.efel_feature_name == fn), False)
+
+
+def test_configure_morphology_dependent_locations(config_dict_bad_recordings, db):
+    """Test configure_morphology_dependent_locations with recordings outside of cell."""
+    config = FitnessCalculatorConfiguration(**config_dict_bad_recordings)
+
+    model_configuration = db.get_model_configuration()
+    cell_model = model.create_cell_model(
+        name=db.emodel_metadata.emodel,
+        model_configuration=model_configuration,
+        morph_modifiers=None,
+    )
+
+    # don't know if I have to specify the mechanism directory here
+    simulator = get_simulator(
+        stochasticity=False,
+        cell_model=cell_model,
+    )
+
+    config.configure_morphology_dependent_locations(cell_model, simulator)
+
+    assert len(config.protocols) == 1
+    # one recording should have been removed
+    assert len(config.protocols[0].recordings) == 2
+    for rec in config.protocols[0].recordings:
+        if rec["type"] == "somadistanceapic":
+            # check _set_morphology_dependent_locations section name replacement
+            assert rec["sec_name"] == "apic"
+            # check that this recording has been removed
+            assert rec["somadistance"] != 10000
+    # one efeature should have been removed
+    assert len(config.efeatures) == 2
+    for feat in config.efeatures:
+        # this efeature should have been removed
+        assert feat.recording_name != "dend10000.v"