Skip to content

Context & layout

The proxies importable from interactive_pipe for sharing state and controlling the display — see the Context & events and Layout guides for usage patterns.

get_context 

get_context() -> Dict[str, Any]

Get shared context for passing data between filters.

This returns a simple dictionary for user-defined state only. Framework internals (keys starting with __) are not accessible here.

Returns:

Type Description
Dict[str, Any]

A shared dictionary for user-defined state.

Raises:

Type Description
RuntimeError

If called outside of pipeline execution.
Example 
# In filter A
@interactive(threshold=(0.5, [0.0, 1.0]))
def detect_objects(img, threshold=0.5):
    objects = find_objects(img, threshold)
    ctx = get_context()
    ctx["detected_objects"] = objects
    ctx["detection_count"] = len(objects)
    return img_with_boxes

# In filter B (runs after A)
@interactive()
def analyze_objects(img):
    ctx = get_context()
    objects = ctx.get("detected_objects", [])
    count = ctx.get("detection_count", 0)
    layout.style("analysis", title=f"Found {count} objects")
    return analysis_result
Source code in src/interactive_pipe/core/context.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def get_context() -> Dict[str, Any]:
    """Get shared context for passing data between filters.

    This returns a simple dictionary for user-defined state only.
    Framework internals (keys starting with __) are not accessible here.

    Returns:
        A shared dictionary for user-defined state.

    Raises:
        RuntimeError: If called outside of pipeline execution.

    Example:
        ```python
        # In filter A
        @interactive(threshold=(0.5, [0.0, 1.0]))
        def detect_objects(img, threshold=0.5):
            objects = find_objects(img, threshold)
            ctx = get_context()
            ctx["detected_objects"] = objects
            ctx["detection_count"] = len(objects)
            return img_with_boxes

        # In filter B (runs after A)
        @interactive()
        def analyze_objects(img):
            ctx = get_context()
            objects = ctx.get("detected_objects", [])
            count = ctx.get("detection_count", 0)
            layout.style("analysis", title=f"Found {count} objects")
            return analysis_result
        ```
    """
    ctx = _user_context.get()
    if ctx is None:
        raise RuntimeError(
            "get_context() called outside of pipeline execution. "
            "This function must be called from within a filter executed through an interactive pipeline."
        )
    return ctx

_ContextProxy

Proxy object for direct dict-like access to user context.

This allows using context["key"] or context.key directly instead of ctx = get_context(); ctx["key"].

Example 
from interactive_pipe import context

@interactive()
def my_filter(img):
    # Direct dict-style access
    context["my_data"] = value
    other_data = context.get("other_key", default)

    # Or attribute-style access
    context.my_data = value
    other_data = context.other_key

    return img
Source code in src/interactive_pipe/core/context.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
class _ContextProxy:
    """Proxy object for direct dict-like access to user context.

    This allows using context["key"] or context.key directly instead of
    ctx = get_context(); ctx["key"].

    Example:
        ```python
        from interactive_pipe import context

        @interactive()
        def my_filter(img):
            # Direct dict-style access
            context["my_data"] = value
            other_data = context.get("other_key", default)

            # Or attribute-style access
            context.my_data = value
            other_data = context.other_key

            return img
        ```
    """

    def __getitem__(self, key: str) -> Any:
        """Get item from context using dict-style access."""
        return get_context()[key]

    def __setitem__(self, key: str, value: Any) -> None:
        """Set item in context using dict-style access."""
        get_context()[key] = value

    def __getattr__(self, name: str) -> Any:
        """Get item from context using attribute-style access."""
        try:
            return get_context()[name]
        except KeyError:
            raise AttributeError(f"Context has no attribute '{name}'. Available keys: {list(get_context().keys())}")

    def __setattr__(self, name: str, value: Any) -> None:
        """Set item in context using attribute-style access."""
        get_context()[name] = value

    def __delitem__(self, key: str) -> None:
        """Delete item from context."""
        del get_context()[key]

    def __contains__(self, key: str) -> bool:
        """Check if key exists in context."""
        return key in get_context()

    def get(self, key: str, default: Any = None) -> Any:
        """Get item from context with default value."""
        return get_context().get(key, default)

    def setdefault(self, key: str, default: Any = None) -> Any:
        """Set default value if key doesn't exist."""
        return get_context().setdefault(key, default)

    def pop(self, key: str, *args) -> Any:
        """Pop item from context."""
        return get_context().pop(key, *args)

    def keys(self):
        """Get context keys."""
        return get_context().keys()

    def values(self):
        """Get context values."""
        return get_context().values()

    def items(self):
        """Get context items."""
        return get_context().items()

    def update(self, *args, **kwargs) -> None:
        """Update context with dict or kwargs."""
        get_context().update(*args, **kwargs)

    def clear(self) -> None:
        """Clear context."""
        get_context().clear()

    def __repr__(self) -> str:
        """String representation."""
        try:
            return f"<ContextProxy: {get_context()}>"
        except RuntimeError:
            return "<ContextProxy: not in pipeline execution>"

__getitem__ 

__getitem__(key: str) -> Any

Get item from context using dict-style access.

Source code in src/interactive_pipe/core/context.py 
361
362
363
def __getitem__(self, key: str) -> Any:
    """Get item from context using dict-style access."""
    return get_context()[key]

__setitem__ 

__setitem__(key: str, value: Any) -> None

Set item in context using dict-style access.

Source code in src/interactive_pipe/core/context.py 
365
366
367
def __setitem__(self, key: str, value: Any) -> None:
    """Set item in context using dict-style access."""
    get_context()[key] = value

__getattr__ 

__getattr__(name: str) -> Any

Get item from context using attribute-style access.

Source code in src/interactive_pipe/core/context.py 
369
370
371
372
373
374
def __getattr__(self, name: str) -> Any:
    """Get item from context using attribute-style access."""
    try:
        return get_context()[name]
    except KeyError:
        raise AttributeError(f"Context has no attribute '{name}'. Available keys: {list(get_context().keys())}")

__setattr__ 

__setattr__(name: str, value: Any) -> None

Set item in context using attribute-style access.

Source code in src/interactive_pipe/core/context.py 
376
377
378
def __setattr__(self, name: str, value: Any) -> None:
    """Set item in context using attribute-style access."""
    get_context()[name] = value

__delitem__ 

__delitem__(key: str) -> None

Delete item from context.

Source code in src/interactive_pipe/core/context.py 
380
381
382
def __delitem__(self, key: str) -> None:
    """Delete item from context."""
    del get_context()[key]

__contains__ 

__contains__(key: str) -> bool

Check if key exists in context.

Source code in src/interactive_pipe/core/context.py 
384
385
386
def __contains__(self, key: str) -> bool:
    """Check if key exists in context."""
    return key in get_context()

get 

get(key: str, default: Any = None) -> Any

Get item from context with default value.

Source code in src/interactive_pipe/core/context.py 
388
389
390
def get(self, key: str, default: Any = None) -> Any:
    """Get item from context with default value."""
    return get_context().get(key, default)

setdefault 

setdefault(key: str, default: Any = None) -> Any

Set default value if key doesn't exist.

Source code in src/interactive_pipe/core/context.py 
392
393
394
def setdefault(self, key: str, default: Any = None) -> Any:
    """Set default value if key doesn't exist."""
    return get_context().setdefault(key, default)

pop 

pop(key: str, *args) -> Any

Pop item from context.

Source code in src/interactive_pipe/core/context.py 
396
397
398
def pop(self, key: str, *args) -> Any:
    """Pop item from context."""
    return get_context().pop(key, *args)

keys 

keys()

Get context keys.

Source code in src/interactive_pipe/core/context.py 
400
401
402
def keys(self):
    """Get context keys."""
    return get_context().keys()

values 

values()

Get context values.

Source code in src/interactive_pipe/core/context.py 
404
405
406
def values(self):
    """Get context values."""
    return get_context().values()

items 

items()

Get context items.

Source code in src/interactive_pipe/core/context.py 
408
409
410
def items(self):
    """Get context items."""
    return get_context().items()

update 

update(*args, **kwargs) -> None

Update context with dict or kwargs.

Source code in src/interactive_pipe/core/context.py 
412
413
414
def update(self, *args, **kwargs) -> None:
    """Update context with dict or kwargs."""
    get_context().update(*args, **kwargs)

clear 

clear() -> None

Clear context.

Source code in src/interactive_pipe/core/context.py 
416
417
418
def clear(self) -> None:
    """Clear context."""
    get_context().clear()

__repr__ 

__repr__() -> str

String representation.

Source code in src/interactive_pipe/core/context.py 
420
421
422
423
424
425
def __repr__(self) -> str:
    """String representation."""
    try:
        return f"<ContextProxy: {get_context()}>"
    except RuntimeError:
        return "<ContextProxy: not in pipeline execution>"

_LayoutProxy

Proxy for display and layout operations using contextvars.

This class provides a clean API for controlling output display properties and grid arrangements without polluting function signatures.

Available methods
  • style(): Set display properties (title, colormap, etc.)
  • grid(): Set the output grid layout
  • row(): Convenience for single-row layout
Source code in src/interactive_pipe/core/context.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class _LayoutProxy:
    """Proxy for display and layout operations using contextvars.

    This class provides a clean API for controlling output display properties
    and grid arrangements without polluting function signatures.

    Available methods:
        - style(): Set display properties (title, colormap, etc.)
        - grid(): Set the output grid layout
        - row(): Convenience for single-row layout
    """

    def style(self, name: str, *, title: Optional[str] = None, **style_kwargs: Any) -> None:
        """Set display properties for an output.

        Args:
            name: Output variable name (must match the variable name in the return statement)
            title: Display title for the output
            **style_kwargs: Additional style properties (colormap, vmin, vmax, etc.)

        Example:
            ```python
            layout.style("processed", title=f"Brightness: {brightness:.2f}")
            layout.style("heatmap", title="Heat Map", colormap="viridis", vmin=0, vmax=1)
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()

        style = {}
        if title is not None:
            style["title"] = title
        style.update(style_kwargs)

        state.output_styles[name] = style

    def grid(self, arrangement: Union[List[str], List[List[str]]]) -> None:
        """Set the output grid layout.

        Args:
            arrangement: Either a flat list of output names (treated as a single row)
                        or a 2D list of output names defining the grid arrangement

        Example:
            ```python
            # Single row (flat list)
            layout.grid(["img1", "img2", "img3"])

            # Single row (nested list - equivalent to above)
            layout.grid([["img1", "img2", "img3"]])

            # 2x2 grid
            layout.grid([
                ["original", "processed"],
                ["histogram", "stats"]
            ])
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()
        pipeline = state.pipeline
        if pipeline is not None:
            # Convert flat list to nested list (single row)
            if arrangement and isinstance(arrangement[0], str):
                arrangement = [arrangement]  # type: ignore[list-item]
            pipeline.outputs = arrangement

    def row(self, outputs: List[str]) -> None:
        """Convenience method for single-row layout.

        This is equivalent to calling grid([outputs]).

        Args:
            outputs: List of output names to display in a single row

        Example:
            ```python
            layout.row(["original", "filtered", "result"])
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        self.grid([outputs])

style 

style(name: str, *, title: Optional[str] = None, **style_kwargs: Any) -> None

Set display properties for an output.

Parameters:

Name Type Description Default
name str

Output variable name (must match the variable name in the return statement)

 required
title Optional[str]

Display title for the output

 None
**style_kwargs Any

Additional style properties (colormap, vmin, vmax, etc.)

 {}
Example 
layout.style("processed", title=f"Brightness: {brightness:.2f}")
layout.style("heatmap", title="Heat Map", colormap="viridis", vmin=0, vmax=1)

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def style(self, name: str, *, title: Optional[str] = None, **style_kwargs: Any) -> None:
    """Set display properties for an output.

    Args:
        name: Output variable name (must match the variable name in the return statement)
        title: Display title for the output
        **style_kwargs: Additional style properties (colormap, vmin, vmax, etc.)

    Example:
        ```python
        layout.style("processed", title=f"Brightness: {brightness:.2f}")
        layout.style("heatmap", title="Heat Map", colormap="viridis", vmin=0, vmax=1)
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()

    style = {}
    if title is not None:
        style["title"] = title
    style.update(style_kwargs)

    state.output_styles[name] = style

grid 

grid(arrangement: Union[List[str], List[List[str]]]) -> None

Set the output grid layout.

Parameters:

Name Type Description Default
arrangement Union[List[str], List[List[str]]]

Either a flat list of output names (treated as a single row) or a 2D list of output names defining the grid arrangement

 required
Example 
# Single row (flat list)
layout.grid(["img1", "img2", "img3"])

# Single row (nested list - equivalent to above)
layout.grid([["img1", "img2", "img3"]])

# 2x2 grid
layout.grid([
    ["original", "processed"],
    ["histogram", "stats"]
])

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def grid(self, arrangement: Union[List[str], List[List[str]]]) -> None:
    """Set the output grid layout.

    Args:
        arrangement: Either a flat list of output names (treated as a single row)
                    or a 2D list of output names defining the grid arrangement

    Example:
        ```python
        # Single row (flat list)
        layout.grid(["img1", "img2", "img3"])

        # Single row (nested list - equivalent to above)
        layout.grid([["img1", "img2", "img3"]])

        # 2x2 grid
        layout.grid([
            ["original", "processed"],
            ["histogram", "stats"]
        ])
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()
    pipeline = state.pipeline
    if pipeline is not None:
        # Convert flat list to nested list (single row)
        if arrangement and isinstance(arrangement[0], str):
            arrangement = [arrangement]  # type: ignore[list-item]
        pipeline.outputs = arrangement

row 

row(outputs: List[str]) -> None

Convenience method for single-row layout.

This is equivalent to calling grid([outputs]).

Parameters:

Name Type Description Default
outputs List[str]

List of output names to display in a single row

 required
Example 
layout.row(["original", "filtered", "result"])

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def row(self, outputs: List[str]) -> None:
    """Convenience method for single-row layout.

    This is equivalent to calling grid([outputs]).

    Args:
        outputs: List of output names to display in a single row

    Example:
        ```python
        layout.row(["original", "filtered", "result"])
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    self.grid([outputs])

_AudioProxy

Proxy for audio playback control using contextvars.

This class provides a clean API for controlling audio playback without polluting function signatures with global_params.

Source code in src/interactive_pipe/core/context.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class _AudioProxy:
    """Proxy for audio playback control using contextvars.

    This class provides a clean API for controlling audio playback
    without polluting function signatures with global_params.
    """

    def set(self, audio_path: str) -> None:
        """Set the audio file to play.

        Args:
            audio_path: Path to the audio file

        Example:
            ```python
            audio.set("path/to/track.mp3")
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()
        # AudioBindings defaults to no-ops, so this silently does nothing
        # outside a GUI (same behavior as before)
        state.audio.set_audio(audio_path)

    def play(self) -> None:
        """Start audio playback.

        Example:
            ```python
            audio.play()
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()
        state.audio.play()

    def pause(self) -> None:
        """Pause audio playback.

        Example:
            ```python
            audio.pause()
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()
        state.audio.pause()

    def stop(self) -> None:
        """Stop audio playback.

        Example:
            ```python
            audio.stop()
            ```

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        state = _get_framework_state()
        state.audio.stop()

set 

set(audio_path: str) -> None

Set the audio file to play.

Parameters:

Name Type Description Default
audio_path str

Path to the audio file

 required
Example 
audio.set("path/to/track.mp3")

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def set(self, audio_path: str) -> None:
    """Set the audio file to play.

    Args:
        audio_path: Path to the audio file

    Example:
        ```python
        audio.set("path/to/track.mp3")
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()
    # AudioBindings defaults to no-ops, so this silently does nothing
    # outside a GUI (same behavior as before)
    state.audio.set_audio(audio_path)

play 

play() -> None

Start audio playback.

Example 
audio.play()

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
def play(self) -> None:
    """Start audio playback.

    Example:
        ```python
        audio.play()
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()
    state.audio.play()

pause 

pause() -> None

Pause audio playback.

Example 
audio.pause()

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
def pause(self) -> None:
    """Pause audio playback.

    Example:
        ```python
        audio.pause()
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()
    state.audio.pause()

stop 

stop() -> None

Stop audio playback.

Example 
audio.stop()

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
def stop(self) -> None:
    """Stop audio playback.

    Example:
        ```python
        audio.stop()
        ```

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    state = _get_framework_state()
    state.audio.stop()

_EventsProxy

Read-only proxy for key-bound context events.

GUI backends raise an event flag when a key registered through InteractivePipeGUI.bind_key_to_context is pressed; filters read the flags here. An event stays True only for the single pipeline run triggered by the key press - the GUI resets it right after.

Example 
from interactive_pipe import events, interactive

@interactive()
def denoise(img):
    strength = 0.5 if events.get("boost_denoise") else 0.1
    return smooth(img, strength)

Prefer events.get(name): it returns False for events that were never bound, so the same filter also runs headless or in a GUI without the key binding.

Source code in src/interactive_pipe/core/context.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
class _EventsProxy:
    """Read-only proxy for key-bound context events.

    GUI backends raise an event flag when a key registered through
    InteractivePipeGUI.bind_key_to_context is pressed; filters read the flags
    here. An event stays True only for the single pipeline run triggered by
    the key press - the GUI resets it right after.

    Example:
        ```python
        from interactive_pipe import events, interactive

        @interactive()
        def denoise(img):
            strength = 0.5 if events.get("boost_denoise") else 0.1
            return smooth(img, strength)
        ```

    Prefer events.get(name): it returns False for events that were never
    bound, so the same filter also runs headless or in a GUI without the
    key binding.
    """

    def __getitem__(self, name: str) -> bool:
        """Strict access: raises KeyError if the event was never bound.

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        return _get_framework_state().events[name]

    def get(self, name: str, default: bool = False) -> bool:
        """Return the event flag, or `default` when the event is not bound.

        Raises:
            RuntimeError: If called outside of filter execution context.
        """
        return _get_framework_state().events.get(name, default)

    def __contains__(self, name: str) -> bool:
        """Check whether an event name is bound."""
        return name in _get_framework_state().events

    def keys(self):
        """Names of all bound events."""
        return _get_framework_state().events.keys()

    def __repr__(self) -> str:
        try:
            return f"<EventsProxy: {dict(_get_framework_state().events)}>"
        except RuntimeError:
            return "<EventsProxy: not in filter execution>"

__getitem__ 

__getitem__(name: str) -> bool

Strict access: raises KeyError if the event was never bound.

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
254
255
256
257
258
259
260
def __getitem__(self, name: str) -> bool:
    """Strict access: raises KeyError if the event was never bound.

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    return _get_framework_state().events[name]

get 

get(name: str, default: bool = False) -> bool

Return the event flag, or default when the event is not bound.

Raises:

Type Description
RuntimeError

If called outside of filter execution context.
Source code in src/interactive_pipe/core/context.py 
262
263
264
265
266
267
268
def get(self, name: str, default: bool = False) -> bool:
    """Return the event flag, or `default` when the event is not bound.

    Raises:
        RuntimeError: If called outside of filter execution context.
    """
    return _get_framework_state().events.get(name, default)

__contains__ 

__contains__(name: str) -> bool

Check whether an event name is bound.

Source code in src/interactive_pipe/core/context.py 
270
271
272
def __contains__(self, name: str) -> bool:
    """Check whether an event name is bound."""
    return name in _get_framework_state().events

keys 

keys()

Names of all bound events.

Source code in src/interactive_pipe/core/context.py 
274
275
276
def keys(self):
    """Names of all bound events."""
    return _get_framework_state().events.keys()

Backend

Bases: str, Enum

Supported GUI backends for interactive pipelines.

Source code in src/interactive_pipe/core/backend.py 
 4
 5
 6
 7
 8
 9
10
11
12
class Backend(str, Enum):
    """Supported GUI backends for interactive pipelines."""

    QT = "qt"
    GRADIO = "gradio"
    MPL = "mpl"
    NB = "nb"
    AUTO = "auto"
    HEADLESS = "headless"

FilterError

Bases: Exception

Clean, user-friendly exception for filter errors.

Displays only the relevant error information without the full framework traceback, making it easier to identify issues in user code.

Source code in src/interactive_pipe/core/engine.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class FilterError(Exception):
    """Clean, user-friendly exception for filter errors.

    Displays only the relevant error information without the full
    framework traceback, making it easier to identify issues in user code.
    """

    def __init__(self, filter_name: str, original_error: Exception, tb=None):
        _install_excepthook()
        self.filter_name = filter_name
        self.original_error = original_error
        self.tb = tb
        self._user_frames = self._extract_user_frames()
        super().__init__(self._format_message())

    def _extract_user_frames(self):
        """Extract relevant frames from the traceback (user code, not framework)."""
        if self.tb is None:
            return []

        # Walk through traceback to find the user's code (not in interactive_pipe/src)
        tb_list = traceback.extract_tb(self.tb)
        user_frames = []
        for frame in tb_list:
            # Skip frames from the interactive_pipe framework itself
            if "interactive_pipe" in frame.filename and "/src/" in frame.filename:
                continue
            user_frames.append(frame)

        # If no user frames found, return last frame from traceback
        if not user_frames and tb_list:
            return [tb_list[-1]]
        return user_frames

    def _format_message(self):
        """Format a clean, readable error message."""
        error_str = str(self.original_error)

        lines = [
            "",
            f"  Filter '{self.filter_name}' raised {type(self.original_error).__name__}:",
            f"    {error_str}",
        ]

        if self._user_frames:
            lines.append("")
            lines.append("  Traceback (user code only):")
            for frame in self._user_frames:
                lines.append(f"    {frame.filename}:{frame.lineno} in {frame.name}()")
                if frame.line:
                    lines.append(f"      >>> {frame.line.strip()}")

        lines.append("")
        return "\n".join(lines)

    def print_compact(self, file=None):
        """Print a compact version of the error to stderr or specified file."""
        if file is None:
            file = sys.stderr
        print(f"\n{'=' * 60}", file=file)
        print("PIPELINE ERROR", file=file)
        print(f"{'=' * 60}", file=file)
        print(str(self), file=file)
        print(f"{'=' * 60}\n", file=file)

print_compact 

print_compact(file=None)

Print a compact version of the error to stderr or specified file.

Source code in src/interactive_pipe/core/engine.py 
68
69
70
71
72
73
74
75
76
def print_compact(self, file=None):
    """Print a compact version of the error to stderr or specified file."""
    if file is None:
        file = sys.stderr
    print(f"\n{'=' * 60}", file=file)
    print("PIPELINE ERROR", file=file)
    print(f"{'=' * 60}", file=file)
    print(str(self), file=file)
    print(f"{'=' * 60}\n", file=file)