Allow access to simulation inputs and synapse replay files (#231)

* Add support for simulation.input and access for sonata reader instance of synapse replay file

* tackle some of the review comments

* Fix: pycodestyle raises "line too long" on _version.py.

The file should not even be checked.
Disabled line length check on pycodestyle: it is handled by other linters.

* Replaced class Input with a function

* ensure coverage is not bothered by  _version.py

* Update CHANGELOG.rst

Co-authored-by: Gianluca Ficarelli <[email protected]>

* Update test_input.py

* restrict pylint<3.0.0, until the cyclic-import issue is resolved

---------

Co-authored-by: Gianluca Ficarelli <[email protected]>

CHANGELOG.rst

@@ -7,6 +7,10 @@ Version v1.1.0
 New Features
 ~~~~~~~~~~~~
 - ``NodeSets`` object can be instantiated with three methods: ``from_file``, ``from_string``, ``from_dict``
+- Simulation inputs are now accessible with ``Simulation.inputs``
+- ``libsonata`` reader of ``synapse_replay`` files can now be accessed with ``simulation.inputs["<input_name>"].reader``
+
+  - only ``h5`` format is supported
 
 Improvements
 ~~~~~~~~~~~~

bluepysnap/input.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2020, EPFL/Blue Brain Project
+
+# This file is part of BlueBrain SNAP library <https://github.com/BlueBrain/snap>
+
+# This library is free software; you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License version 3.0 as published
+# by the Free Software Foundation.
+
+# This library is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+
+# You should have received a copy of the GNU Lesser General Public License
+# along with this library; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+"""Simulation input access."""
+import libsonata
+
+from bluepysnap import BluepySnapError
+
+
+class SynapseReplay:
+    """Wrapper class for libsonata.SynapseReplay to provide the reader as a property."""
+
+    def __init__(self, instance):
+        """Wrap libsonata SynapseReplay object.
+
+        Args:
+            instance (libsonata.SynapseReplay): instance to wrap
+        """
+        self._instance = instance
+
+    def __dir__(self):
+        """Provide wrapped SynapseReplay instance's public attributes in dir."""
+        public_attrs_instance = {attr for attr in dir(self._instance) if not attr.startswith("_")}
+        return list(set(super().__dir__()) | public_attrs_instance)
+
+    def __getattr__(self, name):
+        """Retrieve attributes from the wrapped object."""
+        return getattr(self._instance, name)
+
+    @property
+    def reader(self):
+        """Return a spike reader object for the instance."""
+        return libsonata.SpikeReader(self.spike_file)
+
+
+def get_simulation_inputs(simulation):
+    """Get simulation inputs as a dictionary.
+
+    Args:
+        simulation (libsonata.SimulationConfig): libsonata Simulation instance
+
+    Returns:
+        dict: inputs with input names as keys and corresponding objects as values
+    """
+
+    def _get_input(name):
+        """Helper function to wrap certain objects."""
+        item = simulation.input(name)
+
+        if item.module.name == "synapse_replay":
+            return SynapseReplay(item)
+        return item
+
+    if isinstance(simulation, libsonata.SimulationConfig):
+        return {name: _get_input(name) for name in simulation.list_input_names}
+
+    raise BluepySnapError(f"Unexpected type for 'simulation': {simulation.__class__.__name__}")

bluepysnap/simulation.py

@@ -23,6 +23,7 @@
 
 from bluepysnap.config import SimulationConfig
 from bluepysnap.exceptions import BluepySnapError
+from bluepysnap.input import get_simulation_inputs
 from bluepysnap.node_sets import NodeSets
 
 
@@ -96,6 +97,11 @@ def output(self):
         """Access the output section."""
         return self.to_libsonata.output
 
+    @property
+    def inputs(self):
+        """Access the inputs section."""
+        return get_simulation_inputs(self.to_libsonata)
+
     @property
     def run(self):
         """Access to the complete run dictionary for this simulation."""

pyproject.toml

@@ -15,8 +15,9 @@ known_local_folder  =  [
 
 [tool.coverage.run]
 omit = [
-    "bluepysnap/_version.py",
+    'bluepysnap/_version.py',
 ]
+
 [tool.pylint.main]
 # Files or directories to be skipped. They should be base names, not paths.
 ignore = ["CVS", "_version.py"]

tests/data/simulation_config.json

@@ -26,6 +26,14 @@
       "amp_start": 190.0,
       "delay": 100.0,
       "duration": 800.0
+    },
+    "spikes_1":{
+        "input_type": "spikes",
+        "module": "synapse_replay",
+        "delay": 800,
+        "duration": 100,
+        "node_set": "Layer23",
+        "spike_file": "input_spikes.h5"
     }
   },
 

tests/test_input.py

@@ -0,0 +1,46 @@
+import libsonata
+import numpy.testing as npt
+import pytest
+
+import bluepysnap.input as test_module
+from bluepysnap.exceptions import BluepySnapError
+
+from utils import TEST_DATA_DIR
+
+
+class TestSynapseReplay:
+    def setup_method(self):
+        simulation = libsonata.SimulationConfig.from_file(TEST_DATA_DIR / "simulation_config.json")
+        self.test_obj = test_module.SynapseReplay(simulation.input("spikes_1"))
+
+    def test_all(self):
+        snap_attrs = {a for a in dir(self.test_obj) if not a.startswith("_")}
+        libsonata_attrs = {a for a in dir(self.test_obj._instance) if not a.startswith("_")}
+
+        # check that wrapped instance's public methods are available in the object
+        assert snap_attrs.symmetric_difference(libsonata_attrs) == {"reader"}
+        assert isinstance(self.test_obj.reader, libsonata.SpikeReader)
+
+        for a in libsonata_attrs:
+            assert getattr(self.test_obj, a) == getattr(self.test_obj._instance, a)
+
+        npt.assert_almost_equal(self.test_obj.reader["default"].get(), [[0, 10.775]])
+
+    def test_no_such_attribute(self):
+        """Check that the attribute error is raised from the wrapped libsonata object."""
+        with pytest.raises(AttributeError, match="libsonata._libsonata.SynapseReplay"):
+            self.test_obj.no_such_attribute
+
+
+def test_get_simulation_inputs():
+    simulation = libsonata.SimulationConfig.from_file(TEST_DATA_DIR / "simulation_config.json")
+    inputs = test_module.get_simulation_inputs(simulation)
+
+    assert isinstance(inputs, dict)
+    assert inputs.keys() == {"spikes_1", "current_clamp_1"}
+
+    assert isinstance(inputs["spikes_1"], test_module.SynapseReplay)
+    assert isinstance(inputs["current_clamp_1"], libsonata._libsonata.Linear)
+
+    with pytest.raises(BluepySnapError, match="Unexpected type for 'simulation': str"):
+        test_module.get_simulation_inputs("fail_me")

tests/test_simulation.py

@@ -66,6 +66,7 @@ def test_all():
     }
     assert simulation.node_sets.content == expected_content
 
+    assert isinstance(simulation.inputs, dict)
     assert isinstance(simulation.spikes, SpikeReport)
     assert isinstance(simulation.spikes["default"], PopulationSpikeReport)