Merge branch 'main' into katta/fix_ngv_no_connection_cell

Author: cattabiani

neurodamus/commands.py

@@ -67,6 +67,7 @@ def neurodamus(args=None):
         --num-target-ranks=<number>  Number of ranks to target for dry-run load balancing
         --coreneuron-direct-mode     Run CoreNeuron in direct memory mode transfered from Neuron,
                                      without writing model data to disk.
+        --use-color=[ON, OFF]  If OFF, forces no color to be used in logs; [default: ON]
     """
     from . import __version__
 

neurodamus/core/_neurodamus.py

@@ -28,19 +28,19 @@ def h(self):
         return self._h
 
     @classmethod
-    def _init(cls, **kwargs):
+    def _init(cls, log_filename=LOG_FILENAME, log_use_color=True):
         if cls._pc is not None:
             return
         # Neurodamus requires MPI. We still respect NEURON_INIT_MPI though
         _Neuron._init(int(os.environ.get("NEURON_INIT_MPI", "1")))  # if needed, sets cls._h
 
         # Init logging
-        log_name = kwargs.get("log_filename") or LOG_FILENAME
+        log_filename = log_filename or LOG_FILENAME
         if MPI.rank == 0:
-            open(log_name, "w", encoding="utf-8").close()  # Truncate
+            open(log_filename, "w", encoding="utf-8").close()  # Truncate
         MPI.barrier()  # Sync so that all processes see the file
-        setup_logging(GlobalConfig.verbosity, log_name, MPI.rank)
-        log_stage("Initializing Neurodamus... Logfile: " + log_name)
+        setup_logging(GlobalConfig.verbosity, log_filename, MPI.rank, use_color=log_use_color)
+        log_stage("Initializing Neurodamus... Logfile: " + log_filename)
 
         # Load mods if not available
         cls._load_nrnmechlibs()

neurodamus/node.py

@@ -1,5 +1,7 @@
 # Neurodamus
 # Copyright 2018 - Blue Brain Project, EPFL
+from __future__ import annotations
+
 import gc
 import glob
 import itertools
@@ -331,12 +333,13 @@ class Node:
     _default_population = "All"
     """The default population name for e.g. Reports."""
 
-    def __init__(self, config_file, options=None):
+    def __init__(self, config_file, options: dict | None = None):
         """Creates a neurodamus executor
         Args:
             config_file: A Sonata config file
             options: A dictionary of run options typically coming from cmd line
         """
+        options = options or {}
         assert isinstance(config_file, str), "`config_file` should be a string"
         assert config_file, "`config_file` cannot be empty"
 
@@ -347,7 +350,7 @@ def __init__(self, config_file, options=None):
         import libsonata
 
         conf = libsonata.SimulationConfig.from_file(config_file)
-        Nd.init(log_filename=conf.output.log_file)
+        Nd.init(log_filename=conf.output.log_file, log_use_color=options.pop("use_color", True))
 
         # This is global initialization, happening once, regardless of number of
         # cycles

neurodamus/utils/logging.py

@@ -88,7 +88,7 @@ def _format_msg(self, record, style):
         )
 
 
-def setup_logging(loglevel, logfile=None, rank=None):
+def setup_logging(loglevel, logfile=None, rank=None, use_color=True):
     """Setup neurodamus logging.
     Features tabs and colors output to stdout and pydamus.log
 
@@ -109,17 +109,17 @@ def setup_logging(loglevel, logfile=None, rank=None):
         _logging.DEBUG,
     ]
 
-    # Stdout
+    # stdout
     hdlr = _logging.StreamHandler(sys.stdout)
-    use_color = True
-    if os.environ.get("ENVIRONMENT") == "BATCH":
-        use_color = False
-    else:
-        try:
-            sys.stdout.tell()  # works only if it's file
+    if use_color:
+        if os.environ.get("ENVIRONMENT") == "BATCH":
             use_color = False
-        except OSError:
-            pass
+        else:
+            try:
+                sys.stdout.tell()  # works only if it's file
+                use_color = False
+            except OSError:
+                pass
     hdlr.setFormatter(_LevelColorFormatter(with_time=False, rank=rank, use_color=use_color))
     if rank == 0:
         _logging.root.setLevel(verbosity_levels[loglevel])

pyproject.toml

@@ -153,6 +153,9 @@ combine-as-imports = true
     "PLR2004", # magic value (constant) used in comparison (i.e. expected == 3)
     "S101",    # Use of `assert` detected
     "SLF001",  # private member access
+    "S404",    # `subprocess` module is possibly insecure
+    "S603",    # `subprocess` call: check for execution of untrusted input 
+    "S607",    # Starting a process with a partial executable path
 ]
 
 [tool.ruff.lint.pydocstyle]

tests/conftest.py

@@ -100,7 +100,7 @@ def change_test_dir(monkeypatch, tmp_path):
     monkeypatch.chdir(tmp_path)
 
 
-@pytest.fixture()
+@pytest.fixture
 def copy_memory_files(change_test_dir):
     # Fix values to ensure allocation memory (0,0)[1, 3] (1,0)[2]
     metypes_memory = {

tests/integration-e2e/test_cli_opts.py

@@ -1,167 +1,41 @@
 import json
-import os
 import subprocess
-import tempfile
 from pathlib import Path
 
+import libsonata
+
 SIM_DIR = Path(__file__).parent.parent.absolute() / "simulations" / "v5_sonata"
 CONFIG_FILE_MINI = "simulation_config_mini.json"
 CIRCUIT_DIR = "sub_mini5"
 
 
-def test_cli_prcellgid():
-    from neurodamus import Neurodamus
-    test_folder = tempfile.TemporaryDirectory("cli-test-prcellgid")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-        with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-            json.dump(sim_config_data, f, indent=2)
-
-    os.chdir(test_folder_path)
-    nd = Neurodamus(CONFIG_FILE_MINI, dump_cell_state=1, keep_build=True
-                    )
-    nd.run()
-    assert (test_folder_path / "output_sonata2" / "2_py_Neuron_t0.0.nrndat").is_file()
-    assert (test_folder_path / "output_sonata2"  / "2_py_Neuron_t100.0.nrndat").is_file()
-
+def test_cli_disable_reports(tmp_path):
+    with open(SIM_DIR / CONFIG_FILE_MINI, encoding="utf-8") as fd:
+        sim_config_data = json.load(fd)
 
-def test_cli_disable_reports():
-    test_folder = tempfile.TemporaryDirectory("cli-test-disable-reports")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-        with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-            json.dump(sim_config_data, f, indent=2)
+    sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
+    with open(tmp_path / CONFIG_FILE_MINI, "w", encoding="utf-8") as fd:
+        json.dump(sim_config_data, fd)
 
     subprocess.run(
         ["neurodamus", CONFIG_FILE_MINI, "--disable-reports"],
         check=True,
         capture_output=True,
-        cwd=test_folder_path
-    )
-    # Spikes are present even if we disable reports
-    assert (test_folder_path / sim_config_data["output"]["output_dir"] / "out.h5").is_file()
-    for report in sim_config_data["reports"].keys():
-        report_path = test_folder_path / sim_config_data["output"]["output_dir"] / (report + ".h5")
-        assert not report_path.is_file(), f"File '{report_path}' should NOT exist."
-
-    subprocess.run(
-        ["neurodamus", CONFIG_FILE_MINI],
-        check=True,
-        capture_output=True,
-        cwd=test_folder_path
+        cwd=tmp_path
     )
-    assert (test_folder_path / sim_config_data["output"]["output_dir"] / "out.h5").is_file()
-    for report in sim_config_data["reports"].keys():
-        report_path = test_folder_path / sim_config_data["output"]["output_dir"] / (report + ".h5")
-        assert report_path.is_file(), f"File '{report_path}' not found."
-
-
-def test_cli_keep_build():
-    from neurodamus import Neurodamus
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["target_simulator"] = "CORENEURON"
-        sim_config_data["output"]["output_dir"] = "output_keep_build"
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-
-    test_folder = tempfile.TemporaryDirectory("cli-test-keep-build")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-        json.dump(sim_config_data, f, indent=2)
-
-    os.chdir(test_folder_path)
-    nd = Neurodamus(CONFIG_FILE_MINI, keep_build=True, disable_reports=True)
-    nd.run()
-    coreneuron_input_dir = test_folder_path / "build" / "coreneuron_input"
-    assert coreneuron_input_dir.is_dir(), "Directory 'coreneuron_input' not found."
 
+    sc = libsonata.SimulationConfig.from_file(CONFIG_FILE_MINI)
+    spikes_path = tmp_path / sc.output.output_dir / "out.h5"
 
-def test_cli_build_model():
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["target_simulator"] = "CORENEURON"
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-
-    test_folder = tempfile.TemporaryDirectory("cli-test-build-model")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-        json.dump(sim_config_data, f, indent=2)
-
-    result_model = subprocess.run(
-        ["neurodamus", CONFIG_FILE_MINI, "--simulate-model=OFF", "--disable-reports"],
-        check=True,
-        cwd=test_folder_path,
-        capture_output=True,
-        text=True
-    )
-    assert "[SKIPPED] SIMULATION (MODEL BUILD ONLY)" in result_model.stdout
-
-    result_auto = subprocess.run(
-        ["neurodamus", CONFIG_FILE_MINI, "--disable-reports"],
-        check=True,
-        cwd=test_folder_path,
-        capture_output=True,
-        text=True
-    )
-    assert "SIMULATION (SKIP MODEL BUILD)" in result_auto.stdout
-
-    subprocess.run(
-        ["neurodamus", CONFIG_FILE_MINI, "--simulate-model=OFF", "--disable-reports"],
-        check=True,
-        cwd=test_folder_path
-    )
-    result_off = subprocess.run(
-        ["neurodamus", CONFIG_FILE_MINI, "--build-model=OFF", "--disable-reports"],
-        check=True,
-        cwd=test_folder_path,
-        capture_output=True,
-        text=True
-    )
-    assert "SIMULATION (SKIP MODEL BUILD)" in result_off.stdout
-
-
-def test_cli_lb_mode():
-    test_folder = tempfile.TemporaryDirectory("cli-test-lb-mode")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-        with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-            json.dump(sim_config_data, f, indent=2)
-
-    for lb_mode in ("WholeCell", "MultiSplit"):
-        result = subprocess.run(
-            ["neurodamus", CONFIG_FILE_MINI, f"--lb-mode={lb_mode}", "--disable-reports"],
-            check=True,
-            cwd=test_folder_path,
-            capture_output=True,
-            text=True
-        )
-        assert f"Load Balancing ENABLED. Mode: {lb_mode}" in result.stdout
-        assert (test_folder_path / "mcomplex.dat").is_file(), "File 'mcomplex.dat' not found."
-        assert (test_folder_path / "sim_conf").is_dir(), "Directory 'sim_conf' not found."
-
+    # Spikes are present even if we disable reports
+    assert spikes_path.is_file()
 
-def test_cli_output_path():
-    from neurodamus import Neurodamus
-    test_folder = tempfile.TemporaryDirectory("cli-test-output-path")  # auto removed
-    test_folder_path = Path(test_folder.name)
-    with open(SIM_DIR / CONFIG_FILE_MINI, "r") as f:
-        sim_config_data = json.load(f)
-        sim_config_data["network"] = str(SIM_DIR / CIRCUIT_DIR / "circuit_config.json")
-        with open(test_folder_path / CONFIG_FILE_MINI, "w") as f:
-            json.dump(sim_config_data, f, indent=2)
+    for name in sc.list_report_names:
+        report_path = Path(sc.report(name).file_name)
+        assert not report_path.is_file(), f"File '{report_path}' should NOT exist."
 
-    simconfig_output_path = sim_config_data["output"]["output_dir"]
-    output_path = "new_output"
-    os.chdir(test_folder_path)
-    nd = Neurodamus(CONFIG_FILE_MINI, output_path=output_path)
-    nd.run()
-    # Output directory from simulation configuration is overridden
-    assert not (test_folder_path / simconfig_output_path).is_dir(), \
-           f"Directory '{simconfig_output_path}' should NOT exist."
-    assert (test_folder_path / output_path).is_dir(), f"Directory '{output_path}' not found."
+    subprocess.run(["neurodamus", CONFIG_FILE_MINI], cwd=tmp_path, check=True)
+    assert spikes_path.is_file()
+    for name in sc.list_report_names:
+        report_path = Path(sc.report(name).file_name)
+        assert report_path.is_file(), f"File '{report_path}' not found."