Add NotificationSummary class for structured, version-stamped inspection reports

src/nwb2bids/__init__.py

@@ -9,7 +9,7 @@
 from ._converters._dataset_converter import DatasetConverter
 from ._converters._session_converter import SessionConverter
 from ._converters._run_config import RunConfig
-from .notifications import Notification
+from .notifications import Notification, NotificationSummary
 
 __all__ = [
     # Public methods and classes
@@ -18,6 +18,7 @@
     "DatasetConverter",
     "SessionConverter",
     "Notification",
+    "NotificationSummary",
     # Public submodules
     "bids_models",
     "notifications",

src/nwb2bids/_converters/_dataset_converter.py

@@ -12,7 +12,7 @@
 from ._session_converter import SessionConverter
 from .._converters._base_converter import BaseConverter
 from ..bids_models import BidsSessionMetadata, DatasetDescription
-from ..notifications import Notification
+from ..notifications import Notification, NotificationSummary
 
 
 class DatasetConverter(BaseConverter):
@@ -307,9 +307,12 @@ def convert_to_bids_dataset(self) -> None:
         finally:
             self.run_config.bids_directory.mkdir(exist_ok=True)  # Just in case it failed to create earlier
             self.run_config._nwb2bids_directory.mkdir(exist_ok=True)
-
-            notifications_dump = [notification.model_dump(mode="json") for notification in self.notifications]
-            self.run_config.notifications_json_file_path.write_text(data=json.dumps(obj=notifications_dump, indent=2))
+            notification_summary = NotificationSummary(
+                notifications=self.notifications,
+                run_id=self.run_config.run_id,
+            )
+            notification_summary.to_file(path=self.run_config.notifications_json_file_path)
+            notification_summary.to_file(path=self.run_config.notifications_file_path)
 
     def write_bidsignore(self) -> None:
         """Write the `.bidsignore` file if an archive target of `"dandi"` or `"ember"` is specified."""

src/nwb2bids/notifications/__init__.py

@@ -1,11 +1,13 @@
 from ._notification import Notification
+from ._notification_summary import NotificationSummary
 from ._types import Category, Severity, DataStandard
 from ._definitions import notification_definitions
 
 __all__ = [
     "Category",
     "DataStandard",
     "Notification",
+    "NotificationSummary",
     "Severity",
     "notification_definitions",
 ]

src/nwb2bids/notifications/_notification_summary.py

@@ -0,0 +1,224 @@
+import collections
+import importlib.metadata
+import json
+import pathlib
+
+from ._notification import Notification, _CustomJSONEncoder
+
+
+def _get_nwb2bids_version() -> str:
+    try:
+        return importlib.metadata.version("nwb2bids")
+    except importlib.metadata.PackageNotFoundError:
+        return "unknown"
+
+
+def _group_notifications_by_identifier(notifications: list[Notification]) -> dict[str, list[Notification]]:
+    """Group a list of notifications into an ordered dict keyed by identifier (or title)."""
+    groups: dict[str, list[Notification]] = {}
+    for n in notifications:
+        key = n.identifier or n.title
+        if key not in groups:
+            groups[key] = []
+        groups[key].append(n)
+    return groups
+
+
+def _merge_file_paths(group: list[Notification], attribute: str) -> list[str] | None:
+    """Merge source or target file paths from all notifications in a group into a single deduplicated list."""
+    merged = [str(p) for n in group for p in (getattr(n, attribute) or [])]
+    return merged if merged else None
+
+
+class NotificationSummary:
+    """
+    A structured, version-stamped summary of notifications from an nwb2bids inspection run.
+
+    Provides human-readable text formatting, full JSON serialization, and file output
+    for reviewing inspection results.
+
+    Parameters
+    ----------
+    notifications : list of Notification
+        The list of notifications to summarize.
+    run_id : str, optional
+        The run ID associated with this summary. Used for version-stamping the report.
+    """
+
+    def __init__(
+        self,
+        notifications: list[Notification],
+        run_id: str | None = None,
+    ) -> None:
+        self.notifications = list(notifications)
+        self.run_id = run_id
+        self.nwb2bids_version = _get_nwb2bids_version()
+
+    def __repr__(self) -> str:
+        return (
+            f"NotificationSummary("
+            f"notifications={len(self.notifications)}, "
+            f"run_id={self.run_id!r}, "
+            f"nwb2bids_version={self.nwb2bids_version!r}"
+            f")"
+        )
+
+    def __str__(self) -> str:
+        """
+        Human-readable aggregated text representation of the notification summary.
+
+        Notifications of the same type (matched by ``identifier``) are grouped
+        together with an occurrence count and a merged list of file paths, making
+        this output suitable for printing in notebooks and IPython consoles.
+        """
+        return self._to_text(aggregate=True)
+
+    def _to_text(self, aggregate: bool = True) -> str:
+        """Build the text representation of this summary."""
+        lines = []
+        separator = "=" * 72
+        section_separator = "-" * 72
+
+        # Header
+        lines.append(separator)
+        lines.append("nwb2bids Inspection Report")
+        lines.append(f"  nwb2bids version : {self.nwb2bids_version}")
+        if self.run_id is not None:
+            lines.append(f"  Run ID           : {self.run_id}")
+        lines.append(separator)
+
+        if not self.notifications:
+            lines.append("")
+            lines.append("No issues detected.")
+            lines.append(separator)
+            return "\n".join(lines)
+
+        # Summary counts
+        lines.append("")
+        total = len(self.notifications)
+        unique_types = len({n.identifier or n.title for n in self.notifications})
+        lines.append(f"Found {total} notification(s) across {unique_types} unique type(s):")
+        lines.append("")
+
+        # Per-severity breakdown
+        severity_counts = collections.Counter(n.severity.name for n in self.notifications)
+        for severity_name in ("CRITICAL", "ERROR", "WARNING", "HINT", "INFO"):
+            count = severity_counts.get(severity_name, 0)
+            if count > 0:
+                lines.append(f"  {severity_name}: {count}")
+        lines.append("")
+
+        if aggregate:
+            # Group notifications by identifier and write one block per type
+            for group in _group_notifications_by_identifier(self.notifications).values():
+                representative = group[0]
+                count = len(group)
+                occurrence_label = "occurrence" if count == 1 else "occurrences"
+                lines.append(section_separator)
+                lines.append(
+                    f"[{representative.severity.name} | {representative.category.name}] "
+                    f"{representative.title} ({count} {occurrence_label})"
+                )
+                if representative.field is not None:
+                    lines.append(f"  Field    : {representative.field}")
+                if representative.data_standards:
+                    standards = ", ".join(ds.name for ds in representative.data_standards)
+                    lines.append(f"  Standards: {standards}")
+                lines.append(f"  Reason   : {representative.reason}")
+                lines.append(f"  Solution : {representative.solution}")
+                if representative.examples:
+                    lines.append("  Examples :")
+                    for example in representative.examples:
+                        lines.append(f"    {example}")
+
+                all_source_paths = _merge_file_paths(group, "source_file_paths")
+                all_target_paths = _merge_file_paths(group, "target_file_paths")
+                if all_source_paths:
+                    lines.append("  Source files:")
+                    for path in all_source_paths:
+                        lines.append(f"    {path}")
+                if all_target_paths:
+                    lines.append("  Target files:")
+                    for path in all_target_paths:
+                        lines.append(f"    {path}")
+        else:
+            # Write every individual notification as its own block
+            for n in self.notifications:
+                lines.append(section_separator)
+                lines.append(f"[{n.severity.name} | {n.category.name}] {n.title}")
+                if n.field is not None:
+                    lines.append(f"  Field    : {n.field}")
+                if n.data_standards:
+                    standards = ", ".join(ds.name for ds in n.data_standards)
+                    lines.append(f"  Standards: {standards}")
+                lines.append(f"  Reason   : {n.reason}")
+                lines.append(f"  Solution : {n.solution}")
+                if n.examples:
+                    lines.append("  Examples :")
+                    for example in n.examples:
+                        lines.append(f"    {example}")
+                all_source_paths = _merge_file_paths([n], "source_file_paths")
+                all_target_paths = _merge_file_paths([n], "target_file_paths")
+                if all_source_paths:
+                    lines.append("  Source files:")
+                    for path in all_source_paths:
+                        lines.append(f"    {path}")
+                if all_target_paths:
+                    lines.append("  Target files:")
+                    for path in all_target_paths:
+                        lines.append(f"    {path}")
+
+        lines.append(separator)
+        return "\n".join(lines)
+
+    def to_json(self) -> str:
+        """
+        Return a JSON string with the full notification summary.
+
+        All individual notifications are included without aggregation. The returned
+        object contains the ``nwb2bids_version``, ``run_id``, and a ``notifications``
+        list.
+        """
+        data = {
+            "nwb2bids_version": self.nwb2bids_version,
+            "run_id": self.run_id,
+            "notifications": [n.model_dump(mode="json") for n in self.notifications],
+        }
+        return json.dumps(obj=data, indent=2, cls=_CustomJSONEncoder)
+
+    def to_file(self, path: pathlib.Path, aggregate: bool = True) -> None:
+        """
+        Write the notification summary to a file.
+
+        Parameters
+        ----------
+        path : pathlib.Path
+            The path to the output file. When the file suffix is ``.json``, the output
+            is written as structured JSON. Otherwise, the output is written as
+            human-readable text.
+        aggregate : bool, default: True
+            When ``True``, notifications of the same type (matched by ``identifier``)
+            are aggregated into a single entry that includes an occurrence count and a
+            merged list of source and target file paths.  When ``False``, every
+            individual notification is written as its own entry.
+        """
+        if path.suffix == ".json":
+            if not aggregate:
+                path.write_text(data=self.to_json())
+            else:
+                aggregated_notifications = []
+                for group in _group_notifications_by_identifier(self.notifications).values():
+                    entry = group[0].model_dump(mode="json")
+                    entry["count"] = len(group)
+                    entry["source_file_paths"] = _merge_file_paths(group, "source_file_paths")
+                    entry["target_file_paths"] = _merge_file_paths(group, "target_file_paths")
+                    aggregated_notifications.append(entry)
+
+                data = {
+                    "nwb2bids_version": self.nwb2bids_version,
+                    "run_id": self.run_id,
+                    "notifications": aggregated_notifications,
+                }
+                path.write_text(data=json.dumps(obj=data, indent=2, cls=_CustomJSONEncoder))
+        else:
+            path.write_text(data=self._to_text(aggregate=aggregate))

tests/integration/test_convert_nwb_dataset.py

@@ -49,8 +49,9 @@ def test_minimal_convert_nwb_dataset_from_directory(
     assert dataset_converter.run_config.notifications_json_file_path.exists()
     with dataset_converter.run_config.notifications_json_file_path.open(mode="r") as file_stream:
         notifications_json = json.load(fp=file_stream)
-    expected_notification_json: list[dict[str, object]] = []
-    assert notifications_json == expected_notification_json
+    assert "nwb2bids_version" in notifications_json
+    assert notifications_json["run_id"] == run_config.run_id
+    assert notifications_json["notifications"] == []
 
 
 def test_minimal_convert_nwb_dataset_from_file_path(

tests/integration/test_notifications.py

@@ -30,10 +30,14 @@ def test_notifications_1(problematic_nwbfile_path_1: pathlib.Path, temporary_bid
     assert dataset_converter.run_config.notifications_json_file_path.exists()
     with dataset_converter.run_config.notifications_json_file_path.open(mode="r") as file_stream:
         notifications_json = json.load(fp=file_stream)
+    assert "nwb2bids_version" in notifications_json
+    assert "run_id" in notifications_json
+    assert notifications_json["run_id"] == run_config.run_id
     str_nwb_paths = [str(path) for path in nwb_paths]
-    expected_notification_json = [
+    expected_notification_entries = [
         {
             "category": "SCHEMA_INVALIDATION",
+            "count": 1,
             "data_standards": ["DANDI"],
             "examples": None,
             "field": "nwbfile.subject.species",
@@ -48,6 +52,7 @@ def test_notifications_1(problematic_nwbfile_path_1: pathlib.Path, temporary_bid
         },
         {
             "category": "STYLE_SUGGESTION",
+            "count": 1,
             "data_standards": ["BIDS", "DANDI"],
             "examples": ["`ab_01` -> `ab+01`", "`subject #2` -> `subject+2`", "`id 2 from 9/1/25` -> `id+2+9+1+25`"],
             "field": "nwbfile.subject.subject_id",
@@ -64,6 +69,7 @@ def test_notifications_1(problematic_nwbfile_path_1: pathlib.Path, temporary_bid
         },
         {
             "category": "STYLE_SUGGESTION",
+            "count": 1,
             "data_standards": ["BIDS"],
             "examples": ["`male` -> `M`", "`Female` -> `F`", "`n/a` -> `U`", "`hermaphrodite` -> `O`"],
             "field": "nwbfile.subject.sex",
@@ -77,6 +83,7 @@ def test_notifications_1(problematic_nwbfile_path_1: pathlib.Path, temporary_bid
         },
         {
             "category": "STYLE_SUGGESTION",
+            "count": 1,
             "data_standards": ["DANDI"],
             "examples": ["`male` -> `M`", "`Female` -> `F`", "`n/a` -> `U`", "`hermaphrodite` -> `O`"],
             "field": "nwbfile.subject.sex",
@@ -89,7 +96,19 @@ def test_notifications_1(problematic_nwbfile_path_1: pathlib.Path, temporary_bid
             "title": "Invalid participant sex (archives)",
         },
     ]
-    assert notifications_json == expected_notification_json
+    assert notifications_json["notifications"] == expected_notification_entries
+
+    assert dataset_converter.run_config.notifications_file_path.exists()
+    notifications_text = dataset_converter.run_config.notifications_file_path.read_text()
+    assert "nwb2bids Inspection Report" in notifications_text
+    assert run_config.run_id in notifications_text
+    assert "Invalid species" in notifications_text
+    assert "Invalid participant ID" in notifications_text
+    # Verify each notification type appears exactly once (aggregated) in the text output
+    assert notifications_text.count("Invalid species") == 1
+    assert notifications_text.count("Invalid participant ID") == 1
+    # Each has 1 occurrence
+    assert "1 occurrence" in notifications_text
 
 
 def test_notifications_2(problematic_nwbfile_path_2: pathlib.Path, temporary_bids_directory: pathlib.Path) -> None: