feat: add .view()

.binder/environment.yml

@@ -9,6 +9,7 @@ dependencies:
 - pyiron_snippets =1.4.0
 - typing-extensions =4.16.0
 - bagofholding =0.1.12
+- ipython
 - ipytree =0.2.2
 - python-workflow-definition =0.1.5
 - numpy =2.4.6

.ci_support/environment-optional.yml

@@ -2,5 +2,6 @@ channels:
 - conda-forge
 dependencies:
 - bagofholding =0.1.12
+- ipython
 - ipytree =0.2.2
 - python-workflow-definition =0.1.5

docs/environment.yml

@@ -15,5 +15,6 @@ dependencies:
 - pyiron_snippets =1.4.0
 - typing-extensions =4.16.0
 - bagofholding =0.1.12
+- ipython
 - ipytree =0.2.2
 - python-workflow-definition =0.1.5

pyproject.toml

@@ -56,6 +56,9 @@ storage-widget = [
     "flowrep[storage]",
     "ipytree==0.2.2",
 ]
+dataviewer = [
+    "ipython",
+]
 
 [tool.hatch.build]
 include = [

src/flowrep/retrospective/datastructures.py

@@ -33,6 +33,7 @@
     while_recipe,
     workflow_recipe,
 )
+from flowrep.retrospective import viewer
 
 
 class NotData(metaclass=singleton.Singleton):
@@ -93,6 +94,16 @@ class NodeData(Generic[RecipeType], abc.ABC):
     @abc.abstractmethod
     def from_recipe(cls, recipe: RecipeType) -> Self: ...
 
+    def view(self, expanded: bool = False):
+        """
+        Display this data object as structured JSON in a notebook and return the display
+        object. Falls back to a plain-text representation when IPython is not available.
+        """
+        return viewer.view(self, expanded=expanded)
+
+    def _repr_json_(self):
+        return self.view()._repr_json_()
+
 
 def recipe2data(
     recipe: union_types.RecipeDiscrimination, allow_variadic_inputs: bool = True

src/flowrep/retrospective/viewer.py

@@ -0,0 +1,77 @@
+"""
+Viewer helpers for retrospective data objects.
+
+Provides a notebook-friendly JSON view and a plain-text fallback.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import inspect
+from collections.abc import Mapping, Sequence
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from flowrep.retrospective.datastructures import NodeData
+
+try:
+    from IPython.display import JSON as _JSON
+
+    _has_ipython = True
+except ImportError:  # pragma: no cover
+    _has_ipython = False
+
+
+def _to_jsonable(data: Any) -> Any:
+    if data is None or isinstance(data, bool | int | float | str):
+        return data
+    if isinstance(data, Mapping):
+        return {
+            _to_jsonable_key(key): _to_jsonable(value) for key, value in data.items()
+        }
+    if isinstance(data, Sequence) and not isinstance(data, str):
+        return [_to_jsonable(item) for item in data]
+    if dataclasses.is_dataclass(data) and not isinstance(data, type):
+        serialized = {
+            field.name: _to_jsonable(getattr(data, field.name))
+            for field in dataclasses.fields(data)
+        }
+        serialized["type"] = data.__class__.__name__
+        return serialized
+    if hasattr(data, "model_dump"):
+        model_dump = data.model_dump(mode="json")
+        if isinstance(model_dump, Mapping):
+            model_dump = {**model_dump, "type": data.__class__.__name__}
+        return _to_jsonable(model_dump)
+    if isinstance(data, type):
+        return f"{data.__module__}.{data.__qualname__}"
+    if inspect.isroutine(data):
+        return f"{data.__module__}.{data.__qualname__}"
+    return repr(data)
+
+
+def _to_jsonable_key(key: Any) -> str:
+    jsonable = _to_jsonable(key)
+    if isinstance(jsonable, str):
+        return jsonable
+    return repr(jsonable)
+
+
+def _view_json(data: NodeData, *, expanded: bool = False):
+    """Display *data* as structured JSON (requires IPython)."""
+    return _JSON(_to_jsonable(data), expanded=expanded)
+
+
+def _view_str(data: NodeData) -> str:
+    """Return a plain-text representation of *data*."""
+    return str(data)
+
+
+def view(data: NodeData, *, expanded: bool = False):
+    """
+    Display *data* as structured JSON when IPython is available, otherwise
+    return a plain-text representation.
+    """
+    if _has_ipython:
+        return _view_json(data, expanded=expanded)
+    return _view_str(data)

tests/unit/test_retrospective_wfms.py

@@ -6,6 +6,7 @@
 import pickle
 import unittest
 from typing import TYPE_CHECKING, NamedTuple, get_origin
+from unittest import mock
 
 from pyiron_snippets import versions
 
@@ -21,14 +22,21 @@
     while_recipe,
     workflow_recipe,
 )
-from flowrep.retrospective import datastructures
+from flowrep.retrospective import datastructures, viewer
 from flowrep.retrospective.datastructures import NOT_DATA
 
 from flowrep_static import library
 
 if TYPE_CHECKING:
     from pyiron_snippets.colors import SeabornColors
 
+try:
+    from IPython.display import JSON as IPythonJSON
+
+    _has_ipython = True
+except ImportError:
+    _has_ipython = False
+
 
 # ═══════════════════════════════════════════════════════════════════════════
 # Unavailable annotations
@@ -841,6 +849,71 @@ def test_propagates_into_workflow_children(self):
         self.assertIsInstance(wf.nodes["splitter_0"], datastructures.AtomicData)
 
 
+@unittest.skipUnless(_has_ipython, "IPython not installed")
+class TestDataView(unittest.TestCase):
+    def test_node_data_view(self):
+        node = datastructures.AtomicData.from_recipe(std.identity.flowrep_recipe)
+        shown = node.view(expanded=True)
+        self.assertIsInstance(shown, IPythonJSON)
+        data = shown.data
+        self.assertEqual(data["type"], "AtomicData")
+        self.assertIn("input_ports", data)
+
+    def test_composite_data_view(self):
+        node = datastructures.DagData.from_recipe(_linear_workflow())
+        shown = node.view()
+        self.assertIsInstance(shown, IPythonJSON)
+        data = shown.data
+        self.assertEqual(data["type"], "DagData")
+        self.assertIn("nodes", data)
+        self.assertIn("add_0", data["nodes"])
+
+    def test_view_expanded_flag(self):
+        node = datastructures.AtomicData.from_recipe(std.identity.flowrep_recipe)
+        shown = node.view(expanded=True)
+        _, metadata = shown._repr_json_()
+        self.assertTrue(metadata["expanded"])
+
+    def test_repr_json(self):
+        node = datastructures.AtomicData.from_recipe(std.identity.flowrep_recipe)
+        repr_json = node._repr_json_()
+        self.assertIsInstance(repr_json, tuple)
+        self.assertEqual(len(repr_json), 2)
+
+    def test_display_json_helper(self):
+        shown = viewer._view_json({"x": 1}, expanded=True)
+        self.assertIsInstance(shown, IPythonJSON)
+        self.assertEqual(shown.data, {"x": 1})
+        _, metadata = shown._repr_json_()
+        self.assertTrue(metadata["expanded"])
+
+    def test_type_branch(self):
+        class MyType: ...
+
+        jsond = viewer._to_jsonable(MyType)
+        self.assertEqual(jsond, f"{MyType.__module__}.{MyType.__qualname__}")
+
+    def test_non_string_key_falls_back_to_repr(self):
+        self.assertEqual(viewer._to_jsonable({42: "v"}), {"42": "v"})
+
+
+class TestViewerStrFallback(unittest.TestCase):
+    def test_view_str_returns_string(self):
+        from flowrep.retrospective import viewer
+
+        node = datastructures.AtomicData.from_recipe(std.identity.flowrep_recipe)
+        result = viewer._view_str(node)
+        self.assertIsInstance(result, str)
+
+    def test_view_without_ipython_falls_back_to_str(self):
+        from flowrep.retrospective import viewer
+
+        node = datastructures.AtomicData.from_recipe(std.identity.flowrep_recipe)
+        with mock.patch.object(viewer, "_has_ipython", False):
+            result = viewer.view(node)
+        self.assertIsInstance(result, str)
+
+
 # ═══════════════════════════════════════════════════════════════════════════
 # wfms.py tests — atomic
 # ═══════════════════════════════════════════════════════════════════════════