diff --git a/src/nwb2bids/__init__.py b/src/nwb2bids/__init__.py index c8521f33..7de8b938 100644 --- a/src/nwb2bids/__init__.py +++ b/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", diff --git a/src/nwb2bids/_converters/_dataset_converter.py b/src/nwb2bids/_converters/_dataset_converter.py index 414e4dd3..2c94fbe5 100644 --- a/src/nwb2bids/_converters/_dataset_converter.py +++ b/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.""" diff --git a/src/nwb2bids/notifications/__init__.py b/src/nwb2bids/notifications/__init__.py index 7c807dab..42148a58 100644 --- a/src/nwb2bids/notifications/__init__.py +++ b/src/nwb2bids/notifications/__init__.py @@ -1,4 +1,5 @@ from ._notification import Notification +from ._notification_summary import NotificationSummary from ._types import Category, Severity, DataStandard from ._definitions import notification_definitions @@ -6,6 +7,7 @@ "Category", "DataStandard", "Notification", + "NotificationSummary", "Severity", "notification_definitions", ] diff --git a/src/nwb2bids/notifications/_notification_summary.py b/src/nwb2bids/notifications/_notification_summary.py new file mode 100644 index 00000000..6eb28033 --- /dev/null +++ b/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)) diff --git a/tests/integration/test_convert_nwb_dataset.py b/tests/integration/test_convert_nwb_dataset.py index a6d95e14..95b6a0dd 100644 --- a/tests/integration/test_convert_nwb_dataset.py +++ b/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( diff --git a/tests/integration/test_notifications.py b/tests/integration/test_notifications.py index e239e466..0a1557a2 100644 --- a/tests/integration/test_notifications.py +++ b/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: diff --git a/tests/unit/test_notification_summary.py b/tests/unit/test_notification_summary.py new file mode 100644 index 00000000..eab88df5 --- /dev/null +++ b/tests/unit/test_notification_summary.py @@ -0,0 +1,186 @@ +"""Unit tests for the NotificationSummary class.""" + +import json +import pathlib + +import pytest + +import nwb2bids +from nwb2bids.notifications import Notification, NotificationSummary + + +@pytest.fixture() +def sample_notifications() -> list[Notification]: + return [ + Notification.from_definition(identifier="InvalidSpecies"), + Notification.from_definition(identifier="InvalidParticipantID"), + Notification.from_definition(identifier="InvalidParticipantSexBIDS"), + ] + + +@pytest.fixture() +def notifications_with_paths(tmp_path: pathlib.Path) -> tuple[list[Notification], list[pathlib.Path]]: + nwb_path = tmp_path / "test.nwb" + nwb_path.touch() + nwb_paths = [nwb_path] + notifications = [ + Notification.from_definition(identifier="InvalidSpecies", source_file_paths=nwb_paths), + Notification.from_definition(identifier="InvalidSpecies", source_file_paths=nwb_paths), + Notification.from_definition(identifier="InvalidParticipantID", source_file_paths=nwb_paths), + ] + return notifications, nwb_paths + + +def test_notification_summary_initialization(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications, run_id="test-run-id") + assert summary.notifications == sample_notifications + assert summary.run_id == "test-run-id" + assert isinstance(summary.nwb2bids_version, str) + + +def test_notification_summary_initialization_no_run_id(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications) + assert summary.run_id is None + + +def test_notification_summary_empty() -> None: + summary = NotificationSummary(notifications=[]) + text = str(summary) + assert "nwb2bids Inspection Report" in text + assert "No issues detected." in text + + +def test_notification_summary_str_contains_header(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications, run_id="test-run-id") + text = str(summary) + assert "nwb2bids Inspection Report" in text + assert "test-run-id" in text + assert summary.nwb2bids_version in text + + +def test_notification_summary_str_aggregates_by_default( + notifications_with_paths: tuple[list[Notification], list[pathlib.Path]], +) -> None: + notifications, _ = notifications_with_paths + summary = NotificationSummary(notifications=notifications) + text = str(summary) + # "InvalidSpecies" appears twice in notifications but should appear as one block with "2 occurrences" + assert "2 occurrences" in text + # InvalidParticipantID appears once + assert "1 occurrence" in text + + +def test_notification_summary_str_contains_notification_details(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications) + text = str(summary) + assert "Invalid species" in text + assert "Invalid participant ID" in text + assert "Invalid participant sex (BIDS)" in text + # Check severity labels are present + assert "ERROR" in text + + +def test_notification_summary_repr(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications, run_id="test-run-id") + repr_text = repr(summary) + assert "NotificationSummary" in repr_text + assert "notifications=3" in repr_text + assert "test-run-id" in repr_text + + +def test_notification_summary_to_json_structure(sample_notifications: list[Notification]) -> None: + summary = NotificationSummary(notifications=sample_notifications, run_id="test-run-id") + json_str = summary.to_json() + data = json.loads(json_str) + + assert "nwb2bids_version" in data + assert data["nwb2bids_version"] == summary.nwb2bids_version + assert "run_id" in data + assert data["run_id"] == "test-run-id" + assert "notifications" in data + assert len(data["notifications"]) == 3 + + +def test_notification_summary_to_json_full_not_aggregated( + notifications_with_paths: tuple[list[Notification], list[pathlib.Path]], +) -> None: + notifications, _ = notifications_with_paths + summary = NotificationSummary(notifications=notifications) + json_str = summary.to_json() + data = json.loads(json_str) + # to_json always returns all individual notifications (no aggregation) + assert len(data["notifications"]) == 3 + + +def test_notification_summary_to_file_json_aggregated( + tmp_path: pathlib.Path, + notifications_with_paths: tuple[list[Notification], list[pathlib.Path]], +) -> None: + notifications, _ = notifications_with_paths + summary = NotificationSummary(notifications=notifications, run_id="test-run") + out_path = tmp_path / "report.json" + summary.to_file(path=out_path, aggregate=True) + + assert out_path.exists() + data = json.loads(out_path.read_text()) + assert "nwb2bids_version" in data + assert data["run_id"] == "test-run" + # Two unique identifiers: InvalidSpecies (count=2) and InvalidParticipantID (count=1) + assert len(data["notifications"]) == 2 + species_entry = next(n for n in data["notifications"] if n["identifier"] == "InvalidSpecies") + assert species_entry["count"] == 2 + + +def test_notification_summary_to_file_json_not_aggregated( + tmp_path: pathlib.Path, + notifications_with_paths: tuple[list[Notification], list[pathlib.Path]], +) -> None: + notifications, _ = notifications_with_paths + summary = NotificationSummary(notifications=notifications, run_id="test-run") + out_path = tmp_path / "report.json" + summary.to_file(path=out_path, aggregate=False) + + assert out_path.exists() + data = json.loads(out_path.read_text()) + # With aggregate=False, all 3 individual notifications are included + assert len(data["notifications"]) == 3 + + +def test_notification_summary_to_file_txt( + tmp_path: pathlib.Path, + sample_notifications: list[Notification], +) -> None: + summary = NotificationSummary(notifications=sample_notifications, run_id="test-run") + out_path = tmp_path / "report.txt" + summary.to_file(path=out_path) + + assert out_path.exists() + text = out_path.read_text() + assert "nwb2bids Inspection Report" in text + assert "Invalid species" in text + + +def test_notification_summary_to_file_txt_not_aggregated( + tmp_path: pathlib.Path, + notifications_with_paths: tuple[list[Notification], list[pathlib.Path]], +) -> None: + notifications, _ = notifications_with_paths + summary = NotificationSummary(notifications=notifications) + out_path = tmp_path / "report.txt" + summary.to_file(path=out_path, aggregate=False) + + assert out_path.exists() + text = out_path.read_text() + assert "nwb2bids Inspection Report" in text + # With aggregate=False all 3 entries are separate blocks + assert text.count("Invalid species") == 2 + + +def test_notification_summary_exported_at_top_level() -> None: + assert hasattr(nwb2bids, "NotificationSummary") + assert nwb2bids.NotificationSummary is NotificationSummary + + +def test_notification_summary_exported_in_notifications_submodule() -> None: + assert hasattr(nwb2bids.notifications, "NotificationSummary") + assert nwb2bids.notifications.NotificationSummary is NotificationSummary