Implement SONATA condition modifications

.gitignore

@@ -30,4 +30,5 @@ nrnivmodl.log
 cADpyr_L2TPC_dendrogram.png
 ex2_sonatagraph.pdf
 .venv
-bad_dendritic_fit.png
+bad_dendritic_fit.png
+/build

bluecellulab/circuit/config/bluepy_simulation_config.py

@@ -204,6 +204,10 @@ def add_connection_override(
     ) -> None:
         self._connection_overrides.append(connection_override)
 
+    def get_modifications(self) -> list:
+        """Bluepy configs do not support modifications."""
+        return []
+
     def get_compartment_sets(self) -> dict[str, dict[str, Any]]:
         """Bluepy configs do not support compartment_sets."""
         raise NotImplementedError("Compartment sets are only supported for SonataSimulationConfig.")

bluecellulab/circuit/config/definition.py

@@ -17,7 +17,11 @@
 from typing import Optional, Protocol
 
 
-from bluecellulab.circuit.config.sections import Conditions, ConnectionOverrides
+from bluecellulab.circuit.config.sections import (
+    Conditions,
+    ConnectionOverrides,
+    ModificationBase,
+)
 from bluecellulab.stimulus.circuit_stimulus_definitions import Stimulus
 
 
@@ -36,6 +40,9 @@ def condition_parameters(self) -> Conditions:
     def connection_entries(self) -> list[ConnectionOverrides]:
         raise NotImplementedError
 
+    def get_modifications(self) -> list[ModificationBase]:
+        raise NotImplementedError
+
     def get_compartment_sets(self) -> dict[str, dict]:
         """Return SONATA-style compartment_sets mapping."""
         raise NotImplementedError
@@ -108,8 +115,5 @@ def output_root_path(self) -> str:
     def extracellular_calcium(self) -> Optional[float]:
         raise NotImplementedError
 
-    def add_connection_override(
-        self,
-        connection_override: ConnectionOverrides
-    ) -> None:
+    def add_connection_override(self, connection_override: ConnectionOverrides) -> None:
         raise NotImplementedError

bluecellulab/circuit/config/sections.py

@@ -1,4 +1,5 @@
 # Copyright 2023-2024 Blue Brain Project / EPFL
+# Copyright 2025-2026 Open Brain Institute
 
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,7 +15,7 @@
 """Classes to represent config sections."""
 
 from __future__ import annotations
-from typing import Literal, Optional
+from typing import Any, Literal, Optional
 
 from pydantic import field_validator, Field
 from pydantic.dataclasses import dataclass
@@ -29,6 +30,7 @@
     from libsonata._libsonata import Conditions as LibSonataConditions
 except ImportError:
     from libsonata._libsonata import SimulationConfig
+
     LibSonataConditions = SimulationConfig.Conditions
 
 
@@ -44,13 +46,15 @@ def string_to_bool(value: str) -> bool:
 @dataclass(frozen=True, config=dict(extra="forbid"))
 class ConditionEntry:
     """For mechanism specific conditions."""
+
     minis_single_vesicle: Optional[int] = Field(None, ge=0, le=1)
     init_depleted: Optional[int] = Field(None, ge=0, le=1)
 
 
 @dataclass(frozen=True, config=dict(extra="forbid"))
 class MechanismConditions:
     """For mechanism specific conditions."""
+
     ampanmda: Optional[ConditionEntry] = None
     gabaab: Optional[ConditionEntry] = None
     glusynapse: Optional[ConditionEntry] = None
@@ -59,6 +63,7 @@ class MechanismConditions:
 @dataclass(frozen=True, config=dict(extra="forbid"))
 class Conditions:
     mech_conditions: Optional[MechanismConditions] = None
+    mechanisms: Optional[dict[str, dict[str, Any]]] = None
     celsius: Optional[float] = None
     v_init: Optional[float] = None
     extracellular_calcium: Optional[float] = None
@@ -80,6 +85,7 @@ def from_blueconfig(cls, condition_entries: dict) -> Conditions:
         )
         return cls(
             mech_conditions=mech_conditions,
+            mechanisms=None,
             extracellular_calcium=condition_entries.get("cao_CR_GluSynapse", None),
             randomize_gaba_rise_time=randomize_gaba_risetime,
         )
@@ -111,8 +117,14 @@ def from_sonata(cls, condition_entries: LibSonataConditions) -> Conditions:
             glusynapse=ConditionEntry(msv_glusynapse, init_dep_glusynapse),
         )
 
+        # Store the full generic mechanisms dict from libsonata
+        generic_mechanisms = None
+        if mech_dict is not None:
+            generic_mechanisms = dict(mech_dict)
+
         return cls(
             mech_conditions=mech_conditions,
+            mechanisms=generic_mechanisms,
             celsius=condition_entries.celsius,
             v_init=condition_entries.v_init,
             extracellular_calcium=condition_entries.extracellular_calcium,
@@ -125,13 +137,102 @@ def init_empty(cls) -> Conditions:
         specified."""
         return cls(
             mech_conditions=None,
+            mechanisms=None,
             celsius=None,
             v_init=None,
             extracellular_calcium=None,
             randomize_gaba_rise_time=None,
         )
 
 
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationBase:
+    """Base class for all modification types."""
+
+    name: str
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationNodeSet(ModificationBase):
+    """Modification that targets a node_set."""
+
+    node_set: str
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationTTX(ModificationNodeSet):
+    """TTX modification — blocks Na channels on all sections of target
+    cells."""
+
+    type: Literal["ttx"] = "ttx"
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationConfigureAllSections(ModificationNodeSet):
+    """Applies section_configure to all sections of target cells."""
+
+    section_configure: str
+    type: Literal["configure_all_sections"] = "configure_all_sections"
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationSectionList(ModificationNodeSet):
+    """Applies section_configure to a named section list of target cells."""
+
+    section_configure: str
+    type: Literal["section_list"] = "section_list"
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationSection(ModificationNodeSet):
+    """Applies section_configure to specific named sections of target cells."""
+
+    section_configure: str
+    type: Literal["section"] = "section"
+
+
+@dataclass(frozen=True, config=dict(extra="forbid"))
+class ModificationCompartmentSet(ModificationBase):
+    """Applies section_configure to segments in a compartment set."""
+
+    compartment_set: str
+    section_configure: str
+    type: Literal["compartment_set"] = "compartment_set"
+
+
+def modification_from_libsonata(mod) -> ModificationBase:
+    """Convert a libsonata modification object to a BlueCelluLab dataclass."""
+    type_name = mod.type.name.lower()  # e.g. "ttx", "configure_all_sections", etc.
+    if type_name == "ttx":
+        return ModificationTTX(name=mod.name, node_set=mod.node_set)
+    elif type_name == "configure_all_sections":
+        return ModificationConfigureAllSections(
+            name=mod.name,
+            node_set=mod.node_set,
+            section_configure=mod.section_configure,
+        )
+    elif type_name == "section_list":
+        return ModificationSectionList(
+            name=mod.name,
+            node_set=mod.node_set,
+            section_configure=mod.section_configure,
+        )
+    elif type_name == "section":
+        return ModificationSection(
+            name=mod.name,
+            node_set=mod.node_set,
+            section_configure=mod.section_configure,
+        )
+    elif type_name == "compartment_set":
+        return ModificationCompartmentSet(
+            name=mod.name,
+            compartment_set=mod.compartment_set,
+            section_configure=mod.section_configure,
+        )
+    else:
+        raise ValueError(f"Unknown modification type: {type_name}")
+
+
 @dataclass(frozen=True, config=dict(extra="forbid"))
 class ConnectionOverrides:
     source: str
@@ -148,8 +249,7 @@ class ConnectionOverrides:
     def validate_mod_override(cls, value):
         """Make sure the mod file to override is present."""
         if isinstance(value, str) and not hasattr(neuron.h, value):
-            raise bluecellulab.ConfigError(
-                f"Mod file for {value} is not found.")
+            raise bluecellulab.ConfigError(f"Mod file for {value} is not found.")
         return value
 
     @classmethod

bluecellulab/circuit/config/sonata_simulation_config.py

@@ -19,7 +19,12 @@
 from typing import Optional
 import warnings
 
-from bluecellulab.circuit.config.sections import Conditions, ConnectionOverrides
+from bluecellulab.circuit.config.sections import (
+    Conditions,
+    ConnectionOverrides,
+    ModificationBase,
+    modification_from_libsonata,
+)
 from bluecellulab.stimulus.circuit_stimulus_definitions import Stimulus
 
 from bluepysnap import Simulation as SnapSimulation
@@ -66,33 +71,49 @@ def get_all_stimuli_entries(self) -> list[Stimulus]:
         for value in inputs.values():
             # Validate mutual exclusivity and existence of compartment_set
             if "compartment_set" in value and "node_set" in value:
-                raise ValueError("Stimulus entry must not include both 'node_set' and 'compartment_set'.")
+                raise ValueError(
+                    "Stimulus entry must not include both 'node_set' and 'compartment_set'."
+                )
 
             if "compartment_set" in value:
                 if compartment_sets is None:
-                    raise ValueError("SONATA simulation config references 'compartment_set' in inputs but no 'compartment_sets_file' is configured.")
+                    raise ValueError(
+                        "SONATA simulation config references 'compartment_set' in inputs but no 'compartment_sets_file' is configured."
+                    )
                 comp_name = value["compartment_set"]
                 if comp_name not in compartment_sets:
-                    raise ValueError(f"Compartment set '{comp_name}' not found in compartment_sets file.")
+                    raise ValueError(
+                        f"Compartment set '{comp_name}' not found in compartment_sets file."
+                    )
                 # Validate the list: must be list of triples, sorted and unique by (node_id, sec_ref, seg)
                 comp_entry = compartment_sets[comp_name]
                 comp_nodes = comp_entry.get("compartment_set")
                 if comp_nodes is None:
-                    raise ValueError(f"Compartment set '{comp_name}' does not contain 'compartment_set' key.")
+                    raise ValueError(
+                        f"Compartment set '{comp_name}' does not contain 'compartment_set' key."
+                    )
                 # Validate duplicates and sorted order
                 try:
                     last = None
                     for trip in comp_nodes:
                         if not (isinstance(trip, list) and len(trip) >= 3):
-                            raise ValueError(f"Invalid compartment_set entry '{trip}' in '{comp_name}'; expected [node_id, section, seg].")
+                            raise ValueError(
+                                f"Invalid compartment_set entry '{trip}' in '{comp_name}'; expected [node_id, section, seg]."
+                            )
                         key = (trip[0], trip[1], trip[2])
                         if last is not None and key < last:
-                            raise ValueError(f"Compartment list for '{comp_name}' must be sorted ascending.")
+                            raise ValueError(
+                                f"Compartment list for '{comp_name}' must be sorted ascending."
+                            )
                         if last == key:
-                            raise ValueError(f"Compartment list for '{comp_name}' contains duplicate entry {key}.")
+                            raise ValueError(
+                                f"Compartment list for '{comp_name}' contains duplicate entry {key}."
+                            )
                         last = key
                 except TypeError:
-                    raise ValueError(f"Compartment list for '{comp_name}' contains non-comparable entries.")
+                    raise ValueError(
+                        f"Compartment list for '{comp_name}' contains non-comparable entries."
+                    )
 
             stimulus = Stimulus.from_sonata(value, config_dir=config_dir)
             if stimulus:
@@ -105,6 +126,12 @@ def condition_parameters(self) -> Conditions:
         condition_object = self.impl.conditions
         return Conditions.from_sonata(condition_object)
 
+    @lru_cache(maxsize=1)
+    def get_modifications(self) -> list[ModificationBase]:
+        """Returns the list of modifications from the conditions block."""
+        mods = self.impl.conditions.modifications()
+        return [modification_from_libsonata(m) for m in mods]
+
     @lru_cache(maxsize=1)
     def _connection_entries(self) -> list[ConnectionOverrides]:
         result: list[ConnectionOverrides] = []
@@ -126,7 +153,7 @@ def get_compartment_sets(self) -> dict[str, dict]:
         full_path = Path(filepath)
         if config_dir is not None and not full_path.is_absolute():
             full_path = Path(config_dir) / filepath
-        with open(full_path, 'r') as f:
+        with open(full_path, "r") as f:
             return json.load(f)
 
     @lru_cache(maxsize=1)
@@ -145,7 +172,9 @@ def get_node_sets(self) -> dict[str, dict]:
             base_node_sets.update(sim_node_sets)
 
         if not base_node_sets:
-            raise ValueError("No 'node_sets_file' found in simulation or circuit config.")
+            raise ValueError(
+                "No 'node_sets_file' found in simulation or circuit config."
+            )
 
         return base_node_sets
 
@@ -156,6 +185,8 @@ def get_report_entries(self) -> dict[str, dict]:
         Each key is a report name, and the value is its configuration.
         """
         reports = self.impl.config.get("reports", {})
+        if reports is None:
+            return {}
         if not isinstance(reports, dict):
             raise ValueError("Invalid format for 'reports' in SONATA config.")
         return reports
@@ -215,8 +246,7 @@ def tstop(self) -> float:
     @property
     def duration(self) -> Optional[float]:
         warnings.warn(
-            "`duration` is deprecated. Use `tstop` instead.",
-            DeprecationWarning
+            "`duration` is deprecated. Use `tstop` instead.", DeprecationWarning
         )
         return self.tstop
 
@@ -253,10 +283,7 @@ def spikes_file_path(self) -> Path:
     def extracellular_calcium(self) -> Optional[float]:
         return self.condition_parameters().extracellular_calcium
 
-    def add_connection_override(
-        self,
-        connection_override: ConnectionOverrides
-    ) -> None:
+    def add_connection_override(self, connection_override: ConnectionOverrides) -> None:
         self._connection_overrides.append(connection_override)
 
     def _get_config_dir(self):