Pipes#

Rendering lifecycle recovery#

Widget state can arrive out of order, and an overloaded Jupyter server can drop a state update. The browser therefore reports a stale pipe or viewer instead of waiting indefinitely; the kernel re-emits the relevant widget state with bounded exponential backoff. This is a recovery handshake, not a rendering timeout: it stops as soon as the browser has a renderable source.

        sequenceDiagram
    participant K as Kernel
    participant B as Browser
    K->>B: widget state / run request
    alt inlet, value, or outlet missing
        B->>K: stale (which state is missing)
        K->>B: re-emit endpoint and pipe/viewer state
    else state is renderable
        B->>B: render layout and selection
    end

class ipyelk.pipes.Pipe(*args: t.Any, **kwargs: t.Any)#

A step in the processing pipeline for diagrams.

Attributes#

inlet: MarkElementWidget: input elements to manipulate.
outlet: MarkElementWidget: output elements that potentially have been manipulated.
observed: tuple of str: which potential changes that would require this pipe to be rerun.
reports: tuple of str: types of changes that get added to the output based of rerunning this pipe.
on_progress: callable: Callable function that is executed when the pipe is running.
status: PipeStatus: Captures the disposition of the pipe during the change lifecycle.
status_view: PipeStatusView: Widget to show pipe status as it updates.
enabled: bool: whether the processing step can be run

cancel()#

Cancel the runner and drop its pending requests; True if one was alive.

The only thing that cancels a run. Use it when the pipe is detached, so an in-flight answer is not persisted into an index nobody views. A task on another loop is cancelled through that loop (see util.settle).

Return type:: bool

check_dirty(flow=None)#

Method to test is this pipe should be run given the set of changes.

Parameters:: flow (tuple[str, ...] | None) – the changes to test against; defaults to the pending inlet.flow
Returns:: dirty flag
Return type:: bool

close()#

Cancel and close this pipe and the widgets it created.

An inlet or outlet it was given, such as the source it was connected to, is left open.

error()#

Method to raise any potential errors captured during the running of the pipe.

Raises:: self.status.exception – captured exception during the processing.

async run()#

Run method that takes the input performs checks/changes, and sets the output value.

Subclasses of the pipe will implement their own custom processing logic.

schedule_run(change=None)#

Request a run and return the task that will serve it.

Requests coalesce on the trailing edge: one runner task serves every request made while it is alive, so ten calls in one tick cost one run and a call made mid-run costs one more run after it. Nothing is cancelled here (see cancel); a Pipeline gives way at its next stage boundary (Superseded).

A runner on another event loop (a kernel subshell, see util.settle) is handed over to a new runner on the caller’s loop, so the returned task is always awaitable by the caller. Pending requests survive the hand-over.

Returns None (and schedules nothing) when no event loop is running, e.g. when a diagram is built in a plain script or a test: there is no loop to run the task on, so raising would only crash widget construction (Diagram(source=...) refreshes from a trait observer). The request still counts and is served by the next runner.

Return type:: Task | None

superseded()#

Whether a newer request arrived while the runner’s current run is in flight (a Pipeline checks this between stages).

Return type:: bool

class ipyelk.pipes.Pipeline(*args: t.Any, **kwargs: t.Any)#

cancel()#

Cancel, and re-record the in-flight run’s tags now: the cancellation unwinds a loop turn later, after a new runner may already have taken.

Return type:: bool

check()#

Checks inlets and outlets of the pipeline and raises error is not connected

Raises:: BrokenPipe – Disconnected pipes
Return type:: bool
Returns:: True if no errors in pipeline

check_dirty(flow=None)#

Method to test is this pipe should be run given the set of changes.

Parameters:: flow (tuple[str, ...] | None) – the changes to test against; defaults to the pending inlet.flow
Returns:: dirty flag
Return type:: bool

close()#: Close this pipeline, its sub-pipes and the widgets it created.

async run(flow=None)#

Run the dirty pipes for the pending flow.

The flow is taken from the inlet at the start (MarkElementWidget.take) and re-recorded unless the run succeeds. A source should therefore feed one pipeline: a second would find the flow already taken.

flow is passed by a parent pipeline to a nested one, since the parent already took it.

class ipyelk.pipes.base.PipeStatus(disposition=PipeDisposition.done, elapsed=None, exception=None)#

An immutable snapshot of a pipe’s disposition.

A plain value, not a widget, so replacing Pipe.status opens no comm. Instances compare by identity: assigning a new one always notifies status observers.

class ipyelk.pipes.base.PipeStatusView(*args: t.Any, **kwargs: t.Any)#

Widget to display the pipe status.

Attributes#

include_exception: bool: exception captured
html: ipywidgets.HTML: built status HTML to display

close()#

Close method.

Closes the underlying comm. When the comm is closed, all of the widget views are automatically removed from the front-end.

update(pipe)#: Method to update the status given changes in the pipe.

class ipyelk.pipes.MarkElementWidget(*args: t.Any, **kwargs: t.Any)#

A synced element tree plus the shared index and the pending flow.

value is written from both sides. A browser write is never sent back: echo_update=False stops the echo, and _should_send_property stops the re-serialized copy. Kernel writes are sent exactly once.

Known limitation: a second frontend attached to the same kernel used to learn browser-written values through the echo and no longer does.

build_index(*, assign_ids=True)#

Index value; by default pin the resulting ids onto the elements.

The index’s Registry assigns ids to elements that have none; writing them back is what lets every later serialization – which runs outside the Registry – emit the ids the index is keyed by, so persist can merge what comes back from the browser. Explicit ids are never changed. assign_ids=False leaves id untouched so a caller can report or reject unassigned ids first (ValidationPipe.fix_null_id).

Return type:: MarkIndex

gen#: generation of the run request the browser last answered; 0 with an older extension build (see util.wait_for_answer)

persist(rebuild_index=False)#

Fold value into the shared index.

The index – not value – is the authority for the element hierarchy: hidden elements never survive serialization (see Node.model_dump), so the index must be merged into, never rebuilt from, a value that has been through the browser. rebuild_index is for the full Python-side hierarchy only (initially, or after adding elements); the merge path requires every element to already carry an id.

record(*tags)#

Add tags to the pending flow (order-preserving union).

Writers must record rather than assign, which would overwrite what another writer left pending.

Return type:: tuple[str, ...]

set_trait(name, value)#: Remember which value came from the browser (set_state holds _property_lock), for _should_send_property.

take()#

Consume the pending flow: return it and leave () behind.

Pipeline.run takes at the start, so tags recorded mid-run stay pending for the next run.

Return type:: tuple[str, ...]

class ipyelk.pipes.ElkJS(*args: t.Any, **kwargs: t.Any)#

JupyterLab widget for calling elkjs layout given a valid elkjson dictionary

async run()#

Run method that takes the input performs checks/changes, and sets the output value.

Subclasses of the pipe will implement their own custom processing logic.

timeout#: Seconds to wait for the browser to return a layout before giving up; 0 waits forever (the request is re-sent with backoff until a frontend answers)