Merge pull request nest#3643 from heplesser/test_rate_neurons_mpi

Port test_rate_neurons_mpi from SLI to Py (and improve MPITestWrapper)

Author: heplesser

testsuite/mpitests/test_rate_neurons_mpi.sli

@@ -49,8 +49,8 @@
          - `MULTI_LABEL`
          - `OTHER_LABEL`
       They must be used as `label` for spike recorders and multimeters, respectively,
-      or for other files for output data (CSV files). They are format strings expecting
-      the number of processes with which NEST is run as argument.
+      or for other files for output data (TAB-separated CSV files). They are format
+      strings expecting the number of processes with which NEST is run as argument.
 - Set `debug=True` on the decorator to see debug output and keep the
   temporary directory that has been created (latter works only in
   Python 3.12 and later)
@@ -239,10 +239,13 @@ def _collect_result_by_label(self, tmpdirpath, label):
             label += "-{}.dat"
 
         try:
-            next(tmpdirpath.glob(label.format("*", "*")))
+            first_file = next(tmpdirpath.glob(label.format("*", "*")))
         except StopIteration:
             return None  # no data for this label
 
+        # Confirm we have tab-separated data. Assumes that all data have at least two columns.
+        assert "\t" in open(first_file).read(), "All data files must be tab-separated"
+
         res = {}
         for n_procs in self._procs_lst:
             data = []
@@ -255,6 +258,18 @@ def _collect_result_by_label(self, tmpdirpath, label):
 
         return res
 
+    @staticmethod
+    def _drop_empty_dataframes(data):
+        """
+        Return list of non-empty dataframes in data.
+
+        The data frames collected for a given number of processes may contain empty
+        dataframes. pandas.concat() will not support them any more in the future, so
+        we filter them out for tests that use concat().
+        """
+
+        return [df for df in data if not df.empty]
+
     def collect_results(self, tmpdirpath):
         """
         For each of the result types, build a dictionary mapping number of MPI procs to a list of
@@ -280,36 +295,52 @@ def assert_correct_results(self, tmpdirpath):
         all_res = []
         if self._spike:
             # For each number of procs, combine results across VPs and sort by time and sender
+
+            # Include only frames containing at least one non-nan value so pandas knows datatypes.
+            # .all() returns True for empty arrays.
             all_res.append(
                 [
-                    pd.concat(spikes, ignore_index=True).sort_values(
+                    pd.concat(self._drop_empty_dataframes(spikes), ignore_index=True).sort_values(
                         by=["time_step", "time_offset", "sender"], ignore_index=True
                     )
                     for spikes in self._spike.values()
                 ]
             )
 
         if self._multi:
-            raise NotImplementedError("MULTI is not ready yet")
+            # For each number of procs, combine results across VPs and sort by time and sender
+            # Include only frames containing at least one non-nan value so pandas knows datatypes.
+            # .all() returns True for empty arrays.
+            all_res.append(
+                [
+                    pd.concat(self._drop_empty_dataframes(mmdata), ignore_index=True).sort_values(
+                        by=["time_step", "time_offset", "sender"], ignore_index=True
+                    )
+                    for mmdata in self._multi.values()
+                ]
+            )
 
         if self._other:
             # For each number of procs, combine across ranks or VPs (depends on what test has written) and
             # sort by all columns so that if results for different proc numbers are equal up to a permutation
             # of rows, the sorted frames will compare equal
+            # Include only frames containing at least one non-nan value so pandas knows datatypes.
+            # .all() returns True for empty arrays.
 
             # next(iter(...)) returns the first value in the _other dictionary
             # [0] then picks the first DataFrame from that list
             # columns need to be converted to list() to be passed to sort_values()
             all_columns = list(next(iter(self._other.values()))[0].columns)
             all_res.append(
                 [
-                    pd.concat(others, ignore_index=True).sort_values(by=all_columns, ignore_index=True)
+                    pd.concat(self._drop_empty_dataframes(others), ignore_index=True).sort_values(
+                        by=all_columns, ignore_index=True
+                    )
                     for others in self._other.values()
                 ]
             )
 
         assert all_res, "No test data collected"
-
         for res in all_res:
             assert len(res) == len(self._procs_lst), "Could not collect data for all procs"
 

testsuite/pytests/sli2py_mpi/mpi_test_wrapper.py

@@ -40,4 +40,4 @@ def test_all_to_all(N):
     nest.Connect(nrns, nrns, "all_to_all")
 
     conns = nest.GetConnections().get(output="pandas").drop(labels=["target_thread", "port"], axis=1)
-    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False)  # noqa: F821
+    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False, sep="\t")  # noqa: F821

testsuite/pytests/sli2py_mpi/test_all_to_all.py

@@ -109,4 +109,4 @@ def gids_to_delays(sgids, tgids):
             assert set(actual_weights) <= set(expected_weights)
             assert set(actual_delays) <= set(expected_delays)
 
-    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False)  # noqa: F821
+    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False, sep="\t")  # noqa: F821

testsuite/pytests/sli2py_mpi/test_connect_array_mpi.py

@@ -47,7 +47,7 @@ def test_issue_1957():
     if pre_conns:
         # need to do this here, Disconnect invalidates pre_conns
         df = pd.DataFrame.from_dict(pre_conns.get()).drop(labels="target_thread", axis=1)
-        df.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False)  # noqa: F821
+        df.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False, sep="\t")  # noqa: F821
 
     nest.Disconnect(nrn, nrn)
     nest.Disconnect(nrn, nrn)

testsuite/pytests/sli2py_mpi/test_issue_1957.py

@@ -51,5 +51,5 @@ def test_issue_2119(kind, specs):
     nrn = nest.Create("iaf_psc_alpha", n=4, params={"V_m": nest.CreateParameter(kind, specs)})
 
     pd.DataFrame.from_dict(nrn.get(["global_id", "V_m"])).dropna().to_csv(
-        OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False  # noqa: F821
+        OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False, sep="\t"  # noqa: F821
     )

testsuite/pytests/sli2py_mpi/test_issue_2119.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+#
+# test_rate_neurons_mpi.py
+#
+# This file is part of NEST.
+#
+# Copyright (C) 2004 The NEST Initiative
+#
+# NEST is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# NEST 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with NEST.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+from mpi_test_wrapper import MPITestAssertEqual
+
+
+@pytest.mark.skipif_incompatible_mpi
+@pytest.mark.skipif_missing_threads
+@pytest.mark.skipif_missing_gsl
+@pytest.mark.parametrize(
+    "neuron_model, params1, params2, synspec",
+    [
+        [
+            "lin_rate_ipn",
+            {"mu": 0, "sigma": 0, "rate": 20},
+            {"mu": 0, "sigma": 0},
+            {"synapse_model": "rate_connection_instantaneous", "weight": 5},
+        ],
+        [
+            "siegert_neuron",
+            {"rate": 20},
+            {},
+            {"synapse_model": "diffusion_connection", "diffusion_factor": 2, "drift_factor": 4},
+        ],
+    ],
+)
+@MPITestAssertEqual([1, 2], debug=False)
+def test_rate_neurons_mpi(neuron_model, params1, params2, synspec):
+    """
+    Test that rate neurons are simulated correctly in parallel.
+
+    The test is performed on the multimeter data recorded to MULTI_LABEL during the simulation.
+    """
+
+    import nest
+
+    total_vps = 4
+    h = 0.1
+
+    nest.SetKernelStatus(
+        {
+            "total_num_virtual_procs": total_vps,
+            "resolution": h,
+            "use_wfr": True,
+            "wfr_tol": 0.0001,
+            "wfr_interpolation_order": 3,
+            "wfr_max_iterations": 10,
+            "wfr_comm_interval": 1.0,
+        }
+    )
+
+    neuron1 = nest.Create(neuron_model, params=params1)
+    neuron2 = nest.Create(neuron_model, params=params2)
+    mm = nest.Create(
+        "multimeter",
+        params={
+            "record_from": ["rate"],
+            "interval": 1,
+            "record_to": "ascii",
+            "precision": 8,
+            "time_in_steps": True,
+            "label": MULTI_LABEL.format(nest.num_processes),  # noqa: F821
+        },
+    )
+
+    nest.Connect(mm, neuron1 + neuron2)
+
+    nest.Connect(neuron1, neuron2, syn_spec=synspec)
+
+    nest.Simulate(11)

testsuite/pytests/sli2py_mpi/test_rate_neurons_mpi.py

@@ -40,4 +40,4 @@ def test_get_conns_with_empty_ranks():
     nest.Connect(nrns, nrns)
 
     conns = nest.GetConnections().get(output="pandas").drop(labels=["target_thread", "port"], axis=1, errors="ignore")
-    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False)  # noqa: F821
+    conns.to_csv(OTHER_LABEL.format(nest.num_processes, nest.Rank()), index=False, sep="\t")  # noqa: F821