Skip to content

Serve

How a workbench starts: the femforge bridge, the serve entry points, the command line, the shell, the launcher, the demo and the dev reload loop.

femforge.workbench

The workbench API from the femforge-workbench package, which the workbench extra installs.

Without the extra, importing this module, and so reading femforge.workbench, raises an ImportError that names the install command.

Classes:

  • FieldControlsPanel –

    The viewer's controls: the field, mode and colormap selects, the display switches, the colour range, the body tree and the field-line options.

  • FieldPlot2DPanel –

    The plotly view of a 2D run: every body a Mesh3d with hover values, its legend toggling bodies in step with the controls.

  • FieldViewerPanel –

    The 3D view of the field scene, mirrored in the browser by default; its visibility feeds the controller.

  • FigurePanel –

    Shows one plotly figure; update swaps it (uirevision keeps brushes and cameras), bind routes its events to handlers.

  • Panel –

    One tab: its title, the builder that makes its component once the shell exists, and its share of the axis.

  • PanelComponent –

    A dock panel: its template is built once inside layout; state variables go through key.

  • RunBrowserPanel –

    The run index as a virtual table: text, status, tag and physics filters narrow it in polars, a row click selects the run.

  • RunDetailPanel –

    One run bundle: its manifest summary, stage timeline, results, artifact inputs, files and retention.

  • Split –

    Panels, tab groups or splits sharing one axis by weight; the axis alternates with nesting, horizontal at the root.

  • TablePanel –

    A read-only table whose rows go over once, the client sorting, filtering, faceting, reordering, hiding and grouping them; lead orders the leading columns and pins the first.

  • Tabs –

    Panels sharing one tab group, the first shown.

  • Window –

    Where a served workbench opens.

  • Workbench –

    One trame server with the header, the dock and the panels of a named workbench.

Functions:

  • artifact_store –

    The store the artifact panels read: FEMFORGE_ARTIFACTS when set, else the root the index rows name.

  • field_viewer –

    The field viewer dock: the controls beside the 3D view and the 2D plot as tabs.

  • show –

    Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

FieldControlsPanel

FieldControlsPanel(app: Workbench, id: str, title: str, controller: FieldViewerController)

Bases: PanelComponent

The viewer's controls: the field, mode and colormap selects, the display switches, the colour range, the body tree and the field-line options.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • controller (FieldViewerController) –

    The viewer's controller, shared with the 3D view and the 2D plot.

Methods:

  • build_run_picker –

    The indexed-run select and the bundle-path field, both driving the same run variable; built into the app bar, its one place.

  • opening –

    The bundle-path field's Enter: the text is committed and opened in one client-side step.

  • build_colormap –

    The field colormap picker: searchable, filtered by type and favorites, the arrow keys stepping through it while it is closed.

  • build_body_colormap –

    The body colormap picker as a swatch in the color column, reading Same as field while no body takes it.

  • slider –

    A slider over model whose thumbs move with the drag and whose scene follows the release.

  • build_range –

    The Auto range switch over the two-thumb colour range across the data range, its ends read out above it.

  • build_opacity –

    The Opacity mapping switch and, while it is on, the opacity at each end of the color range; the body table holds the overall opacity.

  • build_field –

    The field section: what is shown and how it is colored, from the field select to the opacity mapping.

  • build_bodies –

    The body table: the All bodies row with the body colormap, then a row per body and per material group, every row checkbox, name, color and opacity.

  • build_lines –

    The field-line section: the switch, the seeds, the region and the two integration sliders.

  • build –

    Builds the busy bar over the field, field-line and body sections.

  • dispose –

    Detaches from the controller, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def __init__(self, app: Workbench, id: str, title: str, controller: FieldViewerController) -> None:
    """Builds the controls and attaches them to the controller.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        controller: The viewer's controller, shared with the 3D view and the 2D plot.
    """
    self.controller = controller

    super().__init__(app, id, title)

    controller.attach(self)

build_run_picker

build_run_picker() -> None

The indexed-run select and the bundle-path field, both driving the same run variable; built into the app bar, its one place.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_run_picker(self) -> None:
    """The indexed-run select and the bundle-path field, both driving the same run variable; built into the app bar, its one place."""
    with labelled("Run"):
        picker(str(FieldVar.RUN), str(FieldVar.RUN_ITEMS), placeholder="No Run Selected", width=RUN_WIDTH)

    Input(
        model=str(FieldVar.RUN_TEXT),
        placeholder="Bundle Path",
        classes=PATH_WIDTH,
        aria_label="Bundle path",
        on_key_down=self.opening(),
    )
    with Button(size="icon-sm", variant="ghost", title="Open the bundle at that path", on_click=self.callback(self.controller.open_text)):
        Icon("folder")

opening

opening() -> Callback

The bundle-path field's Enter: the text is committed and opened in one client-side step.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def opening(self) -> react.Callback:
    """The bundle-path field's Enter: the text is committed and opened in one client-side step."""
    text = f"{EVENT}.target.value"

    return react.Callback(f'{EVENT}.key === "{ENTER}" && (trame.state.set("{FieldVar.RUN_TEXT}", {text}), trame.state.set("{FieldVar.RUN}", {text}))')

build_colormap

build_colormap() -> None

The field colormap picker: searchable, filtered by type and favorites, the arrow keys stepping through it while it is closed.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_colormap(self) -> None:
    """The field colormap picker: searchable, filtered by type and favorites, the arrow keys stepping through it while it is closed."""
    Combobox(
        model=str(FieldVar.COLORMAP),
        items=react.Bind(str(FieldVar.COLORMAP_ITEMS)),
        filters=COLORMAP_FILTERS,
        search_placeholder="Search Colormaps",
        aria_label="Colormap",
    )

build_body_colormap

build_body_colormap() -> None

The body colormap picker as a swatch in the color column, reading Same as field while no body takes it.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_body_colormap(self) -> None:
    """The body colormap picker as a swatch in the color column, reading Same as field while no body takes it."""
    Combobox(
        value=react.Bind(str(FieldVar.BODY_COLORMAP_SHOWN)),
        on_value_change=self.callback(self.controller.on_body_colormap_pick, args=f"[{EVENT}]"),
        items=react.Bind(str(FieldVar.COLORMAP_ITEMS)),
        lead=react.Bind(str(FieldVar.BODY_LEAD)),
        filters=COLORMAP_FILTERS,
        compact=True,
        search_placeholder="Search Colormaps",
        aria_label="Body colormap",
    )

slider

slider(model: FieldVar, label: str, bounds: tuple[float, float | Bind], step: float) -> None

A slider over model whose thumbs move with the drag and whose scene follows the release.

Parameters:

  • model (FieldVar) –

    The state variable holding the thumb values.

  • label (str) –

    The slider's accessible name.

  • bounds (tuple[float, float | Bind]) –

    The low and high ends; the high end may be a client expression.

  • step (float) –

    The step between values.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def slider(self, model: FieldVar, label: str, bounds: tuple[float, float | react.Bind], step: float) -> None:
    """A slider over `model` whose thumbs move with the drag and whose scene follows the release.

    Args:
        model: The state variable holding the thumb values.
        label: The slider's accessible name.
        bounds: The low and high ends; the high end may be a client expression.
        step: The step between values.
    """
    Slider(
        value=react.Bind(str(model)),
        on_value_change=setter(str(model)),
        on_value_commit=self.callback(self.controller.on_commit),
        min_=bounds[0],
        max_=bounds[1],
        step=step,
        aria_label=label,
    )

build_range

build_range() -> None

The Auto range switch over the two-thumb colour range across the data range, its ends read out above it.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_range(self) -> None:
    """The Auto range switch over the two-thumb colour range across the data range, its ends read out above it."""
    setting(FieldVar.AUTO_CLIM, "Auto range")
    with html.Div(classes=STACK):
        readouts(react.Bind(f"{FieldVar.CLIM}[0].toPrecision({PRECISION})"), react.Bind(f"{FieldVar.CLIM}[1].toPrecision({PRECISION})"))
        Slider(
            value=react.Bind(str(FieldVar.CLIM)),
            on_value_change=setter(str(FieldVar.CLIM)),
            on_value_commit=self.callback(self.controller.on_commit),
            min_=react.Bind(str(FieldVar.CLIM_MIN)),
            max_=react.Bind(str(FieldVar.CLIM_MAX)),
            step=react.Bind(str(FieldVar.CLIM_STEP)),
            disabled=react.Bind(str(FieldVar.AUTO_CLIM)),
            aria_label="Color range",
        )

build_opacity

build_opacity() -> None

The Opacity mapping switch and, while it is on, the opacity at each end of the color range; the body table holds the overall opacity.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_opacity(self) -> None:
    """The Opacity mapping switch and, while it is on, the opacity at each end of the color range; the body table holds the overall opacity."""
    setting(FieldVar.OPACITY_MAPPING, "Opacity mapping")
    with react.If(str(FieldVar.OPACITY_MAPPING)), html.Div(classes=STACK):
        readouts(react.Bind(f"{FieldVar.OPACITY_RANGE}[0].toFixed(2)"), react.Bind(f"{FieldVar.OPACITY_RANGE}[1].toFixed(2)"))
        self.slider(FieldVar.OPACITY_RANGE, "Opacity", OPACITY_RANGE, OPACITY_STEP)

build_field

build_field() -> None

The field section: what is shown and how it is colored, from the field select to the opacity mapping.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_field(self) -> None:
    """The field section: what is shown and how it is colored, from the field select to the opacity mapping."""
    with captioned("Quantity"):
        picker(str(FieldVar.FIELD), str(FieldVar.FIELDS), placeholder=NO_RUN, width=FULL_WIDTH)

    with captioned("Mode"):
        picker(str(FieldVar.MODE), str(FieldVar.MODES), placeholder=NO_RUN, width=FULL_WIDTH)

    with captioned("Colormap"):
        self.build_colormap()

    with html.Div(classes=STACK):
        setting(FieldVar.INVERT, "Invert colormap")
        setting(FieldVar.LOG_SCALE, "Log scale")
        setting(FieldVar.OUTLINE, "Outline")

    self.build_range()
    self.build_opacity()

build_bodies

build_bodies() -> None

The body table: the All bodies row with the body colormap, then a row per body and per material group, every row checkbox, name, color and opacity.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_bodies(self) -> None:
    """The body table: the All bodies row with the body colormap, then a row per body and per material group, every row checkbox, name, color and opacity."""
    controller = self.controller
    visible = str(FieldVar.VISIBLE_BODIES)
    row = ITEM
    members = f"{row}.{RowKey.MEMBERS}"

    def opacity_slider(value: str, change: react.Callback, commit: react.Callback, label: str | react.Bind) -> None:
        Slider(
            value=react.Bind(value),
            on_value_change=change,
            on_value_commit=commit,
            min_=OPACITY_RANGE[0],
            max_=OPACITY_RANGE[1],
            step=OPACITY_STEP,
            aria_label=label,
        )

    with html.Div(classes=BODY_TABLE):
        with html.Div(classes=BODY_ROW):
            Checkbox(
                checked=react.Bind(f"{visible}.length === 0 ? false : {visible}.length === {FieldVar.BODY_ROWS}.filter(r => !r.{RowKey.GROUP}).length || 'indeterminate'"),
                on_checked_change=react.Callback(f"trame.state.set({json.dumps(visible)}, {EVENT} === true ? {FieldVar.BODY_ROWS}.filter(r => !r.{RowKey.GROUP}).map(r => r.{RowKey.ID}) : [])"),
                aria_label="Show every body",
            )
            html.Span(ALL_BODIES, classes=GROUP_NAME)
            self.build_body_colormap()
            opacity_slider(str(FieldVar.OPACITY), setter(str(FieldVar.OPACITY)), self.callback(controller.on_commit), "Opacity of every body")

        with react.For(items=str(FieldVar.BODY_ROWS), name=row), html.Div(classes=BODY_ROW):
            Checkbox(
                checked=react.Bind(f"{members}.every(m => {visible}.includes(m)) || ({members}.some(m => {visible}.includes(m)) && 'indeterminate')"),
                on_checked_change=self.callback(controller.on_row_visible, args=f"[{row}.{RowKey.ID}, {EVENT}]"),
                aria_label=react.Bind(f"'Show ' + {row}.{RowKey.TITLE}"),
            )
            html.Span(
                [react.Bind(f"{row}.{RowKey.TITLE}")],
                classes=react.Bind(f"{row}.{RowKey.GROUP} ? {json.dumps(GROUP_NAME)} : {row}.{RowKey.MEMBER} ? {json.dumps(MEMBER_NAME)} : {json.dumps(BODY_NAME)}"),
                title=react.Bind(f"{row}.{RowKey.TITLE}"),
            )
            with html.Span(classes=CHIP_SLOT):
                painted = f"{row}.{RowKey.PAINT} !== {json.dumps(str(BodyPaint.FIELD))}"
                with react.If(painted):
                    ColorInput(
                        value=react.Bind(f"{row}.{RowKey.COLOR}"),
                        on_change=self.callback(controller.on_row_color, args=f"[{row}.{RowKey.ID}, {EVENT}.target.value]"),
                        classes=COLOR_CHIP,
                        title=react.Bind(f"{row}.{RowKey.PAINT} === {json.dumps(str(BodyPaint.COLORMAP))} ? 'From the body colormap: pick to fix a color' : 'A fixed color'"),
                        aria_label=react.Bind(f"{row}.{RowKey.TITLE} + ' color'"),
                    )
                    with Button(
                        size="icon-xs",
                        variant="ghost",
                        classes=CHIP_CLEAR,
                        title="Back to the field colormap",
                        aria_label=react.Bind(f"{row}.{RowKey.TITLE} + ' back to the field colormap'"),
                        on_click=self.callback(controller.on_row_field, args=f"[{row}.{RowKey.ID}]"),
                    ):
                        Icon("x")

                with react.If(f"!({painted})"):
                    Button(
                        size="icon-xs",
                        variant="ghost",
                        classes=CHIP,
                        style={"background": react.Bind(str(FieldVar.FIELD_GRADIENT))},
                        title="Field colormap: click for the body colormap",
                        aria_label=react.Bind(f"{row}.{RowKey.TITLE} + ' to the body colormap'"),
                        on_click=self.callback(controller.on_row_colormap, args=f"[{row}.{RowKey.ID}]"),
                    )

            opacity_slider(
                f"{row}.{RowKey.OPACITY}",
                self.callback(controller.on_row_opacity, args=f"[{row}.{RowKey.ID}, {EVENT}[0]]"),
                self.callback(controller.on_commit),
                react.Bind(f"{row}.{RowKey.TITLE} + ' opacity'"),
            )

build_lines

build_lines() -> None

The field-line section: the switch, the seeds, the region and the two integration sliders.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build_lines(self) -> None:
    """The field-line section: the switch, the seeds, the region and the two integration sliders."""
    setting(FieldVar.STREAMLINES, "Show field lines", disabled=react.Bind(f"!{FieldVar.HAS_FLUX}"))

    with captioned("Seeds"):
        picker(str(FieldVar.SEED_PLACEMENT), str(FieldVar.SEED_PLACEMENTS), width=FULL_WIDTH)

    with captioned("Integrate within"):
        picker(str(FieldVar.LINES_BODY), str(FieldVar.LINES_BODIES), placeholder=WHOLE_DOMAIN_LABEL, width=FULL_WIDTH)

    with html.Div(classes=STACK):
        readouts("Seed count", react.Bind(seed_count()))
        self.slider(FieldVar.LINE_COUNT, "Seeds per axis", (LINE_COUNT_MIN, react.Bind(str(FieldVar.LINE_COUNT_MAX))), 1)

    with html.Div(classes=STACK):
        readouts("Length", react.Bind(f"{FieldVar.MAX_LENGTH}[0].toPrecision(2)"))
        self.slider(FieldVar.MAX_LENGTH, "Line length", MAX_LENGTH_RANGE, MAX_LENGTH_STEP)

build

build() -> None

Builds the busy bar over the field, field-line and body sections.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def build(self) -> None:
    """Builds the busy bar over the field, field-line and body sections."""

    def section(value: str, heading: str, body: Callable[[], None]) -> None:
        with AccordionItem(value=value, classes=str(Cls.SECTION_ITEM)):
            AccordionTrigger(heading, classes=str(Cls.SECTION_TRIGGER))
            with AccordionContent(classes=str(Cls.SECTION_BODY)):
                body()

    with html.Div(classes=str(Cls.CONTROLS)):
        with react.If(str(FieldVar.BUSY)):
            Progress(value=BUSY_FULL, classes=str(Cls.BUSY))

        with Accordion(type_="multiple", default_value=OPEN_SECTIONS, classes=str(Cls.SECTIONS)):
            section(FIELD_SECTION, "Field", self.build_field)
            section(LINES_SECTION, "Field Lines", self.build_lines)
            section(BODIES_SECTION, "Bodies", self.build_bodies)

dispose

dispose() -> None

Detaches from the controller, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def dispose(self) -> None:
    """Detaches from the controller, then releases the panel."""
    self.controller.detach(self)
    super().dispose()

FieldPlot2DPanel

FieldPlot2DPanel(app: Workbench, id: str, title: str, controller: FieldViewerController)

Bases: FigurePanel

The plotly view of a 2D run: every body a Mesh3d with hover values, its legend toggling bodies in step with the controls.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • controller (FieldViewerController) –

    The viewer's controller, shared with the controls and the 3D view.

Methods:

  • show –

    Replaces the figure, remembering which body each mesh trace is.

  • clear –

    Empties the figure.

  • on_visible –

    Tells the controller whether the plot is on screen.

  • dispose –

    Detaches from the controller, then releases the panel.

  • on_restyle –

    Mirrors a legend toggle into the visible-bodies state, so the tree and the 3D view follow.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def __init__(self, app: Workbench, id: str, title: str, controller: FieldViewerController) -> None:
    """Builds the 2D plot and attaches it to the controller.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        controller: The viewer's controller, shared with the controls and the 3D view.
    """
    self.controller = controller
    self.bodies: list[str] = []

    super().__init__(app, id, title)

    self.bind(RESTYLE_EVENT, self.on_restyle)
    controller.plot = self
    controller.attach(self)

show

show(figure: Figure, bodies: Sequence[str]) -> None

Replaces the figure, remembering which body each mesh trace is.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def show(self, figure: go.Figure, bodies: Sequence[str]) -> None:
    """Replaces the figure, remembering which body each mesh trace is."""
    self.bodies = list(bodies)
    self.update(figure)

clear

clear() -> None

Empties the figure.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def clear(self) -> None:
    """Empties the figure."""
    self.bodies = []
    self.update(go.Figure())

on_visible

on_visible(visible: bool) -> None

Tells the controller whether the plot is on screen.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def on_visible(self, visible: bool) -> None:
    """Tells the controller whether the plot is on screen."""
    self.controller.on_visible(self.id, visible)

dispose

dispose() -> None

Detaches from the controller, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def dispose(self) -> None:
    """Detaches from the controller, then releases the panel."""
    self.controller.detach(self)
    super().dispose()

on_restyle

on_restyle(payload: object) -> None

Mirrors a legend toggle into the visible-bodies state, so the tree and the 3D view follow.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def on_restyle(self, payload: object) -> None:
    """Mirrors a legend toggle into the visible-bodies state, so the tree and the 3D view follow."""
    parts = cast("list[object]", payload) if isinstance(payload, list) else []
    if len(parts) != 2 or not isinstance(parts[0], Mapping) or not isinstance(parts[1], list):
        return

    flags = cast("Mapping[str, object]", parts[0]).get("visible")
    indices = cast("list[object]", parts[1])
    if not isinstance(flags, list):
        return

    shown = set(names_of(self.state[FieldVar.VISIBLE_BODIES]))
    for index, flag in zip(indices, cast("list[object]", flags), strict=False):
        if not isinstance(index, int) or not 0 <= index < len(self.bodies):
            continue

        name = self.bodies[index]
        if flag is True:
            shown.add(name)
        else:
            shown.discard(name)

    with self.state:
        self.state[FieldVar.VISIBLE_BODIES] = [name for name in self.bodies if name in shown]

FieldViewerPanel

FieldViewerPanel(app: Workbench, id: str, title: str, controller: FieldViewerController)

Bases: ScenePanel

The 3D view of the field scene, mirrored in the browser by default; its visibility feeds the controller.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • controller (FieldViewerController) –

    The viewer's controller, shared with the controls and the 2D plot.

Methods:

  • clear –

    Empties the plotter and pushes the empty frame.

  • on_visible –

    Tells the scene and the controller whether the view is on screen.

  • dispose –

    Detaches from the controller, then frees the view.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def __init__(self, app: Workbench, id: str, title: str, controller: FieldViewerController) -> None:
    """Builds the 3D view and attaches it to the controller.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        controller: The viewer's controller, shared with the controls and the 2D plot.
    """
    self.controller = controller

    super().__init__(app, id, title)

    controller.viewer = self
    controller.drawn = None
    controller.framed = None
    controller.attach(self)
    if controller.scene is None:
        self.notice(NO_RUN)

clear

clear() -> None

Empties the plotter and pushes the empty frame.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def clear(self) -> None:
    """Empties the plotter and pushes the empty frame."""
    self.plotter.clear()
    self.renderable = None
    self.push()

on_visible

on_visible(visible: bool) -> None

Tells the scene and the controller whether the view is on screen.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def on_visible(self, visible: bool) -> None:
    """Tells the scene and the controller whether the view is on screen."""
    super().on_visible(visible)
    self.controller.on_visible(self.id, visible)

dispose

dispose() -> None

Detaches from the controller, then frees the view.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def dispose(self) -> None:
    """Detaches from the controller, then frees the view."""
    self.controller.detach(self)
    super().dispose()

FigurePanel

FigurePanel(app: Workbench, id: str, title: str, figure: Figure | None = None, *, mode_bar: bool = True)

Bases: PanelComponent

Shows one plotly figure; update swaps it (uirevision keeps brushes and cameras), bind routes its events to handlers.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • figure (Figure | None, default: None ) –

    The figure shown first; None starts empty.

  • mode_bar (bool, default: True ) –

    Whether plotly's modebar shows.

Methods:

  • build –

    Pushes the figure into the state and builds the plot bound to it.

  • encode –

    The figure as plain JSON with the modebar switch and the export name it carries.

  • update –

    Replaces the figure.

  • bind –

    Routes a plotly event to handler with the event payload as its one argument.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/figure.py
def __init__(self, app: "Workbench", id: str, title: str, figure: go.Figure | None = None, *, mode_bar: bool = True) -> None:
    """Builds the panel around its first figure.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        figure: The figure shown first; None starts empty.
        mode_bar: Whether plotly's modebar shows.
    """
    self.initial = figure if figure is not None else go.Figure()
    self.mode_bar = mode_bar
    self.plot: Plot | None = None

    super().__init__(app, id, title)

build

build() -> None

Pushes the figure into the state and builds the plot bound to it.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/figure.py
def build(self) -> None:
    """Pushes the figure into the state and builds the plot bound to it."""
    # A rebuilt panel must push its figure: a bound default only fills a variable the last build left empty.
    self.state[self.key(FIGURE_KEY)] = self.encode(self.initial)
    self.plot = Plot(figure=react.Bind(self.key(FIGURE_KEY)))

encode

encode(figure: Figure) -> Json

The figure as plain JSON with the modebar switch and the export name it carries.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/figure.py
def encode(self, figure: go.Figure) -> Json:
    """The figure as plain JSON with the modebar switch and the export name it carries."""
    document = cast("Json", json.loads(figure.to_json()))
    document["config"] = {"displayModeBar": self.mode_bar, "toImageButtonOptions": export_options(figure)}

    return document

update

update(figure: Figure) -> None

Replaces the figure.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/figure.py
def update(self, figure: go.Figure) -> None:
    """Replaces the figure."""
    with self.state:
        self.state[self.key(FIGURE_KEY)] = self.encode(figure)

bind

bind(event: PlotEvent, handler: EventHandler) -> None

Routes a plotly event to handler with the event payload as its one argument.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/figure.py
def bind(self, event: PlotEvent, handler: EventHandler) -> None:
    """Routes a plotly event to `handler` with the event payload as its one argument."""
    if self.plot is None:
        return

    setattr(self.plot, f"{EVENT_PREFIX}{event}", self.callback(handler, args="[$event]"))
    self.refresh()

Panel dataclass

Panel(name: str, build: Builder, weight: float = 1, id: str | None = None)

One tab: its title, the builder that makes its component once the shell exists, and its share of the axis.

Attributes:

  • name (str) –

    The tab title, from which the panel id is derived when id is None.

  • build (Builder) –

    Makes the panel's component from the shell, the panel id and the title, once the shell exists.

  • weight (float) –

    [-]: The panel's share of its parent's axis, relative to its siblings' weights.

  • id (str | None) –

    The panel id; None derives it from name.

  • panel_id (str) –

    The id the dock knows the panel by: id when given, else name lowercased with dashes.

name instance-attribute

name: str

The tab title, from which the panel id is derived when id is None.

build instance-attribute

build: Builder

Makes the panel's component from the shell, the panel id and the title, once the shell exists.

weight class-attribute instance-attribute

weight: float = 1

[-]: The panel's share of its parent's axis, relative to its siblings' weights.

id class-attribute instance-attribute

id: str | None = None

The panel id; None derives it from name.

panel_id property

panel_id: str

The id the dock knows the panel by: id when given, else name lowercased with dashes.

PanelComponent

PanelComponent(app: Workbench, id: str, title: str)

Bases: TrameComponent

A dock panel: its template is built once inside layout; state variables go through key.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

Methods:

  • build –

    Adds the panel's widgets to its layout.

  • key –

    A state variable or trigger name namespaced to this panel.

  • status –

    Reports in the shell's status line.

  • refresh –

    Pushes the template again after its widgets changed.

  • callback –

    A client event handler calling fn with the JS args, unregistered again when the panel is disposed.

  • listen –

    Registers a state listener like state.change, one the panel drops again when disposed.

  • dispose –

    Releases what the panel registered on the server once it left the dock: triggers, state listeners and controller slots.

  • succeed –

    Called on a panel rebuilt under the id of previous, already disposed, to carry over what a fresh build would lose.

  • on_visible –

    Called when dockview shows or hides the panel.

  • on_layout_changed –

    Called after the dock re-laid out its groups.

Attributes:

  • listeners (list[Listener]) –

    The state listeners registered through listen, dropped again by dispose.

  • bound (list[Handler]) –

    The callables handed to callback, whose triggers dispose unregisters.

  • legacy (dict[Carry, object]) –

    What dispose leaves for a successor; read by its succeed, never through the reloadable classes.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def __init__(self, app: "Workbench", id: str, title: str) -> None:
    """Builds the panel's template inside its own layout.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
    """
    self.listeners: list[Listener] = []
    """The state listeners registered through `listen`, dropped again by `dispose`."""
    self.bound: list[Handler] = []
    """The callables handed to `callback`, whose triggers `dispose` unregisters."""
    self.legacy: dict[Carry, object] = {}
    """What `dispose` leaves for a successor; read by its `succeed`, never through the reloadable classes."""

    super().__init__(app.server)
    self.app = app
    self.id = id
    self.title = title
    self.template_name = template_name(id)
    self.layout = DivLayout(app.server, template_name=self.template_name, style=FILL)

    with self.layout:
        _ = app.theme()
        self.build()

listeners instance-attribute

listeners: list[Listener] = []

The state listeners registered through listen, dropped again by dispose.

bound instance-attribute

bound: list[Handler] = []

The callables handed to callback, whose triggers dispose unregisters.

legacy instance-attribute

legacy: dict[Carry, object] = {}

What dispose leaves for a successor; read by its succeed, never through the reloadable classes.

build

build() -> None

Adds the panel's widgets to its layout.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def build(self) -> None:
    """Adds the panel's widgets to its layout."""
    raise NotImplementedError

key

key(name: str) -> str

A state variable or trigger name namespaced to this panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def key(self, name: str) -> str:
    """A state variable or trigger name namespaced to this panel."""
    return f"{self.id}_{name}".replace("-", "_")

status

status(message: str) -> None

Reports in the shell's status line.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def status(self, message: str) -> None:
    """Reports in the shell's status line."""
    self.app.status(message)

refresh

refresh() -> None

Pushes the template again after its widgets changed.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def refresh(self) -> None:
    """Pushes the template again after its widgets changed."""
    self.layout.flush_content()

callback

callback(fn: Handler, args: str | None = None) -> Callback

A client event handler calling fn with the JS args, unregistered again when the panel is disposed.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def callback(self, fn: Handler, args: str | None = None) -> react.Callback:
    """A client event handler calling `fn` with the JS `args`, unregistered again when the panel is disposed."""
    self.bound.append(fn)

    return react.Callback(fn, args=args)

listen

listen(*names: str) -> Callable[[L], L]

Registers a state listener like state.change, one the panel drops again when disposed.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def listen[L: Listener](self, *names: str) -> Callable[[L], L]:
    """Registers a state listener like `state.change`, one the panel drops again when disposed."""

    def register(listener: L) -> L:
        _ = self.state.change(*names)(listener)
        self.listeners.append(listener)

        return listener

    return register

dispose

dispose() -> None

Releases what the panel registered on the server once it left the dock: triggers, state listeners and controller slots.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def dispose(self) -> None:
    """Releases what the panel registered on the server once it left the dock: triggers, state listeners and controller slots."""
    self._unbind_annotated_methods()

    # trame offers no public removal of a listener, nor a list of the triggers it named for a widget's callbacks; its own unbind reaches into the same dicts.
    for registered in self.state._change_callbacks.values():
        registered[:] = [(fn, translator) for fn, translator in registered if fn not in self.listeners]

    owned = [fn for fn in self.ctrl._triggers_fn2name if getattr(fn, "__self__", None) is self or fn in self.bound]
    for fn in owned:
        _ = self.ctrl.trigger_unregister(fn)

    self.listeners.clear()
    self.bound.clear()

succeed

succeed(previous: PanelComponent) -> None

Called on a panel rebuilt under the id of previous, already disposed, to carry over what a fresh build would lose.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def succeed(self, previous: "PanelComponent") -> None:
    """Called on a panel rebuilt under the id of `previous`, already disposed, to carry over what a fresh build would lose."""

on_visible

on_visible(visible: bool) -> None

Called when dockview shows or hides the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def on_visible(self, visible: bool) -> None:
    """Called when dockview shows or hides the panel."""

on_layout_changed

on_layout_changed() -> None

Called after the dock re-laid out its groups.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/__init__.py
def on_layout_changed(self) -> None:
    """Called after the dock re-laid out its groups."""

RunBrowserPanel

RunBrowserPanel(app: Workbench, id: str, title: str, *, telemetry: Telemetry)

Bases: TablePanel

The run index as a virtual table: text, status, tag and physics filters narrow it in polars, a row click selects the run.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • telemetry (Telemetry) –

    The telemetry root whose run index the table lists.

Methods:

  • query –

    The run index query before the toolbar's filters.

  • index –

    The narrowed index frame, with the live claims joined on.

  • table_frame –

    One index frame as the table's rows, remembering which of its runs are claimed and where their bundles live.

  • vocabulary –

    The distinct non-null values of one index column, for a filter's choices.

  • tag_vocabulary –

    Every key=value pair the indexed runs carry, deduplicated.

  • build –

    Builds the filter toolbar over the run table and listens to the filters.

  • count –

    The status text: how many runs the table holds and how many are live.

  • chosen –

    The one value a filter names, or nothing while it names any.

  • on_narrowing –

    Re-queries the index after a toolbar filter changed.

  • poll –

    Refreshed rows while a run is claimed or a claim just came or went, else None: an idle table costs one directory glob.

  • publish –

    Writes one index frame into the table.

  • on_pick –

    Publishes the clicked run for the detail panel.

  • on_visible –

    Runs the live refresh only while the panel is on screen.

  • dispose –

    Stops the live refresh, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def __init__(self, app: "Workbench", id: str, title: str, *, telemetry: Telemetry) -> None:
    """Builds the browser over a telemetry root's run index.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        telemetry: The telemetry root whose run index the table lists.
    """
    self.telemetry = telemetry
    self.narrowing = RunFilters()
    self.live: set[str] = set()
    self.directories: set[str] = set()
    self.feed: Feed[pl.DataFrame | None] = Feed(self, LIVE_PERIOD_S, self.poll, self.publish)

    super().__init__(app, id, title, self.table_frame(self.index()), filterable=True, virtual=True, pick=self.on_pick)

query

query() -> RunQuery

The run index query before the toolbar's filters.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def query(self) -> RunQuery:
    """The run index query before the toolbar's filters."""
    return RunQuery(self.telemetry.runs.scan())

index

index() -> DataFrame

The narrowed index frame, with the live claims joined on.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def index(self) -> pl.DataFrame:
    """The narrowed index frame, with the live claims joined on."""
    return self.narrowing.apply(self.query()).frame()

table_frame

table_frame(frame: DataFrame) -> DataFrame

One index frame as the table's rows, remembering which of its runs are claimed and where their bundles live.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def table_frame(self, frame: pl.DataFrame) -> pl.DataFrame:
    """One index frame as the table's rows, remembering which of its runs are claimed and where their bundles live."""
    self.live = live_ids(frame)
    self.directories = set(cast("list[str]", frame[RunColumn.RUNS_DIR].unique().to_list()))

    return pl.DataFrame(run_rows(frame, self.live), schema=RUN_SCHEMA)

vocabulary

vocabulary(column: RunColumn) -> list[str]

The distinct non-null values of one index column, for a filter's choices.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def vocabulary(self, column: RunColumn) -> list[str]:
    """The distinct non-null values of one index column, for a filter's choices."""
    found: pl.DataFrame = self.telemetry.runs.scan().select(pl.col(column).cast(pl.String)).drop_nulls().unique().sort(column).collect()

    return cast("list[str]", found[column].to_list())

tag_vocabulary

tag_vocabulary() -> list[str]

Every key=value pair the indexed runs carry, deduplicated.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def tag_vocabulary(self) -> list[str]:
    """Every ``key=value`` pair the indexed runs carry, deduplicated."""
    found: pl.DataFrame = self.telemetry.runs.scan().select(RunColumn.TAGS).drop_nulls().unique().collect()
    labels = {label for tags in cast("list[str]", found[RunColumn.TAGS].to_list()) for label in tag_pairs(tags)}

    return sorted(labels)

build

build() -> None

Builds the filter toolbar over the run table and listens to the filters.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build(self) -> None:
    """Builds the filter toolbar over the run table and listens to the filters."""
    with html.Div(classes=str(Cls.TOOLBAR)):
        Input(model=self.key(BrowserVar.TEXT), placeholder="Search", aria_label="Search the runs", classes="h-6 w-40")
        with labelled("Status"):
            chooser(self.key(BrowserVar.STATUS), [ANY, *self.vocabulary(RunColumn.STATUS)])

        with labelled("Tag"):
            chooser(self.key(BrowserVar.TAG), [ANY, *self.tag_vocabulary()], width="w-48", label=as_written)

        with labelled("Physics"):
            chooser(self.key(BrowserVar.PHYSICS), [ANY, *self.vocabulary(RunColumn.PHYSICS)], width="w-44")

        with labelled("Newest"):
            chooser(self.key(BrowserVar.LIMIT), [str(limit) for limit in LIMITS], width="w-20")

        html.Span([react.Bind(self.key(BrowserVar.COUNT))], classes=str(Cls.STATUS))

    with html.Div(classes=str(Cls.FILL)):
        super().build()

    self.state.update(
        {
            self.key(BrowserVar.TEXT): "",
            self.key(BrowserVar.STATUS): ANY,
            self.key(BrowserVar.TAG): ANY,
            self.key(BrowserVar.PHYSICS): ANY,
            self.key(BrowserVar.LIMIT): str(DEFAULT_LIMIT),
            self.key(BrowserVar.COUNT): self.count(),
        }
    )
    _ = self.listen(
        self.key(BrowserVar.TEXT),
        self.key(BrowserVar.STATUS),
        self.key(BrowserVar.TAG),
        self.key(BrowserVar.PHYSICS),
        self.key(BrowserVar.LIMIT),
    )(self.on_narrowing)

count

count() -> str

The status text: how many runs the table holds and how many are live.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def count(self) -> str:
    """The status text: how many runs the table holds and how many are live."""
    return f"{self.frame.height} runs · {len(self.live)} live"

chosen

chosen(name: BrowserVar) -> tuple[str, ...]

The one value a filter names, or nothing while it names any.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def chosen(self, name: BrowserVar) -> tuple[str, ...]:
    """The one value a filter names, or nothing while it names ``any``."""
    value = str(self.state[self.key(name)] or ANY)

    return () if value == ANY else (value,)

on_narrowing

on_narrowing(**_: object) -> None

Re-queries the index after a toolbar filter changed.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def on_narrowing(self, **_: object) -> None:
    """Re-queries the index after a toolbar filter changed."""
    self.narrowing = replace(
        self.narrowing,
        text=cast("str | None", self.state[self.key(BrowserVar.TEXT)]) or "",
        statuses=self.chosen(BrowserVar.STATUS),
        tags=self.chosen(BrowserVar.TAG),
        physics=self.chosen(BrowserVar.PHYSICS),
        limit=int(str(self.state[self.key(BrowserVar.LIMIT)])),
    )
    self.publish(self.index())

poll

poll() -> DataFrame | None

Refreshed rows while a run is claimed or a claim just came or went, else None: an idle table costs one directory glob.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def poll(self) -> pl.DataFrame | None:
    """Refreshed rows while a run is claimed or a claim just came or went, else None: an idle table costs one directory glob."""
    claims: set[str] = {run_id for directory in self.directories for run_id in live_ids(live_runs(directory))}
    if not claims and claims == self.live:
        return None

    return self.index()

publish

publish(frame: DataFrame | None) -> None

Writes one index frame into the table.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def publish(self, frame: pl.DataFrame | None) -> None:
    """Writes one index frame into the table."""
    if frame is None:
        return

    self.set_rows(self.table_frame(frame))
    self.state[self.key(BrowserVar.COUNT)] = self.count()

on_pick

on_pick(row: Row) -> None

Publishes the clicked run for the detail panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def on_pick(self, row: Row) -> None:
    """Publishes the clicked run for the detail panel."""
    run_id = str(row.get(RunRow.RUN_ID, ""))
    if not run_id:
        return

    select_run(self.state, run_id)

    log.bind(panel=self.id, run_id=run_id).info("Run selected.")

on_visible

on_visible(visible: bool) -> None

Runs the live refresh only while the panel is on screen.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def on_visible(self, visible: bool) -> None:
    """Runs the live refresh only while the panel is on screen."""
    if visible:
        self.feed.start()
        return

    self.feed.stop()

dispose

dispose() -> None

Stops the live refresh, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def dispose(self) -> None:
    """Stops the live refresh, then releases the panel."""
    self.feed.stop()
    super().dispose()

RunDetailPanel

RunDetailPanel(app: Workbench, id: str, title: str, *, telemetry: Telemetry, store: LocalArtifactStore)

Bases: PanelComponent

One run bundle: its manifest summary, stage timeline, results, artifact inputs, files and retention.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • telemetry (Telemetry) –

    The telemetry root whose run index locates a run's bundle.

  • store (LocalArtifactStore) –

    The artifact store a bundle's inputs resolve in and rehydrate from.

Methods:

  • build –

    Builds the toolbar, the bundle's sections and the purge dialog, then follows the shared run selection.

  • build_header –

    Builds the title, the status badge, the bundle path and the manifest summary.

  • build_stages –

    Builds the stage timeline, shown once the run has stages.

  • build_scalars –

    Builds the result scalars, shown once the run has a result.

  • build_tables –

    Builds the component forces, currents, inputs, files and retention passes, each shown once it has rows.

  • build_purge_dialog –

    Builds the purge dialog: a preset, a rule per retention class, and the preview and delete actions.

  • rule_key –

    The state variable holding one retention class's rule in the purge dialog.

  • bytes_key –

    The state variable holding one retention class's planned size in the purge dialog.

  • load_preset –

    Fills every per-class rule from a preset and clears the last preview.

  • policy –

    The policy the dialog's per-class choices describe.

  • on_preset –

    Fills the per-class rules from the preset just chosen.

  • locate –

    The bundle directory of an indexed run, or None when the index does not know it.

  • load –

    Read one bundle into what the panel shows; the whole read happens off the server loop.

  • summary –

    The manifest's identity and provenance as label/value pairs.

  • publish –

    Writes one view into the panel's state variables.

  • on_selected –

    Shows the run the shared selection names.

  • show –

    Loads a bundle off the server loop and publishes it, re-reading it while it solves; a failed read lands in the status line.

  • reread –

    The shown run read again, named so a reply that outlives its selection is dropped.

  • refresh –

    Publishes a re-read of the shown run, and ends the watch once the run has.

  • selected –

    The run the panel shows, reporting in the status line when there is none.

  • open_fields –

    Hands the bundle to the field viewer, or says so when no viewer is registered.

  • open_ref –

    Selects a store-sourced input, so the artifact detail follows.

  • reveal –

    Opens the bundle's directory in the file manager.

  • rehydrate –

    Fetches the bundle's store-sourced inputs back into its directory.

  • open_purge –

    Opens the purge dialog when the panel shows a run.

  • close_purge –

    Closes the purge dialog.

  • on_dialog –

    Follows the dialog the client closed with Escape or the backdrop.

  • preview_purge –

    Plans the dialog's policy against the bundle and shows the bytes each class would free.

  • apply_purge –

    Applies the previewed policy to the bundle and reloads the panel from what is left.

  • succeed –

    Shows the selected run again, which a fresh build would leave blank.

  • dispose –

    Stops the watch, then releases the panel.

Attributes:

  • run_id –

    The run the panel shows, which the watch re-reads while it solves.

  • showing (Task[None] | None) –

    The show a rebuild started, held until it finishes.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def __init__(self, app: "Workbench", id: str, title: str, *, telemetry: Telemetry, store: LocalArtifactStore) -> None:
    """Builds the detail panel, which follows the shared run selection.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        telemetry: The telemetry root whose run index locates a run's bundle.
        store: The artifact store a bundle's inputs resolve in and rehydrate from.
    """
    self.telemetry = telemetry
    self.store = store
    self.run: Run | None = None
    self.run_id = ""
    """The run the panel shows, which the watch re-reads while it solves."""
    self.plan: CleanupPlan | None = None
    self.watch: Feed[tuple[str, RunView]] = Feed(self, WATCH_PERIOD_S, self.reread, self.refresh)
    self.showing: asyncio.Task[None] | None = None
    """The show a rebuild started, held until it finishes."""

    super().__init__(app, id, title)

run_id instance-attribute

run_id = ''

The run the panel shows, which the watch re-reads while it solves.

showing instance-attribute

showing: Task[None] | None = None

The show a rebuild started, held until it finishes.

build

build() -> None

Builds the toolbar, the bundle's sections and the purge dialog, then follows the shared run selection.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build(self) -> None:
    """Builds the toolbar, the bundle's sections and the purge dialog, then follows the shared run selection."""
    with html.Div(classes=str(Cls.TOOLBAR)):
        Button(
            "Fields",
            size=BUTTON_SIZE,
            variant="secondary",
            disabled=react.Bind(f"!{self.key(DetailVar.HAS_FIELDS)}"),
            title=react.Bind(self.key(DetailVar.FIELDS_NOTE)),
            on_click=self.callback(self.open_fields),
        )
        Button("Folder", size=BUTTON_SIZE, variant="secondary", on_click=self.callback(self.reveal))
        Button("Rehydrate", size=BUTTON_SIZE, variant="ghost", on_click=self.callback(self.rehydrate))
        Button("Purge", size=BUTTON_SIZE, variant="ghost", on_click=self.callback(self.open_purge))
        html.Span([react.Bind(f"{self.key(DetailVar.TITLE)} || 'No Run Selected'")], classes=str(Cls.STATUS))

    with html.Div(classes=f"{Cls.COLUMN} {Cls.SCROLL}"):
        self.build_header()
        self.build_stages()
        self.build_scalars()
        self.build_tables()

    self.build_purge_dialog()

    self.state.update(
        {
            Shared.SELECTED_RUN_ID: self.state.setdefault(Shared.SELECTED_RUN_ID, ""),
            self.key(DetailVar.COMPONENT_COLUMNS): columns(
                (ComponentColumn.COMPONENT, "Component"),
                (ComponentColumn.FX, "Fx [N]", True),
                (ComponentColumn.FY, "Fy [N]", True),
                (ComponentColumn.FZ, "Fz [N]", True),
                (ComponentColumn.FORCE, "|F| [N]", True),
                (ComponentColumn.TORQUE, "|M| [N·m]", True),
            ),
            self.key(DetailVar.CURRENT_COLUMNS): columns((CurrentColumn.TERMINAL, "Terminal"), (CurrentColumn.AMPS, "I [A]", True)),
            self.key(DetailVar.FILE_COLUMNS): columns(
                (BundleFileColumn.PATH, "File"),
                (BundleFileColumn.SIZE, "Bytes", True),
                (BundleFileColumn.FILE_CLASS, "Class"),
                (BundleFileColumn.CACHED, "Cached"),
            ),
            self.key(DetailVar.CLEANUP_COLUMNS): columns(
                (CleanupColumn.APPLIED, "Applied"),
                (CleanupColumn.TRIGGER, "Trigger"),
                (CleanupColumn.FILES, "Files", True),
                (CleanupColumn.MB, "MB", True),
                (CleanupColumn.CLASSES, "Classes"),
            ),
        }
    )
    self.publish(RunView())
    _ = self.listen(str(Shared.SELECTED_RUN_ID))(self.on_selected)

build_header

build_header() -> None

Builds the title, the status badge, the bundle path and the manifest summary.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build_header(self) -> None:
    """Builds the title, the status badge, the bundle path and the manifest summary."""
    with html.Div(classes=str(Cls.SECTION)):
        html.Span([react.Bind(self.key(DetailVar.TITLE))])
        with react.If(self.key(DetailVar.STATUS)):
            Badge([react.Bind(self.key(DetailVar.STATUS))], variant=react.Bind(self.key(DetailVar.STATUS_VARIANT)))

    html.Div([react.Bind(self.key(DetailVar.PATH))], classes=f"{Cls.MONO} {Cls.CAPTION} px-2")
    build_pairs(self.key(DetailVar.META))

build_stages

build_stages() -> None

Builds the stage timeline, shown once the run has stages.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build_stages(self) -> None:
    """Builds the stage timeline, shown once the run has stages."""
    with react.If(f"{self.key(DetailVar.STAGES)}.length"):
        html.Div("Stages", classes=str(Cls.SECTION))
        with html.Ol(classes=str(Cls.TIMELINE)), react.For(items=self.key(DetailVar.STAGES), name=ROW):
            with html.Li(classes=str(Cls.TIMELINE_ITEM)):
                html.Span(classes=react.Bind(f"{ROW}.{StageField.DOT}"))
                html.Span([react.Bind(f"{ROW}.{StageField.STAGE}")])
                html.Span([react.Bind(f"{ROW}.{StageField.SECONDS}"), " s · ", react.Bind(f"{ROW}.{StageField.STATE}")], classes=str(Cls.CAPTION))

build_scalars

build_scalars() -> None

Builds the result scalars, shown once the run has a result.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build_scalars(self) -> None:
    """Builds the result scalars, shown once the run has a result."""
    with react.If(f"{self.key(DetailVar.SCALARS)}.length"):
        html.Div("Result", classes=str(Cls.SECTION))
        build_pairs(self.key(DetailVar.SCALARS))

build_tables

build_tables() -> None

Builds the component forces, currents, inputs, files and retention passes, each shown once it has rows.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build_tables(self) -> None:
    """Builds the component forces, currents, inputs, files and retention passes, each shown once it has rows."""

    def table(label: str, rows: DetailVar, spec: DetailVar) -> None:
        with react.If(f"{self.key(rows)}.length"):
            html.Div(label, classes=str(Cls.SECTION))
            build_table(self.key(spec), self.key(rows))

    table("Component Forces", DetailVar.COMPONENTS, DetailVar.COMPONENT_COLUMNS)
    table("Currents", DetailVar.CURRENTS, DetailVar.CURRENT_COLUMNS)

    with react.If(f"{self.key(DetailVar.REFS)}.length"):
        html.Div("Store-Sourced Inputs", classes=str(Cls.SECTION))
        with html.Div(classes=str(Cls.CHIPS)), react.For(items=self.key(DetailVar.REFS), name=REF):
            Badge(
                [react.Bind(f"{REF}.{RefColumn.KIND}"), "/", react.Bind(f"{REF}.{RefColumn.NAME}")],
                variant="outline",
                classes="cursor-pointer",
                title=react.Bind(f"{REF}.{RefColumn.PATH}"),
                on_click=self.callback(self.open_ref, args=f"[{REF}.{RefColumn.PATH}]"),
            )

    table("Files", DetailVar.FILES, DetailVar.FILE_COLUMNS)
    table("Retention", DetailVar.CLEANUPS, DetailVar.CLEANUP_COLUMNS)

build_purge_dialog

build_purge_dialog() -> None

Builds the purge dialog: a preset, a rule per retention class, and the preview and delete actions.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def build_purge_dialog(self) -> None:
    """Builds the purge dialog: a preset, a rule per retention class, and the preview and delete actions."""
    with Dialog(open_=react.Bind(self.key(PurgeVar.DIALOG)), on_open_change=self.callback(self.on_dialog, args="[$event]")), DialogContent(classes="max-w-2xl"):
        with DialogHeader():
            DialogTitle("Purge This Bundle")

        with html.Div(classes=str(Cls.COLUMN)):
            with labelled("Preset"):
                chooser(self.key(PurgeVar.PRESET), [preset.value for preset in Preset])

            with html.Div(classes=str(Cls.RULES)):
                for file_class in POLICY_CLASSES:
                    html.Span(file_class.value, classes=str(Cls.CAPTION))
                    chooser(self.rule_key(file_class), [rule.value for rule in Cleanup])
                    html.Span([react.Bind(self.bytes_key(file_class))], classes=str(Cls.MONO))

            html.Div([react.Bind(self.key(PurgeVar.TOTAL))], classes=str(Cls.CAPTION))

        with DialogFooter():
            Button("Close", size="sm", variant="secondary", on_click=self.callback(self.close_purge))
            Button("Preview", size="sm", variant="secondary", on_click=self.callback(self.preview_purge))
            Button(
                "Delete",
                size="sm",
                variant="destructive",
                disabled=react.Bind(f"!{self.key(PurgeVar.PLANNED)}"),
                on_click=self.callback(self.apply_purge),
            )

    self.state[self.key(PurgeVar.DIALOG)] = False
    self.state[self.key(PurgeVar.PRESET)] = Preset.DEFAULT.value
    self.load_preset(Preset.DEFAULT)
    _ = self.listen(self.key(PurgeVar.PRESET))(self.on_preset)

rule_key

rule_key(file_class: RunFile) -> str

The state variable holding one retention class's rule in the purge dialog.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def rule_key(self, file_class: RunFile) -> str:
    """The state variable holding one retention class's rule in the purge dialog."""
    return self.key(f"rule_{file_class.value}")

bytes_key

bytes_key(file_class: RunFile) -> str

The state variable holding one retention class's planned size in the purge dialog.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def bytes_key(self, file_class: RunFile) -> str:
    """The state variable holding one retention class's planned size in the purge dialog."""
    return self.key(f"bytes_{file_class.value}")

load_preset

load_preset(preset: Preset) -> None

Fills every per-class rule from a preset and clears the last preview.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def load_preset(self, preset: Preset) -> None:
    """Fills every per-class rule from a preset and clears the last preview."""
    policy: RetentionPolicy = RetentionPolicy.from_preset(preset)
    self.plan = None
    self.state.update({self.rule_key(file_class): rule.value for file_class, rule in policy.rules()})
    self.state.update({self.bytes_key(file_class): "" for file_class in POLICY_CLASSES})
    self.state.update({self.key(PurgeVar.PLANNED): False, self.key(PurgeVar.TOTAL): "preview to see what would go"})

policy

policy() -> RetentionPolicy

The policy the dialog's per-class choices describe.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def policy(self) -> RetentionPolicy:
    """The policy the dialog's per-class choices describe."""
    rules = {file_class.value: Cleanup(str(self.state[self.rule_key(file_class)])) for file_class in POLICY_CLASSES}

    return RetentionPolicy(**rules)

on_preset

on_preset(**_: object) -> None

Fills the per-class rules from the preset just chosen.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def on_preset(self, **_: object) -> None:
    """Fills the per-class rules from the preset just chosen."""
    self.load_preset(Preset(str(self.state[self.key(PurgeVar.PRESET)])))

locate

locate(run_id: str) -> Path | None

The bundle directory of an indexed run, or None when the index does not know it.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def locate(self, run_id: str) -> Path | None:
    """The bundle directory of an indexed run, or None when the index does not know it."""
    found: pl.DataFrame = RunQuery(self.telemetry.runs.scan()).where(pl.col(RunColumn.RUN_ID) == run_id).frame()
    if found.height == 0:
        return None

    directories = cast("list[str]", found[RunColumn.RUNS_DIR].to_list())

    return Path(directories[0]) / run_id

load

load(run_id: str) -> RunView

Read one bundle into what the panel shows; the whole read happens off the server loop.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def load(self, run_id: str) -> RunView:
    """Read one bundle into what the panel shows; the whole read happens off the server loop."""
    sim_dir: Path | None = self.locate(run_id)
    if sim_dir is None or not sim_dir.is_dir():
        self.run = None

        return RunView(title=run_id, status="gone", path=str(sim_dir or ""), fields_note="the bundle directory is gone")

    run = Run(sim_dir)
    self.run = run
    try:
        manifest: RunManifest = run.manifest
    except FileNotFoundError:
        return RunView(title=run_id, status=STARTING, path=str(sim_dir), fields_note="the run has not written its manifest yet")

    purged: CleanupRecord | None = run.purged(RunFile.FIELDS)
    note: str = f"fields purged on {purged.applied_at:%Y-%m-%d} by {purged.trigger.value}" if purged is not None else "open the solved fields"

    return RunView(
        title=manifest.run_id,
        status=manifest.status,
        path=str(sim_dir),
        meta=self.summary(manifest),
        stages=stage_rows(manifest.stage_history, datetime.now(timezone.utc), status=manifest.status),
        scalars=scalar_rows(run.result) if manifest.status != "running" else [],
        components=component_rows(run),
        currents=current_rows(run),
        refs=ref_rows(manifest.artifacts, self.store),
        files=file_rows(bundle_inventory(sim_dir), manifest.cached_inputs()),
        cleanups=cleanup_rows(manifest.cleanups),
        fields_note=note,
        has_fields=run.vtu_path is not None,
    )

summary

summary(manifest: RunManifest) -> list[Row]

The manifest's identity and provenance as label/value pairs.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def summary(self, manifest: RunManifest) -> list[Row]:
    """The manifest's identity and provenance as label/value pairs."""
    meta = manifest.config.meta
    stats = manifest.mesh_stats

    return [
        pair("created", f"{manifest.created_at:%Y-%m-%d %H:%M:%S} UTC"),
        pair("physics", title(manifest.physics.value)),
        pair("description", meta.description or ""),
        pair("tags", ", ".join(f"{key}={value}" for key, value in meta.tags.items())),
        pair("cache key", manifest.cache_key[:KEY_CHARS]),
        pair("mesh", f"{manifest.mesh_name} ({manifest.mesh_key[:KEY_CHARS]})"),
        pair("elements", stats.element_count if stats is not None else ""),
        pair("elmer", manifest.elmer_version or ""),
        pair("code", manifest.git_sha or ""),
        pair("published", manifest.published.name if manifest.published is not None else "no"),
    ]

publish

publish(view: RunView) -> None

Writes one view into the panel's state variables.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def publish(self, view: RunView) -> None:
    """Writes one view into the panel's state variables."""
    self.state.update(
        {
            self.key(DetailVar.TITLE): view.title,
            self.key(DetailVar.STATUS): title(view.status),
            self.key(DetailVar.STATUS_VARIANT): StatusVariant.of(view.status),
            self.key(DetailVar.PATH): view.path,
            self.key(DetailVar.META): list(view.meta),
            self.key(DetailVar.STAGES): list(view.stages),
            self.key(DetailVar.SCALARS): list(view.scalars),
            self.key(DetailVar.COMPONENTS): list(view.components),
            self.key(DetailVar.CURRENTS): list(view.currents),
            self.key(DetailVar.REFS): list(view.refs),
            self.key(DetailVar.FILES): list(view.files),
            self.key(DetailVar.CLEANUPS): list(view.cleanups),
            self.key(DetailVar.FIELDS_NOTE): view.fields_note,
            self.key(DetailVar.HAS_FIELDS): view.has_fields,
        }
    )

on_selected async

on_selected(**_: object) -> None

Shows the run the shared selection names.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
async def on_selected(self, **_: object) -> None:
    """Shows the run the shared selection names."""
    run_id = str(self.state[Shared.SELECTED_RUN_ID] or "")
    if not run_id:
        return

    await self.show(run_id)

show async

show(run_id: str) -> None

Loads a bundle off the server loop and publishes it, re-reading it while it solves; a failed read lands in the status line.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
async def show(self, run_id: str) -> None:
    """Loads a bundle off the server loop and publishes it, re-reading it while it solves; a failed read lands in the status line."""
    self.run_id = run_id
    self.watch.stop()
    try:
        view: RunView = await asyncio.to_thread(self.load, run_id)
    except Exception as error:
        log.opt(exception=True).warning("Run bundle could not be read.")
        self.status(f"{type(error).__name__}: {error}")
        return

    if run_id != self.run_id:
        return

    with self.state:
        self.publish(view)
        self.load_preset(Preset(str(self.state[self.key(PurgeVar.PRESET)])))

    if view.status in WATCHED:
        self.watch.start()

reread

reread() -> tuple[str, RunView]

The shown run read again, named so a reply that outlives its selection is dropped.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def reread(self) -> tuple[str, RunView]:
    """The shown run read again, named so a reply that outlives its selection is dropped."""
    run_id = self.run_id

    return run_id, self.load(run_id)

refresh

refresh(reply: tuple[str, RunView]) -> None

Publishes a re-read of the shown run, and ends the watch once the run has.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def refresh(self, reply: tuple[str, RunView]) -> None:
    """Publishes a re-read of the shown run, and ends the watch once the run has."""
    run_id, view = reply
    if run_id != self.run_id:
        return

    self.publish(view)
    if view.status not in WATCHED:
        self.watch.stop()
        log.bind(panel=self.id, run_id=run_id, status=view.status).info("Run watch ended.")

selected

selected() -> Run | None

The run the panel shows, reporting in the status line when there is none.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def selected(self) -> Run | None:
    """The run the panel shows, reporting in the status line when there is none."""
    if self.run is None:
        self.status("select a run first")

    return self.run

open_fields

open_fields() -> None

Hands the bundle to the field viewer, or says so when no viewer is registered.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def open_fields(self) -> None:
    """Hands the bundle to the field viewer, or says so when no viewer is registered."""
    run: Run | None = self.selected()
    if run is None:
        return

    hook = self.ctrl[Hook.FIELD_OPEN]
    if not hook.exists():
        self.status("no field viewer in this workbench")
        return

    _ = hook(run.path)

open_ref

open_ref(path: str) -> None

Selects a store-sourced input, so the artifact detail follows.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def open_ref(self, path: str) -> None:
    """Selects a store-sourced input, so the artifact detail follows."""
    self.state[Shared.SELECTED_ARTIFACT_PATH] = path

reveal

reveal() -> None

Opens the bundle's directory in the file manager.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def reveal(self) -> None:
    """Opens the bundle's directory in the file manager."""
    run: Run | None = self.selected()
    if run is None:
        return

    open_folder(run.path)

rehydrate

rehydrate() -> None

Fetches the bundle's store-sourced inputs back into its directory.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def rehydrate(self) -> None:
    """Fetches the bundle's store-sourced inputs back into its directory."""
    run: Run | None = self.selected()
    if run is None:
        return

    try:
        handles = rehydrate_bundle(run.path, ArtifactCache(self.store))
    except Exception as error:
        log.opt(exception=True).warning("Rehydrate failed.")
        self.status(f"{type(error).__name__}: {error}")
        return

    names = [name for handle in handles for name in handle.files]
    self.status(f"rehydrated {counted(len(names), 'file')} into {run.path.name}")

open_purge

open_purge() -> None

Opens the purge dialog when the panel shows a run.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def open_purge(self) -> None:
    """Opens the purge dialog when the panel shows a run."""
    if self.selected() is None:
        return

    self.state[self.key(PurgeVar.DIALOG)] = True

close_purge

close_purge() -> None

Closes the purge dialog.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def close_purge(self) -> None:
    """Closes the purge dialog."""
    self.state[self.key(PurgeVar.DIALOG)] = False

on_dialog

on_dialog(open_: object) -> None

Follows the dialog the client closed with Escape or the backdrop.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def on_dialog(self, open_: object) -> None:
    """Follows the dialog the client closed with Escape or the backdrop."""
    self.state[self.key(PurgeVar.DIALOG)] = bool(open_)

preview_purge

preview_purge() -> None

Plans the dialog's policy against the bundle and shows the bytes each class would free.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def preview_purge(self) -> None:
    """Plans the dialog's policy against the bundle and shows the bytes each class would free."""
    run: Run | None = self.selected()
    if run is None:
        return

    plan: CleanupPlan = plan_cleanup(run.path, self.policy(), status=run.manifest.status, cached=run.manifest.cached_inputs())
    self.plan = plan
    sizes = plan.bytes_by_class()
    self.state.update({self.bytes_key(file_class): megabytes(sizes.get(file_class, 0)) for file_class in POLICY_CLASSES})
    self.state.update(
        {
            self.key(PurgeVar.PLANNED): True,
            self.key(PurgeVar.TOTAL): f"{counted(len(plan.entries), 'path')} · {megabytes(plan.total_bytes)} would go",
        }
    )

    log.bind(run_id=run.path.name, paths=len(plan.entries), bytes=plan.total_bytes).info("Purge previewed.")

apply_purge

apply_purge() -> None

Applies the previewed policy to the bundle and reloads the panel from what is left.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def apply_purge(self) -> None:
    """Applies the previewed policy to the bundle and reloads the panel from what is left."""
    run: Run | None = self.selected()
    if run is None or self.plan is None:
        self.status("preview the purge first")
        return

    report = cleanup_bundle(run.path, self.plan.policy, dry_run=False, cached=run.manifest.cached_inputs())
    self.state[self.key(PurgeVar.DIALOG)] = False
    self.status(f"{counted(len(report.removed), 'path')} removed · {megabytes(report.bytes_freed)}")
    self.plan = None
    self.publish(self.load(run.path.name))

    log.bind(run_id=run.path.name, paths=len(report.removed), bytes=report.bytes_freed).info("Bundle purged.")

succeed

succeed(previous: PanelComponent) -> None

Shows the selected run again, which a fresh build would leave blank.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def succeed(self, previous: PanelComponent) -> None:
    """Shows the selected run again, which a fresh build would leave blank."""
    run_id = str(self.state[Shared.SELECTED_RUN_ID] or "")
    if not run_id:
        return

    self.showing = create_task(self.show(run_id))

dispose

dispose() -> None

Stops the watch, then releases the panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/runs.py
def dispose(self) -> None:
    """Stops the watch, then releases the panel."""
    self.watch.stop()
    super().dispose()

Split dataclass

Split(children: Sequence[Node], weight: float = 1)

Panels, tab groups or splits sharing one axis by weight; the axis alternates with nesting, horizontal at the root.

Attributes:

  • children (Sequence[Node]) –

    The panels, tab groups and splits along the axis, in order.

  • weight (float) –

    [-]: The split's share of its parent's axis, relative to its siblings' weights; ignored at the root.

children instance-attribute

children: Sequence[Node]

The panels, tab groups and splits along the axis, in order.

weight class-attribute instance-attribute

weight: float = 1

[-]: The split's share of its parent's axis, relative to its siblings' weights; ignored at the root.

TablePanel

TablePanel(app: Workbench, id: str, title: str, frame: DataFrame, *, grouped: bool = False, filterable: bool = False, lead: Sequence[str] = (), hidden: Sequence[str] = (), pick: Pick | None = None, selectable: Selectable = NO_SELECTION, select: Select | None = None, virtual: bool = False)

Bases: PanelComponent

A read-only table whose rows go over once, the client sorting, filtering, faceting, reordering, hiding and grouping them; lead orders the leading columns and pins the first.

Parameters:

  • app (Workbench) –

    The workbench shell the panel is built in.

  • id (str) –

    The panel id, which names its template and namespaces its state.

  • title (str) –

    The panel's tab title.

  • frame (DataFrame) –

    The rows to show.

  • grouped (bool, default: False ) –

    Groups the rows by the group label column that group_rows adds.

  • filterable (bool, default: False ) –

    Shows the column filters.

  • lead (Sequence[str], default: () ) –

    The columns sent first, in order; the first is pinned.

  • hidden (Sequence[str], default: () ) –

    The frame's columns sent with the rows but given no column, so a pick still sees them.

  • pick (Pick | None, default: None ) –

    Receives a clicked row; None leaves the rows unclickable.

  • selectable (Selectable, default: NO_SELECTION ) –

    Whether the user selects no rows, one or several.

  • select (Select | None, default: None ) –

    Receives the selected rows on every change; None ignores the selection.

  • virtual (bool, default: False ) –

    Renders only the rows in view, for frames too long to draw whole.

Methods:

  • build –

    Pushes the columns, rows and options into the state and builds the table bound to them.

  • names –

    The columns sent: lead first, the group label while grouped, then the rest without hidden; the client applies the persisted order.

  • columns –

    The column specs: numbers sort, filter and facet as numbers; the group label never facets.

  • defaults –

    The authored layout: grouped by the label column while grouped.

  • options –

    The table options: the layout the last session left, the authored defaults, and the filter, selection and virtual switches.

  • set_rows –

    Replaces the rows; the columns follow a changed schema, the client keeping its order, visibility, sort and filters.

  • restore –

    Applies a layout document on every client, the authored defaults for None.

  • reset_layout –

    Returns the table to the authored layout on every client.

  • scroll_to_end –

    Scrolls the table to its last row.

  • scroll_to_row –

    Scrolls the row at index of the shown rows into view.

  • open_config –

    Shows, or hides, the column configuration panel.

  • on_layout –

    Persists the layout the client reports and pushes it back, so a page that reloads is not served the layout this one replaced.

  • on_click –

    Hands a clicked data row to the pick handler.

  • on_select –

    Hands the selected rows to the select handler.

Attributes:

  • saved (Layout | None) –

    The layout the last session left, handed to the client through the options.

  • store_key (str) –

    The persisted document of this panel's layout.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def __init__(
    self,
    app: "Workbench",
    id: str,
    title: str,
    frame: pl.DataFrame,
    *,
    grouped: bool = False,
    filterable: bool = False,
    lead: Sequence[str] = (),
    hidden: Sequence[str] = (),
    pick: Pick | None = None,
    selectable: Selectable = NO_SELECTION,
    select: Select | None = None,
    virtual: bool = False,
) -> None:
    """Builds the panel over a frame, its layout restored from the store.

    Args:
        app: The workbench shell the panel is built in.
        id: The panel id, which names its template and namespaces its state.
        title: The panel's tab title.
        frame: The rows to show.
        grouped: Groups the rows by the `group` label column that `group_rows` adds.
        filterable: Shows the column filters.
        lead: The columns sent first, in order; the first is pinned.
        hidden: The frame's columns sent with the rows but given no column, so a pick still sees them.
        pick: Receives a clicked row; None leaves the rows unclickable.
        selectable: Whether the user selects no rows, one or several.
        select: Receives the selected rows on every change; None ignores the selection.
        virtual: Renders only the rows in view, for frames too long to draw whole.
    """
    self.frame = frame
    self.grouped = grouped
    self.filterable = filterable
    self.lead = tuple(lead)
    self.hidden = tuple(hidden)
    self.pick = pick
    self.selectable: Selectable = selectable
    self.select = select
    self.virtual = virtual
    self.saved: Layout | None = None
    """The layout the last session left, handed to the client through the options."""
    self.table: Table | None = None

    super().__init__(app, id, title)

saved instance-attribute

saved: Layout | None = None

The layout the last session left, handed to the client through the options.

store_key property

store_key: str

The persisted document of this panel's layout.

build

build() -> None

Pushes the columns, rows and options into the state and builds the table bound to them.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def build(self) -> None:
    """Pushes the columns, rows and options into the state and builds the table bound to them."""
    self.saved = self.app.persist.load(self.app.scope, self.store_key)
    pick = self.callback(self.on_click, args="[$event]") if self.pick is not None else None
    select = self.callback(self.on_select, args="[$event]") if self.selectable != NO_SELECTION and self.select is not None else None

    # A rebuilt panel must push its rows: a bound default only fills a variable the last build left empty.
    self.state[self.key(StateVar.COLUMNS)] = self.columns()
    self.state[self.key(StateVar.ROWS)] = to_rows(self.frame)
    self.state[self.key(StateVar.OPTIONS)] = self.options()
    self.table = Table(
        columns=react.Bind(self.key(StateVar.COLUMNS)),
        rows=react.Bind(self.key(StateVar.ROWS)),
        options=react.Bind(self.key(StateVar.OPTIONS)),
        on_layout_change=self.callback(self.on_layout, args="[$event]"),
        on_row_click=pick,
        on_selection_change=select,
    )

names

names() -> list[str]

The columns sent: lead first, the group label while grouped, then the rest without hidden; the client applies the persisted order.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def names(self) -> list[str]:
    """The columns sent: `lead` first, the group label while grouped, then the rest without `hidden`; the client applies the persisted order."""
    trailing = [name for name in self.frame.columns if name not in self.lead and name not in self.hidden and name != GROUP_COLUMN]
    label = [GROUP_COLUMN] if self.grouped else []

    return [*self.lead, *label, *trailing]

columns

columns() -> list[ColumnSpec]

The column specs: numbers sort, filter and facet as numbers; the group label never facets.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def columns(self) -> list[ColumnSpec]:
    """The column specs: numbers sort, filter and facet as numbers; the group label never facets."""

    def spec(name: str) -> ColumnSpec:
        column = ColumnSpec(field=name, title=title(name), numeric=self.frame.schema[name].is_numeric())
        if name == GROUP_COLUMN:
            column["facet"] = "none"

        return column

    return [spec(name) for name in self.names()]

defaults

defaults() -> LayoutDefaults

The authored layout: grouped by the label column while grouped.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def defaults(self) -> LayoutDefaults:
    """The authored layout: grouped by the label column while grouped."""
    if not self.grouped:
        return LayoutDefaults()

    return LayoutDefaults(groupBy=[GROUP_COLUMN])

options

options() -> TableOptions

The table options: the layout the last session left, the authored defaults, and the filter, selection and virtual switches.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def options(self) -> TableOptions:
    """The table options: the layout the last session left, the authored defaults, and the filter, selection and virtual switches."""
    return TableOptions(
        layout=self.saved,
        defaults=self.defaults(),
        filterable=self.filterable,
        selectable=self.selectable,
        virtual=self.virtual,
    )

set_rows

set_rows(frame: DataFrame) -> None

Replaces the rows; the columns follow a changed schema, the client keeping its order, visibility, sort and filters.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def set_rows(self, frame: pl.DataFrame) -> None:
    """Replaces the rows; the columns follow a changed schema, the client keeping its order, visibility, sort and filters."""
    changed = frame.columns != self.frame.columns
    self.frame = frame

    with self.state:
        if changed:
            self.state[self.key(StateVar.COLUMNS)] = self.columns()

        self.state[self.key(StateVar.ROWS)] = to_rows(frame)

restore

restore(layout: Layout | None) -> None

Applies a layout document on every client, the authored defaults for None.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def restore(self, layout: Layout | None) -> None:
    """Applies a layout document on every client, the authored defaults for None."""
    if self.table is not None:
        self.table.restore(layout)

reset_layout

reset_layout() -> None

Returns the table to the authored layout on every client.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def reset_layout(self) -> None:
    """Returns the table to the authored layout on every client."""
    if self.table is not None:
        self.table.reset_layout()

scroll_to_end

scroll_to_end() -> None

Scrolls the table to its last row.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def scroll_to_end(self) -> None:
    """Scrolls the table to its last row."""
    if self.table is not None:
        self.table.scroll_to_end()

scroll_to_row

scroll_to_row(index: int) -> None

Scrolls the row at index of the shown rows into view.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def scroll_to_row(self, index: int) -> None:
    """Scrolls the row at `index` of the shown rows into view."""
    if self.table is not None:
        self.table.scroll_to_row(index)

open_config

open_config(open: bool = True) -> None

Shows, or hides, the column configuration panel.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def open_config(self, open: bool = True) -> None:
    """Shows, or hides, the column configuration panel."""
    if self.table is not None:
        self.table.open_config(open)

on_layout

on_layout(document: object) -> None

Persists the layout the client reports and pushes it back, so a page that reloads is not served the layout this one replaced.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def on_layout(self, document: object) -> None:
    """Persists the layout the client reports and pushes it back, so a page that reloads is not served the layout this one replaced."""
    if not isinstance(document, Mapping):
        return

    self.saved = dict(cast("Mapping[str, object]", document))
    self.app.persist.save(self.app.scope, self.store_key, self.saved)
    with self.state:
        self.state[self.key(StateVar.OPTIONS)] = self.options()

    log.bind(panel=self.id, keys=sorted(self.saved)).info("Table layout saved.")

on_click

on_click(row: object) -> None

Hands a clicked data row to the pick handler.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def on_click(self, row: object) -> None:
    """Hands a clicked data row to the pick handler."""
    if self.pick is None or not isinstance(row, Mapping):
        return

    self.pick(dict(cast("Mapping[str, object]", row)))

on_select

on_select(selection: object) -> None

Hands the selected rows to the select handler.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/table.py
def on_select(self, selection: object) -> None:
    """Hands the selected rows to the select handler."""
    if self.select is None or not isinstance(selection, Mapping):
        return

    rows = cast("Mapping[str, object]", selection).get(SelectionKey.ROWS)
    if not isinstance(rows, list):
        return

    chosen = [dict(cast("Mapping[str, object]", row)) for row in cast("list[object]", rows) if isinstance(row, Mapping)]
    self.select(chosen)

Tabs dataclass

Tabs(panels: Sequence[Panel], weight: float = 1)

Panels sharing one tab group, the first shown.

Attributes:

  • panels (Sequence[Panel]) –

    The panels sharing the tab group, the first one shown.

  • weight (float) –

    [-]: The group's share of its parent's axis, relative to its siblings' weights.

panels instance-attribute

panels: Sequence[Panel]

The panels sharing the tab group, the first one shown.

weight class-attribute instance-attribute

weight: float = 1

[-]: The group's share of its parent's axis, relative to its siblings' weights.

Window

Bases: StrEnum

Where a served workbench opens.

Attributes:

  • BROWSER –

    A tab in the default browser.

  • APP –

    A Chrome/Edge app window; closing it stops the server.

  • WEBVIEW –

    pywebview, from the app extra.

BROWSER class-attribute instance-attribute

BROWSER = 'browser'

A tab in the default browser.

APP class-attribute instance-attribute

APP = 'app'

A Chrome/Edge app window; closing it stops the server.

WEBVIEW class-attribute instance-attribute

WEBVIEW = 'webview'

pywebview, from the app extra.

Workbench

Workbench(name: str, *, palette: Palette | None = None, title: str = 'femforge workbench', server: Server | None = None, persist: UiStateStore | None = None, user: str | None = None, catalogue: Sequence[WorkbenchSpec] | None = None)

Bases: TrameApp

One trame server with the header, the dock and the panels of a named workbench.

Parameters:

  • name (str) –

    The workbench name, which names the trame server and scopes the persisted state.

  • palette (Palette | None, default: None ) –

    The colours the shell and its panels paint in; None loads the dist's dark palette.

  • title (str, default: 'femforge workbench' ) –

    The title the header and the browser tab show.

  • server (Server | None, default: None ) –

    The trame server to run on; None makes one named after name.

  • persist (UiStateStore | None, default: None ) –

    Where layouts and opened cards persist; None keeps them in JSON files under the per-user config directory.

  • user (str | None, default: None ) –

    The user the persisted state is scoped to; None takes the OS user.

  • catalogue (Sequence[WorkbenchSpec] | None, default: None ) –

    The cards the title menu and a launcher panel list; None lists the discovered ones.

Methods:

  • load_catalogue –

    The discovered specs, empty with a warning when an entry point is broken.

  • build_ui –

    The theme switch, the header and the dock, whose panel bodies are the server templates the placements name.

  • theme –

    The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too.

  • build_switcher –

    The title as a menu over the catalogue.

  • add –

    Builds every panel's component and template; the dock arranges them once the client reports ready.

  • open –

    Adds a Split's panels to the running dock, building those not built yet, and activates the first.

  • open_spec –

    Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it.

  • opened_first –

    The first panel id of the open card of that name, None when it is not open.

  • close –

    Takes an opened card out of the dock and the shell and stops rebuilding it on save.

  • close_all –

    Closes every opened card.

  • discard –

    Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten.

  • register –

    Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken).

  • retire –

    Disposes the built panels of those ids and drops them from the shell.

  • drop_header –

    Removes a Split's widgets from the header, those still there.

  • watch –

    Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

  • unwatch –

    Stops rebuilding a card on save.

  • rebuild –

    Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

  • replace_root –

    Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement.

  • replace_card –

    Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

  • swap –

    Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

  • find_spec –

    The catalogue card of that name, None when the catalogue has none.

  • restore_opened –

    Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split.

  • remember_opened –

    Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards.

  • prune_opened –

    Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last.

  • place –

    add_panel for every placement of a Split, its first panel within reference's group when given; ids already in the dock are left alone.

  • panel –

    The built component of a panel, checked to be of kind.

  • status –

    Writes the status line, flushing so it also lands from outside a handler.

  • marshal –

    Runs fn(*args) on the server loop from any thread.

  • when_ready –

    Calls fn on the server loop: now when the server runs, else once it is ready.

  • authored_layout –

    The dockview layout of the authored Split; RuntimeError before add.

  • on_dock_ready –

    Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap.

  • on_layout_changed –

    Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel.

  • on_panel_removed –

    Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again.

  • rebuild_closed –

    Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live.

  • on_panel_visibility –

    Records that the dock showed or hid a panel and tells that panel.

  • reset_layout –

    Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window.

  • open_workbench –

    Hands a catalogue selection to the open_workbench hook, or names the command that opens it.

  • on_server_ready –

    Takes the server loop and opens the app window when one was asked for.

  • on_server_exited –

    Stops every reloader and flushes the persisted state.

  • open_app_window –

    Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab.

  • install_break_handler –

    Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way.

  • on_break –

    Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks.

  • run –

    Serves until stopped, or until an app window closes.

Attributes:

  • authored (Opened | None) –

    The Split add() built, named after the spec it serves when it is to be rebuilt on save.

  • opened (dict[str, Opened]) –

    Splits opened into the running dock by their first panel id.

  • pending (set[str]) –

    Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close.

  • on_screen (set[str]) –

    Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here.

  • catalogue –

    The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own.

  • header –

    The header's slot for a workbench's own controls: build widgets inside with app.header: and flush_content() after a later change.

  • root (Split | None) –

    The authored Split, None before add.

  • panel_ids (list[str]) –

    The ids of the built panels, in build order.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def __init__(
    self,
    name: str,
    *,
    palette: Palette | None = None,
    title: str = "femforge workbench",
    server: Server | None = None,
    persist: UiStateStore | None = None,
    user: str | None = None,
    catalogue: Sequence[WorkbenchSpec] | None = None,
) -> None:
    """Builds the shell: the header, the dock and the shared state, before any panel is added.

    Args:
        name: The workbench name, which names the trame server and scopes the persisted state.
        palette: The colours the shell and its panels paint in; None loads the dist's dark palette.
        title: The title the header and the browser tab show.
        server: The trame server to run on; None makes one named after `name`.
        persist: Where layouts and opened cards persist; None keeps them in JSON files under the per-user config directory.
        user: The user the persisted state is scoped to; None takes the OS user.
        catalogue: The cards the title menu and a launcher panel list; None lists the discovered ones.
    """
    gate_libraries(dict.fromkeys(QUIET_LIBRARIES, LogLevel.WARNING))
    super().__init__(server=server if server is not None else name, client_type="react")

    self.name = name
    self.title = title
    self.palette = palette if palette is not None else Palette.load()
    self.palette.install()
    self.persist: UiStateStore = persist if persist is not None else LocalJsonStore()
    self.scope = (user if user is not None else default_user(), name)
    self.loop: asyncio.AbstractEventLoop | None = None
    self.window = Window.BROWSER
    self.authored: Opened | None = None
    """The Split add() built, named after the spec it serves when it is to be rebuilt on save."""
    self.panels: list[PanelComponent] = []
    self.opened: dict[str, Opened] = {}
    """Splits opened into the running dock by their first panel id."""
    self.watches: list[Watch] = []
    self.pending: set[str] = set()
    """Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close."""
    self.on_screen: set[str] = set()
    """Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here."""
    self.catalogue = list(catalogue) if catalogue is not None else self.load_catalogue()
    """The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own."""
    self.dock: Dock | None = None
    self.header = DivLayout(self.server, template_name=str(Template.HEADER), classes=str(Cls.HEADER_SLOT))
    """The header's slot for a workbench's own controls: build widgets inside `with app.header:` and flush_content() after a later change."""

    cards = [{CardKey.NAME: spec.name, CardKey.TITLE: spec.title, CardKey.DESCRIPTION: spec.description, CardKey.GROUP: spec.group} for spec in self.catalogue]
    self.state.update(
        {
            StateVar.STATUS: "ready",
            StateVar.TITLE: title,
            StateVar.DARK: self.palette.is_dark,
            StateVar.WORKBENCHES: cards,
            StateVar.OPENED: [],
            TITLE_VAR: title,
        }
    )

    for library in WIDGET_LIBRARIES:
        library.initialize(self.server)

    self.ui = self.build_ui()

authored instance-attribute

authored: Opened | None = None

The Split add() built, named after the spec it serves when it is to be rebuilt on save.

opened instance-attribute

opened: dict[str, Opened] = {}

Splits opened into the running dock by their first panel id.

pending instance-attribute

pending: set[str] = set()

Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close.

on_screen instance-attribute

on_screen: set[str] = set()

Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here.

catalogue instance-attribute

catalogue = list(catalogue) if catalogue is not None else self.load_catalogue()

The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own.

header instance-attribute

header = DivLayout(self.server, template_name=str(Template.HEADER), classes=str(Cls.HEADER_SLOT))

The header's slot for a workbench's own controls: build widgets inside with app.header: and flush_content() after a later change.

root property

root: Split | None

The authored Split, None before add.

panel_ids property

panel_ids: list[str]

The ids of the built panels, in build order.

load_catalogue

load_catalogue() -> list[WorkbenchSpec]

The discovered specs, empty with a warning when an entry point is broken.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def load_catalogue(self) -> list[WorkbenchSpec]:
    """The discovered specs, empty with a warning when an entry point is broken."""
    try:
        return discover()
    except (TypeError, ValueError, ImportError):
        log.opt(exception=True).warning("Workbench catalogue unavailable.")
        return []

build_ui

build_ui() -> DivLayout

The theme switch, the header and the dock, whose panel bodies are the server templates the placements name.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def build_ui(self) -> DivLayout:
    """The theme switch, the header and the dock, whose panel bodies are the server templates the placements name."""
    with DivLayout(self.server, classes=str(Cls.ROOT)) as layout:
        _ = self.theme()
        if os.environ.get(DEV_ENV):
            client.Script(reload_script())

        with html.Header(classes=str(Cls.HEADER)):
            if self.catalogue:
                self.build_switcher()
            else:
                html.Span([react.Bind(StateVar.TITLE)], classes=str(Cls.TITLE))

            client.ServerTemplate(name=str(Template.HEADER))
            with react.If(f"{StateVar.OPENED}.length"):
                Button("Close All", variant="destructive", size="sm", title="Close every opened card", on_click=react.Callback(self.close_all))

            with Button(
                size="icon-sm",
                variant="ghost",
                title="Reset the layout",
                aria_label="Reset the layout",
                on_click=react.Callback(self.reset_layout),
            ):
                Icon("layout-grid")

            html.Span([react.Bind(StateVar.STATUS)], classes=str(Cls.STATUS))

        with html.Main(classes=str(Cls.MAIN)):
            with react.Slot(params=[PANEL_PARAM]) as render_panel:
                client.ServerTemplate(name=react.Bind(f"{PANEL_PARAM}.{dock.Key.TEMPLATE_NAME}"))

            self.dock = Dock(
                render_panel=render_panel,
                dark=react.Bind(StateVar.DARK),
                popout_url=asset_url(POPOUT_PAGE),
                floating_group_bounds="boundedWithinViewport",
                on_ready=react.Callback(self.on_dock_ready),
                on_layout_change=react.Callback(self.on_layout_changed, args="[$event]"),
                on_visibility=react.Callback(self.on_panel_visibility, args="[$event]"),
                on_panel_removed=react.Callback(self.on_panel_removed, args="[$event]"),
            )

    return layout

theme

theme() -> Theme

The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def theme(self) -> Theme:
    """The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too."""
    return Theme(dark=react.Bind(StateVar.DARK))

build_switcher

build_switcher() -> None

The title as a menu over the catalogue.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def build_switcher(self) -> None:
    """The title as a menu over the catalogue."""
    with DropdownMenu():
        with DropdownMenuTrigger(as_child=True), Button(variant="ghost", size="sm", classes=str(Cls.TITLE)):
            html.Span([react.Bind(StateVar.TITLE)])
            Icon("chevron-down")

        with DropdownMenuContent(align="start"), react.For(items=StateVar.WORKBENCHES, name="card"):
            with DropdownMenuItem(on_select=react.Callback(self.open_workbench, args=f"[card.{CardKey.NAME}]"), classes=str(Cls.MENU_ITEM)):
                html.Span([react.Bind(f"card.{CardKey.TITLE}")])
                html.Span([react.Bind(f"card.{CardKey.DESCRIPTION}")], classes=str(Cls.MENU_DESCRIPTION))

add

add(root: Split, *, spec: WorkbenchSpec | None = None, watch: Sequence[Path] = ()) -> None

Builds every panel's component and template; the dock arranges them once the client reports ready.

Parameters:

  • root (Split) –

    The Split to serve.

  • spec (WorkbenchSpec | None, default: None ) –

    The card the Split was built from; given, the Split is rebuilt on every save.

  • watch (Sequence[Path], default: () ) –

    More directories whose .py saves rebuild the Split, beside the spec's own roots.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def add(self, root: dock.Split, *, spec: WorkbenchSpec | None = None, watch: Sequence[Path] = ()) -> None:
    """Builds every panel's component and template; the dock arranges them once the client reports ready.

    Args:
        root: The Split to serve.
        spec: The card the Split was built from; given, the Split is rebuilt on every save.
        watch: More directories whose .py saves rebuild the Split, beside the spec's own roots.
    """
    leaves = dock.panels(root)
    ids = [leaf.panel_id for leaf in leaves]
    check_unique(ids)

    header_before = len(self.header.root.children)
    self.panels = []
    _ = self.register(leaves)
    self.authored = Opened(root, None, spec.name if spec is not None else None, tuple(self.header.root.children[header_before:]))

    log.bind(workbench=self.name, panels=ids).info("Panels built.")

    self.restore_opened()
    if spec is not None:
        self.watch(spec, extra=watch)

open

open(root: Split, *, reference: str | None = None, name: str | None = None) -> None

Adds a Split's panels to the running dock, building those not built yet, and activates the first.

Parameters:

  • root (Split) –

    The Split whose panels to add.

  • reference (str | None, default: None ) –

    The panel whose group the first panel joins; None lets the dock place it.

  • name (str | None, default: None ) –

    The card the Split comes from, remembered so the next start reopens it; None for a bare Split.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open(self, root: dock.Split, *, reference: str | None = None, name: str | None = None) -> None:
    """Adds a Split's panels to the running dock, building those not built yet, and activates the first.

    Args:
        root: The Split whose panels to add.
        reference: The panel whose group the first panel joins; None lets the dock place it.
        name: The card the Split comes from, remembered so the next start reopens it; None for a bare Split.
    """
    leaves = dock.panels(root)
    ids = [leaf.panel_id for leaf in leaves]
    check_unique(ids)

    known = self.panel_ids
    fresh = [leaf for leaf in leaves if leaf.panel_id not in known]
    header_before = len(self.header.root.children)
    _ = self.register(fresh)
    self.header.flush_content()
    self.opened[ids[0]] = Opened(root, reference, name, tuple(self.header.root.children[header_before:]))

    self.place(root, reference)
    if self.dock is not None:
        self.dock.active_panel(ids[0])

    if name is not None:
        self.remember_opened()

    log.bind(workbench=self.name, panels=ids, built=[leaf.panel_id for leaf in fresh]).info("Panels opened.")

open_spec

open_spec(spec: WorkbenchSpec, *, reference: str | None = None) -> None

Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open_spec(self, spec: WorkbenchSpec, *, reference: str | None = None) -> None:
    """Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it."""
    if spec.kind is not WorkbenchKind.PANEL:
        raise ValueError(f"{spec.name} is not a dock")

    factory = resolve(spec.target)
    if not callable(factory):
        raise TypeError(f"{spec.target} is not callable")

    root = factory()
    if not isinstance(root, dock.Split):
        raise TypeError(f"{spec.target}() returned {type(root).__name__}, not a Split")

    self.open(root, reference=reference, name=spec.name)
    self.watch(spec, reference=reference)

opened_first

opened_first(name: str) -> str | None

The first panel id of the open card of that name, None when it is not open.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def opened_first(self, name: str) -> str | None:
    """The first panel id of the open card of that name, None when it is not open."""
    return next((first for first, opened in self.opened.items() if opened.name == name), None)

close

close(name: str) -> None

Takes an opened card out of the dock and the shell and stops rebuilding it on save.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def close(self, name: str) -> None:
    """Takes an opened card out of the dock and the shell and stops rebuilding it on save."""
    self.discard(name)
    self.unwatch(name)

close_all

close_all() -> None

Closes every opened card.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def close_all(self) -> None:
    """Closes every opened card."""
    names = [opened.name for opened in list(self.opened.values()) if opened.name is not None]
    for name in names:
        self.close(name)

    self.status(f"{len(names)} cards closed")

discard

discard(name: str) -> None

Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def discard(self, name: str) -> None:
    """Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten."""
    first = self.opened_first(name)
    if first is None:
        raise KeyError(f"{name} is not open")

    opened = self.opened.pop(first)
    ids = dock.panel_ids(opened.root)
    self.pending.difference_update(ids)
    self.on_screen.difference_update(ids)
    if self.dock is not None:
        for panel_id in ids:
            self.dock.remove_panel(panel_id)

    self.retire(ids)
    self.drop_header(opened.header)
    self.header.flush_content()
    self.remember_opened()
    log.bind(workbench=self.name, card=name, panels=ids).info("Card closed.")

register

register(leaves: Sequence[Panel]) -> list[PanelComponent]

Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken).

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def register(self, leaves: Sequence[dock.Panel]) -> list[PanelComponent]:
    """Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken)."""
    built: list[PanelComponent] = []
    try:
        for leaf in leaves:
            panel = leaf.build(self, leaf.panel_id, leaf.name)
            built.append(panel)
            self.panels.append(panel)
    except Exception:
        for panel in built:
            panel.dispose()
            self.panels.remove(panel)

        raise

    return built

retire

retire(ids: Sequence[str]) -> None

Disposes the built panels of those ids and drops them from the shell.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def retire(self, ids: Sequence[str]) -> None:
    """Disposes the built panels of those ids and drops them from the shell."""
    for panel in [panel for panel in self.panels if panel.id in ids]:
        panel.dispose()
        self.panels.remove(panel)

drop_header

drop_header(widgets: Sequence[object]) -> None

Removes a Split's widgets from the header, those still there.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def drop_header(self, widgets: Sequence[object]) -> None:
    """Removes a Split's widgets from the header, those still there."""
    for widget in widgets:
        if widget in self.header.root.children:
            self.header.root.children.remove(widget)

watch

watch(spec: WorkbenchSpec, *, reference: str | None = None, extra: Sequence[Path] = ()) -> None

Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

Parameters:

  • spec (WorkbenchSpec) –

    The card whose target builds the Split.

  • reference (str | None, default: None ) –

    The panel whose group the rebuilt card joins when it is no longer open.

  • extra (Sequence[Path], default: () ) –

    More directories to watch, beside the femforge core, the workbench package and the spec's own source root.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def watch(self, spec: WorkbenchSpec, *, reference: str | None = None, extra: Sequence[Path] = ()) -> None:
    """Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

    Args:
        spec: The card whose target builds the Split.
        reference: The panel whose group the rebuilt card joins when it is no longer open.
        extra: More directories to watch, beside the femforge core, the workbench package and the spec's own source root.
    """
    self.unwatch(spec.name)
    if persistent(spec.module):
        return

    roots = [*default_roots(spec.module), *extra]

    def apply(root: dock.Split, seconds: float) -> None:
        self.rebuild(spec.name, root, reference=reference, seconds=seconds)

    def failed(error: Exception) -> None:
        self.status(f"{type(error).__name__}: {error}")

    reloader = Reloader(spec.target, roots, apply, kind=dock.Split, marshal=self.marshal, failed=failed)
    watch = Watch(spec.name, reloader)
    self.watches.append(watch)
    if self.loop is not None:
        reloader.start()
        return

    def start(**_: object) -> None:
        if watch in self.watches:
            reloader.start()

    _ = self.ctrl.on_server_ready.add(start)

unwatch

unwatch(name: str) -> None

Stops rebuilding a card on save.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def unwatch(self, name: str) -> None:
    """Stops rebuilding a card on save."""
    for watch in [watch for watch in self.watches if watch.name == name]:
        watch.reloader.stop()
        self.watches.remove(watch)

rebuild

rebuild(name: str, root: Split, *, reference: str | None = None, seconds: float = 0.0) -> None

Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

Parameters:

  • name (str) –

    The card to rebuild.

  • root (Split) –

    The fresh build of its Split.

  • reference (str | None, default: None ) –

    The panel whose group the card joins when it is no longer open.

  • seconds (float, default: 0.0 ) –

    The seconds the build took, added to the rebuild time the status line reports.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def rebuild(self, name: str, root: dock.Split, *, reference: str | None = None, seconds: float = 0.0) -> None:
    """Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

    Args:
        name: The card to rebuild.
        root: The fresh build of its Split.
        reference: The panel whose group the card joins when it is no longer open.
        seconds: The seconds the build took, added to the rebuild time the status line reports.
    """
    started = time.perf_counter()
    # Delivered through call_soon_threadsafe, outside any trame handler, so the state transaction is explicit for the flush.
    with self.state:
        if self.authored is not None and self.authored.name == name:
            self.replace_root(root)
        else:
            self.replace_card(name, root, reference)

    elapsed = seconds + time.perf_counter() - started
    log.bind(workbench=self.name, card=name, seconds=round(elapsed, 2)).info("Card rebuilt.")
    self.status(f"rebuilt {name} in {elapsed:.1f} s")

replace_root

replace_root(root: Split) -> None

Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def replace_root(self, root: dock.Split) -> None:
    """Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement."""
    authored = self.authored
    if authored is None:
        raise RuntimeError("add() a Split before rebuilding it")

    retired = dock.panel_ids(authored.root)
    header = self.swap(root, retired, authored.header)
    self.authored = replace(authored, root=root, header=header)
    if dock.panel_ids(root) != retired:
        self.reset_layout()

replace_card

replace_card(name: str, root: Split, reference: str | None) -> None

Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

Parameters:

  • name (str) –

    The card to rebuild.

  • root (Split) –

    The fresh build of its Split.

  • reference (str | None) –

    The panel whose group the card joins when it is not open.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def replace_card(self, name: str, root: dock.Split, reference: str | None) -> None:
    """Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

    Args:
        name: The card to rebuild.
        root: The fresh build of its Split.
        reference: The panel whose group the card joins when it is not open.
    """
    first = self.opened_first(name)
    if first is None:
        self.open(root, reference=reference, name=name)
        return

    opened = self.opened[first]
    retired = dock.panel_ids(opened.root)
    if dock.panel_ids(root) != retired:
        self.discard(name)
        self.open(root, reference=opened.reference, name=name)
        return

    header = self.swap(root, retired, opened.header)
    self.opened[first] = replace(opened, root=root, header=header)

swap

swap(root: Split, retired: Sequence[str], header: Sequence[object]) -> tuple[object, ...]

Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

Parameters:

  • root (Split) –

    The Split whose leaves to build.

  • retired (Sequence[str]) –

    The ids of the panels to dispose first.

  • header (Sequence[object]) –

    The header widgets the retired panels added, dropped with them.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def swap(self, root: dock.Split, retired: Sequence[str], header: Sequence[object]) -> tuple[object, ...]:
    """Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

    Args:
        root: The Split whose leaves to build.
        retired: The ids of the panels to dispose first.
        header: The header widgets the retired panels added, dropped with them.
    """
    leaves = dock.panels(root)
    check_unique([leaf.panel_id for leaf in leaves])
    previous = [panel for panel in self.panels if panel.id in retired]
    self.retire(retired)
    self.drop_header(header)
    header_before = len(self.header.root.children)

    try:
        built = self.register(leaves)
    except Exception:
        del self.header.root.children[header_before:]
        self.header.flush_content()
        raise

    for panel in built:
        predecessor = next((old for old in previous if old.id == panel.id), None)
        if predecessor is not None:
            panel.succeed(predecessor)

        if panel.id in self.on_screen:
            panel.on_visible(True)

    self.header.flush_content()

    return tuple(self.header.root.children[header_before:])

find_spec

find_spec(name: str) -> WorkbenchSpec | None

The catalogue card of that name, None when the catalogue has none.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def find_spec(self, name: str) -> WorkbenchSpec | None:
    """The catalogue card of that name, None when the catalogue has none."""
    return next((spec for spec in self.catalogue if spec.name == name), None)

restore_opened

restore_opened() -> None

Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def restore_opened(self) -> None:
    """Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split."""
    saved = self.persist.load(self.scope, StateKey.OPENED) or {}
    raw = saved.get(OpenedKey.ENTRIES)
    entries = cast("list[object]", raw) if isinstance(raw, list) else []

    for entry in entries:
        record = cast("dict[str, object]", entry) if isinstance(entry, dict) else {}
        name = str(record.get(OpenedKey.NAME))
        reference = record.get(OpenedKey.REFERENCE)
        spec = self.find_spec(name)
        if spec is None or spec.kind is not WorkbenchKind.PANEL:
            log.bind(workbench=self.name, card=name).warning("Opened card is no longer a dock in the catalogue.")
            continue

        anchor = reference if isinstance(reference, str) and reference in self.panel_ids else None
        try:
            self.open_spec(spec, reference=anchor)
        except Exception:  # noqa: BLE001 - a card that fails to build costs its panels, never the start
            log.bind(workbench=self.name, card=name).opt(exception=True).warning("Opened card failed to rebuild.")

    self.remember_opened()

remember_opened

remember_opened() -> None

Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def remember_opened(self) -> None:
    """Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards."""
    cards = [opened for opened in self.opened.values() if opened.name is not None]
    entries = [{str(OpenedKey.NAME): opened.name, str(OpenedKey.REFERENCE): opened.reference} for opened in cards]

    self.persist.save(self.scope, StateKey.OPENED, {str(OpenedKey.ENTRIES): entries})
    with self.state:
        self.state[StateVar.OPENED] = [opened.name for opened in cards]

prune_opened

prune_opened(layout: Mapping[str, object]) -> None

Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def prune_opened(self, layout: Mapping[str, object]) -> None:
    """Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last."""
    panels = layout.get(dock.Key.PANELS)
    present = set(cast("Mapping[str, object]", panels).keys()) if isinstance(panels, Mapping) else set[str]()
    if self.pending:
        if not self.pending <= present:
            return

        self.pending.clear()

    closed = [first for first in self.opened if first not in present]
    if not closed:
        return

    for first in closed:
        del self.opened[first]

    self.remember_opened()
    log.bind(workbench=self.name, panels=closed).info("Opened panels closed.")

place

place(root: Split, reference: str | None = None) -> None

add_panel for every placement of a Split, its first panel within reference's group when given; ids already in the dock are left alone.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def place(self, root: dock.Split, reference: str | None = None) -> None:
    """add_panel for every placement of a Split, its first panel within `reference`'s group when given; ids already in the dock are left alone."""
    if self.dock is None:
        return

    for placement in dock.placements(root):
        anchored = replace(placement, reference=reference, direction=dock.PlacementDirection.WITHIN) if placement.reference is None and reference is not None else placement
        self.dock.add_panel(anchored.panel.panel_id, anchored.panel.name, dock.template_name(anchored.panel.panel_id), **anchored.add_on())
        self.pending.add(anchored.panel.panel_id)

panel

panel(panel_id: str, kind: type[P]) -> P

The built component of a panel, checked to be of kind.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def panel[P: PanelComponent](self, panel_id: str, kind: type[P]) -> P:
    """The built component of a panel, checked to be of `kind`."""
    found = next((panel for panel in self.panels if panel.id == panel_id), None)
    if not isinstance(found, kind):
        raise KeyError(f"no {kind.__name__} with id {panel_id!r}")

    return found

status

status(message: str) -> None

Writes the status line, flushing so it also lands from outside a handler.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def status(self, message: str) -> None:
    """Writes the status line, flushing so it also lands from outside a handler."""
    with self.state:
        self.state[StateVar.STATUS] = message

marshal

marshal(fn: Callable[..., object], *args: object) -> None

Runs fn(*args) on the server loop from any thread.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def marshal(self, fn: Callable[..., object], *args: object) -> None:
    """Runs `fn(*args)` on the server loop from any thread."""
    if self.loop is None:
        raise RuntimeError("the server is not running")

    _ = self.loop.call_soon_threadsafe(fn, *args)

when_ready

when_ready(fn: Callable[[], None]) -> None

Calls fn on the server loop: now when the server runs, else once it is ready.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def when_ready(self, fn: Callable[[], None]) -> None:
    """Calls `fn` on the server loop: now when the server runs, else once it is ready."""
    if self.loop is not None:
        fn()
        return

    def ready(**_: object) -> None:
        fn()

    _ = self.ctrl.on_server_ready.add(ready)

authored_layout

authored_layout() -> Layout

The dockview layout of the authored Split; RuntimeError before add.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def authored_layout(self) -> dock.Layout:
    """The dockview layout of the authored Split; RuntimeError before `add`."""
    if self.root is None:
        raise RuntimeError("add() a Split before the dock is ready")

    return dock.layout_json(self.root)

on_dock_ready

on_dock_ready() -> None

Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_dock_ready(self) -> None:
    """Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap."""
    if self.dock is None or self.root is None:
        return

    saved = dock.restorable(self.persist.load(self.scope, StateKey.LAYOUT), self.panel_ids)
    self.dock.restore_layout(saved if saved is not None else self.authored_layout())

    self.place(self.root)
    for opened in self.opened.values():
        self.place(opened.root, opened.reference)

    log.bind(workbench=self.name, restored=saved is not None).info("Dock ready.")

on_layout_changed

on_layout_changed(layout: Mapping[str, object]) -> None

Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_layout_changed(self, layout: Mapping[str, object]) -> None:
    """Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel."""
    self.persist.save(self.scope, StateKey.LAYOUT, layout)
    self.prune_opened(layout)

    for panel in self.panels:
        panel.on_layout_changed()

on_panel_removed

on_panel_removed(event: Mapping[str, object]) -> None

Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_panel_removed(self, event: Mapping[str, object]) -> None:
    """Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again."""
    panel_id = str(event.get("id"))
    if panel_id not in self.panel_ids:
        return

    self.pending.discard(panel_id)
    self.on_screen.discard(panel_id)
    self.retire([panel_id])
    log.bind(workbench=self.name, panel=panel_id).info("Panel closed.")

rebuild_closed

rebuild_closed() -> None

Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def rebuild_closed(self) -> None:
    """Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live."""
    cards: list[tuple[str | None, Opened]] = [(first, opened) for first, opened in self.opened.items()]
    if self.authored is not None:
        cards.insert(0, (None, self.authored))

    for first, card in cards:
        ids = dock.panel_ids(card.root)
        if all(panel_id in self.panel_ids for panel_id in ids):
            continue

        rebuilt = replace(card, header=self.swap(card.root, ids, card.header))
        if first is None:
            self.authored = rebuilt
        else:
            self.opened[first] = rebuilt

        log.bind(workbench=self.name, panels=ids).info("Closed panels rebuilt.")

on_panel_visibility

on_panel_visibility(change: Mapping[str, object]) -> None

Records that the dock showed or hid a panel and tells that panel.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_panel_visibility(self, change: Mapping[str, object]) -> None:
    """Records that the dock showed or hid a panel and tells that panel."""
    panel_id = str(change.get("id"))
    visible = bool(change.get("visible"))
    if visible:
        self.on_screen.add(panel_id)
    else:
        self.on_screen.discard(panel_id)

    for panel in self.panels:
        if panel.id == panel_id:
            panel.on_visible(visible)

reset_layout

reset_layout() -> None

Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def reset_layout(self) -> None:
    """Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window."""
    if self.dock is None:
        return

    self.rebuild_closed()
    self.dock.restore_layout(self.authored_layout())
    for opened in self.opened.values():
        self.place(opened.root, opened.reference)

    self.status("layout reset")

open_workbench

open_workbench(name: str) -> None

Hands a catalogue selection to the open_workbench hook, or names the command that opens it.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def open_workbench(self, name: str) -> None:
    """Hands a catalogue selection to the open_workbench hook, or names the command that opens it."""
    spec = self.find_spec(name)
    if spec is None:
        self.status(f"unknown workbench {name}")
        return

    if spec.name == self.name:
        self.status(f"{spec.title} is open")
        return

    hook = self.ctrl[Hook.OPEN_WORKBENCH]
    if hook.exists():
        _ = hook(spec)
        return

    self.status(f"open with: {' '.join(spec.command())}")
    log.bind(workbench=spec.name, command=spec.command()).info("Workbench selected.")

on_server_ready

on_server_ready(**_: object) -> None

Takes the server loop and opens the app window when one was asked for.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@life_cycle.server_ready
def on_server_ready(self, **_: object) -> None:
    """Takes the server loop and opens the app window when one was asked for."""
    self.loop = asyncio.get_running_loop()
    if self.window is Window.APP:
        self.open_app_window()

on_server_exited

on_server_exited(**_: object) -> None

Stops every reloader and flushes the persisted state.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@life_cycle.server_exited
def on_server_exited(self, **_: object) -> None:
    """Stops every reloader and flushes the persisted state."""
    for watch in self.watches:
        watch.reloader.stop()

    self.persist.flush()

open_app_window

open_app_window() -> None

Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open_app_window(self) -> None:
    """Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab."""
    url = f"http://localhost:{self.server.port}/"
    browser = next((path for path in BROWSERS if path.exists()), None)
    if browser is None:
        log.bind(url=url).warning("No Chrome or Edge found; opening the system browser.")
        _ = webbrowser.open(url)
        return

    profile = Path(platformdirs.user_cache_dir(APP_NAME, appauthor=False)) / APP_PROFILE
    profile.mkdir(parents=True, exist_ok=True)
    command = [
        str(browser),
        f"--app={url}",
        f"--user-data-dir={profile}",
        "--no-first-run",
        "--no-default-browser-check",
        f"--window-size={APP_WINDOW_SIZE}",
    ]
    process = subprocess.Popen(command)
    log.bind(browser=browser.name, pid=process.pid).info("App window opened.")

    def stop_with_window() -> None:
        code = process.wait()
        log.bind(returncode=code).info("App window closed; stopping the server.")
        if self.loop is not None:
            _ = asyncio.run_coroutine_threadsafe(self.server.stop(), self.loop)

    threading.Thread(target=stop_with_window, name="app-window", daemon=True).start()

install_break_handler

install_break_handler() -> None

Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def install_break_handler(self) -> None:
    """Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way."""
    _ = signal.signal(signal.SIGTERM, self.on_break)
    if sys.platform == "win32":
        _ = signal.signal(signal.SIGBREAK, self.on_break)

on_break

on_break(signum: int, frame: FrameType | None) -> None

Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def on_break(self, signum: int, frame: FrameType | None) -> None:
    """Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks."""
    loop = self.loop
    if loop is None:
        return

    def stop() -> None:
        _ = asyncio.ensure_future(self.server.stop())

    loop.call_soon_threadsafe(stop)

run

run(*, port: int = 8080, window: Window = BROWSER) -> None

Serves until stopped, or until an app window closes.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def run(self, *, port: int = 8080, window: Window = Window.BROWSER) -> None:
    """Serves until stopped, or until an app window closes."""
    self.window = window
    # trame's own --app flag means pywebview; the app window is ours.
    sys.argv = [arg for arg in sys.argv if arg != APP_FLAG]
    self.install_break_handler()

    log.bind(workbench=self.name, port=port, window=str(window)).info("Starting the workbench.")
    _ = self.server.start(port=port, exec_mode="desktop" if window is Window.WEBVIEW else "main", open_browser=window is Window.BROWSER, timeout=0)

artifact_store

artifact_store(store: Telemetry) -> LocalArtifactStore

The store the artifact panels read: FEMFORGE_ARTIFACTS when set, else the root the index rows name.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/browsers.py
def artifact_store(store: Telemetry) -> LocalArtifactStore:
    """The store the artifact panels read: ``FEMFORGE_ARTIFACTS`` when set, else the root the index rows name."""
    named: str | None = os.environ.get(STORE_ENV)
    if named:
        return LocalArtifactStore(named)

    roots: pl.DataFrame = store.artifacts.scan().select(ArtifactColumn.STORE_ROOT).drop_nulls().unique().sort(ArtifactColumn.STORE_ROOT).collect()
    found: list[str] = cast("list[str]", roots[ArtifactColumn.STORE_ROOT].to_list())

    return LocalArtifactStore(found[0] if found else store.root)

field_viewer

field_viewer(run: Result | Location | None = None, *, display: DisplaySpec = DEFAULTS, telemetry: str | Path | None = None) -> Split

The field viewer dock: the controls beside the 3D view and the 2D plot as tabs.

Parameters:

  • run (Result | Location | None, default: None ) –

    The run to open once the server runs: a Result, a Run, a run id or a bundle directory; None opens none.

  • display (DisplaySpec, default: DEFAULTS ) –

    The display the controls start from.

  • telemetry (str | Path | None, default: None ) –

    The telemetry root whose run index the run picker lists; None reads FEMFORGE_TELEMETRY.

Source code in packages/femforge-workbench/src/femforge_workbench/panels/field_viewer.py
def workbench(run: Result | Location | None = None, *, display: DisplaySpec = DEFAULTS, telemetry: str | Path | None = None) -> dock.Split:
    """The field viewer dock: the controls beside the 3D view and the 2D plot as tabs.

    Args:
        run: The run to open once the server runs: a Result, a Run, a run id or a bundle directory; None opens none.
        display: The display the controls start from.
        telemetry: The telemetry root whose run index the run picker lists; None reads FEMFORGE_TELEMETRY.
    """
    root = resolve_root(telemetry)
    index = Telemetry(root) if root is not None else None
    location = Run.from_result(run) if isinstance(run, Result) else run
    shared: FieldViewerController | None = None

    def controller_for(app: Workbench) -> FieldViewerController:
        nonlocal shared
        if shared is None or shared.app is not app:
            shared = FieldViewerController(app, telemetry=index, display=display)
            if location is not None:
                app.when_ready(partial(shared.open, location))

        return shared

    def controls(app: Workbench, id: str, title: str) -> FieldControlsPanel:
        panel = FieldControlsPanel(app, id, title, controller_for(app))
        with app.header:
            panel.build_run_picker()

        return panel

    def view_3d(app: Workbench, id: str, title: str) -> FieldViewerPanel:
        return FieldViewerPanel(app, id, title, controller_for(app))

    def plot_2d(app: Workbench, id: str, title: str) -> FieldPlot2DPanel:
        return FieldPlot2DPanel(app, id, title, controller_for(app))

    views = dock.Tabs([dock.Panel("3D view", view_3d, id=FieldPanel.VIEW_3D), dock.Panel("2D plot", plot_2d, id=FieldPanel.PLOT_2D)], weight=3)

    return dock.Split([dock.Panel(FIELD_VIEWER.title, controls, id=FieldPanel.CONTROLS, weight=1), views])

show

show(workbench: Split | Factory | WorkbenchSpec | None = None, *, title: str | None = None, workbenches: Sequence[WorkbenchSpec] = (), window: Window = APP, port: int = DEFAULT_PORT, telemetry: str | Path | None = None, persist: str | Path | None = None, watch: Sequence[str | Path] = ()) -> None

Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

Parameters:

  • workbench (Split | Factory | WorkbenchSpec | None, default: None ) –

    The Split, factory or card to serve; None serves the launcher.

  • title (str | None, default: None ) –

    The header title; None keeps the card's own, or femforge workbench for a Split.

  • workbenches (Sequence[WorkbenchSpec], default: () ) –

    Cards the launcher lists beside the discovered ones, and the only cards a single workbench's title menu lists.

  • window (Window, default: APP ) –

    Where the workbench opens.

  • port (int, default: DEFAULT_PORT ) –

    The port the server listens on.

  • telemetry (str | Path | None, default: None ) –

    The telemetry root the panels read, exported as FEMFORGE_TELEMETRY; None keeps the environment's.

  • persist (str | Path | None, default: None ) –

    The directory layouts and opened cards persist under; None uses the per-user config directory.

  • watch (Sequence[str | Path], default: () ) –

    More directories whose .py saves rebuild a factory or card, beside its own roots.

Source code in packages/femforge-workbench/src/femforge_workbench/serve.py
def show(
    workbench: Split | Factory | WorkbenchSpec | None = None,
    *,
    title: str | None = None,
    workbenches: Sequence[WorkbenchSpec] = (),
    window: Window = Window.APP,
    port: int = DEFAULT_PORT,
    telemetry: str | Path | None = None,
    persist: str | Path | None = None,
    watch: Sequence[str | Path] = (),
) -> None:
    """Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

    Args:
        workbench: The Split, factory or card to serve; None serves the launcher.
        title: The header title; None keeps the card's own, or femforge workbench for a Split.
        workbenches: Cards the launcher lists beside the discovered ones, and the only cards a single workbench's title menu lists.
        window: Where the workbench opens.
        port: The port the server listens on.
        telemetry: The telemetry root the panels read, exported as FEMFORGE_TELEMETRY; None keeps the environment's.
        persist: The directory layouts and opened cards persist under; None uses the per-user config directory.
        watch: More directories whose .py saves rebuild a factory or card, beside its own roots.
    """
    if telemetry is not None:
        os.environ[TELEMETRY_ENV] = str(telemetry)

    store = LocalJsonStore(Path(persist).resolve()) if persist is not None else None
    if isinstance(workbench, Split):
        shown = title or DEFAULT_TITLE
        app = Workbench(shown.lower().replace(" ", "-"), title=shown, persist=store, catalogue=workbenches)
        app.add(workbench)
    else:
        spec = card(workbench)
        if spec.kind is WorkbenchKind.SCRIPT:
            log.bind(command=spec.command()).info("Running the script.")
            _ = subprocess.run(spec.command(), cwd=spec.cwd, check=False)
            return

        catalogue = [*discover(), *workbenches] if spec.name == LAUNCHER.name else workbenches
        app = build(replace(spec, title=title) if title else spec, catalogue=catalogue, watch=[Path(root).resolve() for root in watch], persist=store)

    app.run(port=port, window=window)

femforge_workbench.serve

Serves a workbench from a script or the command line: a dock, a catalogue card, or the launcher.

Type Aliases:

  • Factory –

    A module-level function building a dock Split or a Renderable.

Classes:

  • NotAWorkbench –

    A target that builds neither a dock Split nor a Renderable.

Functions:

  • show –

    Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

  • card –

    The spec to open: the launcher for None, a dock card named after a factory.

  • build –

    A shell holding the spec's dock, rebuilt on every save under its roots, or a viewer of its Renderable.

  • viewer –

    One scene panel that shows the spec's Renderable once the server runs, rebuilding it on every save under its roots.

Attributes:

  • DEFAULT_PORT –

    The port a workbench serves on unless told otherwise.

DEFAULT_PORT module-attribute

DEFAULT_PORT = 8080

The port a workbench serves on unless told otherwise.

Factory

Factory = Callable[[], Split | Renderable]

A module-level function building a dock Split or a Renderable.

NotAWorkbench

Bases: TypeError

A target that builds neither a dock Split nor a Renderable.

show

show(workbench: Split | Factory | WorkbenchSpec | None = None, *, title: str | None = None, workbenches: Sequence[WorkbenchSpec] = (), window: Window = APP, port: int = DEFAULT_PORT, telemetry: str | Path | None = None, persist: str | Path | None = None, watch: Sequence[str | Path] = ()) -> None

Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

Parameters:

  • workbench (Split | Factory | WorkbenchSpec | None, default: None ) –

    The Split, factory or card to serve; None serves the launcher.

  • title (str | None, default: None ) –

    The header title; None keeps the card's own, or femforge workbench for a Split.

  • workbenches (Sequence[WorkbenchSpec], default: () ) –

    Cards the launcher lists beside the discovered ones, and the only cards a single workbench's title menu lists.

  • window (Window, default: APP ) –

    Where the workbench opens.

  • port (int, default: DEFAULT_PORT ) –

    The port the server listens on.

  • telemetry (str | Path | None, default: None ) –

    The telemetry root the panels read, exported as FEMFORGE_TELEMETRY; None keeps the environment's.

  • persist (str | Path | None, default: None ) –

    The directory layouts and opened cards persist under; None uses the per-user config directory.

  • watch (Sequence[str | Path], default: () ) –

    More directories whose .py saves rebuild a factory or card, beside its own roots.

Source code in packages/femforge-workbench/src/femforge_workbench/serve.py
def show(
    workbench: Split | Factory | WorkbenchSpec | None = None,
    *,
    title: str | None = None,
    workbenches: Sequence[WorkbenchSpec] = (),
    window: Window = Window.APP,
    port: int = DEFAULT_PORT,
    telemetry: str | Path | None = None,
    persist: str | Path | None = None,
    watch: Sequence[str | Path] = (),
) -> None:
    """Serves a workbench until its window closes: a Split as built, a factory or card rebuilt on every save, or the launcher when none is given.

    Args:
        workbench: The Split, factory or card to serve; None serves the launcher.
        title: The header title; None keeps the card's own, or femforge workbench for a Split.
        workbenches: Cards the launcher lists beside the discovered ones, and the only cards a single workbench's title menu lists.
        window: Where the workbench opens.
        port: The port the server listens on.
        telemetry: The telemetry root the panels read, exported as FEMFORGE_TELEMETRY; None keeps the environment's.
        persist: The directory layouts and opened cards persist under; None uses the per-user config directory.
        watch: More directories whose .py saves rebuild a factory or card, beside its own roots.
    """
    if telemetry is not None:
        os.environ[TELEMETRY_ENV] = str(telemetry)

    store = LocalJsonStore(Path(persist).resolve()) if persist is not None else None
    if isinstance(workbench, Split):
        shown = title or DEFAULT_TITLE
        app = Workbench(shown.lower().replace(" ", "-"), title=shown, persist=store, catalogue=workbenches)
        app.add(workbench)
    else:
        spec = card(workbench)
        if spec.kind is WorkbenchKind.SCRIPT:
            log.bind(command=spec.command()).info("Running the script.")
            _ = subprocess.run(spec.command(), cwd=spec.cwd, check=False)
            return

        catalogue = [*discover(), *workbenches] if spec.name == LAUNCHER.name else workbenches
        app = build(replace(spec, title=title) if title else spec, catalogue=catalogue, watch=[Path(root).resolve() for root in watch], persist=store)

    app.run(port=port, window=window)

card

card(workbench: Factory | WorkbenchSpec | None) -> WorkbenchSpec

The spec to open: the launcher for None, a dock card named after a factory.

Source code in packages/femforge-workbench/src/femforge_workbench/serve.py
def card(workbench: Factory | WorkbenchSpec | None) -> WorkbenchSpec:
    """The spec to open: the launcher for None, a dock card named after a factory."""
    if workbench is None:
        return LAUNCHER

    if isinstance(workbench, WorkbenchSpec):
        return workbench

    return panel(workbench.__name__, workbench.__name__.replace("_", " "), target=workbench)

build

build(spec: WorkbenchSpec, *, catalogue: Sequence[WorkbenchSpec] | None = None, watch: Sequence[Path] = (), persist: UiStateStore | None = None) -> Workbench

A shell holding the spec's dock, rebuilt on every save under its roots, or a viewer of its Renderable.

Parameters:

  • spec (WorkbenchSpec) –

    The card to serve.

  • catalogue (Sequence[WorkbenchSpec] | None, default: None ) –

    The cards the title menu lists; None lists the discovered ones.

  • watch (Sequence[Path], default: () ) –

    More directories whose .py saves rebuild the card, beside its own roots.

  • persist (UiStateStore | None, default: None ) –

    Where layouts and opened cards persist; None uses JSON files under the per-user config directory.

Source code in packages/femforge-workbench/src/femforge_workbench/serve.py
def build(spec: WorkbenchSpec, *, catalogue: Sequence[WorkbenchSpec] | None = None, watch: Sequence[Path] = (), persist: UiStateStore | None = None) -> Workbench:
    """A shell holding the spec's dock, rebuilt on every save under its roots, or a viewer of its Renderable.

    Args:
        spec: The card to serve.
        catalogue: The cards the title menu lists; None lists the discovered ones.
        watch: More directories whose .py saves rebuild the card, beside its own roots.
        persist: Where layouts and opened cards persist; None uses JSON files under the per-user config directory.
    """
    app = Workbench(spec.name, title=spec.title, persist=persist, catalogue=catalogue)
    if spec.kind is WorkbenchKind.RENDERABLE:
        _ = viewer(app, spec, watch)
        return app

    factory = resolve(spec.target)
    if not callable(factory):
        raise NotAWorkbench(f"{spec.target} is not callable")

    root = factory()
    if isinstance(root, Split):
        app.add(root, spec=spec, watch=watch)
        for scene in (built for built in app.panels if isinstance(built, SceneViewerPanel)):
            scene.watch.extend(watch)

        return app

    if isinstance(root, Renderable):
        _ = viewer(app, replace(spec, kind=WorkbenchKind.RENDERABLE), watch)
        return app

    raise NotAWorkbench(f"{spec.target}() returned {type(root).__name__}; a Split or a Renderable is needed")

viewer

viewer(app: Workbench, spec: WorkbenchSpec, watch: Sequence[Path]) -> SceneViewerPanel

One scene panel that shows the spec's Renderable once the server runs, rebuilding it on every save under its roots.

Parameters:

  • app (Workbench) –

    The shell to add the scene panel to.

  • spec (WorkbenchSpec) –

    The card whose target builds the Renderable.

  • watch (Sequence[Path]) –

    More directories whose .py saves rebuild the Renderable, beside the spec's own roots.

Source code in packages/femforge-workbench/src/femforge_workbench/serve.py
def viewer(app: Workbench, spec: WorkbenchSpec, watch: Sequence[Path]) -> SceneViewerPanel:
    """One scene panel that shows the spec's Renderable once the server runs, rebuilding it on every save under its roots.

    Args:
        app: The shell to add the scene panel to.
        spec: The card whose target builds the Renderable.
        watch: More directories whose .py saves rebuild the Renderable, beside the spec's own roots.
    """

    def scene_panel(app: Workbench, id: str, title: str) -> SceneViewerPanel:
        return SceneViewerPanel(app, id, title, watch=watch)

    app.add(Split([Panel(spec.title, scene_panel, id=VIEWER_PANEL)]))
    scene = app.panel(VIEWER_PANEL, SceneViewerPanel)

    def stop(**_: object) -> None:
        scene.stop()

    app.when_ready(lambda: scene.show(spec))
    _ = app.ctrl.on_server_exited.add(stop)

    return scene

femforge_workbench.cli

The femforge-workbench command: parses its flags and shows the launcher, a catalogue entry or a module through serve.show.

Classes:

  • CliError –

    A user-facing failure: main prints the message and exits 2.

  • Options –

    The parsed command line.

Functions:

  • select –

    The spec to open: a catalogue entry by name, a module given as pkg.module[:attr] (main() -> Renderable when no attr), else the launcher.

  • build_parser –

    The femforge-workbench argument parser with every flag registered.

  • main –

    Parses the command line and shows the chosen workbench, or supervises the server under --dev.

CliError

Bases: Exception

A user-facing failure: main prints the message and exits 2.

Options dataclass

Options(name: str | None = None, module: str | None = None, app: bool = False, webview: bool = False, dev: bool = False, port: int = DEFAULT_PORT, watch: list[str] = list(), telemetry: str | None = None, persist: str | None = None)

The parsed command line.

Attributes:

  • window (Window) –

    The window the flags ask for: an app window, pywebview, else a browser tab.

window property

window: Window

The window the flags ask for: an app window, pywebview, else a browser tab.

select

select(name: str | None, module: str | None) -> WorkbenchSpec

The spec to open: a catalogue entry by name, a module given as pkg.module[:attr] (main() -> Renderable when no attr), else the launcher.

Source code in packages/femforge-workbench/src/femforge_workbench/cli.py
def select(name: str | None, module: str | None) -> WorkbenchSpec:
    """The spec to open: a catalogue entry by name, a module given as `pkg.module[:attr]` (main() -> Renderable when no attr), else the launcher."""
    if module:
        module_name, attr = split_target(module)
        if attr is None:
            return legacy(module_name)

        return panel(attr, attr.replace("_", " "), target=module, description=module)

    if not name:
        return LAUNCHER

    catalogue = discover()
    spec = next((spec for spec in catalogue if spec.name == name), None)
    if spec is None:
        names = ", ".join(spec.name for spec in catalogue) or "none"
        raise CliError(f"unknown workbench {name!r}; known: {names}")

    return spec

build_parser

build_parser() -> ArgumentParser

The femforge-workbench argument parser with every flag registered.

Source code in packages/femforge-workbench/src/femforge_workbench/cli.py
def build_parser() -> argparse.ArgumentParser:
    """The ``femforge-workbench`` argument parser with every flag registered."""
    parser = argparse.ArgumentParser(prog="femforge-workbench", description="Open a workbench in the trame shell.")
    _ = parser.add_argument("name", nargs="?", help="a catalogue entry, e.g. demo")
    _ = parser.add_argument("-m", "--module", help="pkg.module[:attr]: attr returns a dock Split or a Renderable; without attr, main() -> Renderable")
    _ = parser.add_argument("--app", action="store_true", help="open a Chrome/Edge app window and stop the server when it closes")
    _ = parser.add_argument("--webview", action="store_true", help="open in pywebview (needs the app extra)")
    _ = parser.add_argument(DEV_FLAG, action="store_true", help="run the server as a child restarted on every save under the shell package; the page reloads itself")
    _ = parser.add_argument("--port", type=int, default=DEFAULT_PORT)
    _ = parser.add_argument("--watch", action="append", default=[], metavar="DIR", help="extra directory to watch and purge when a workbench is rebuilt (repeatable)")
    _ = parser.add_argument("--telemetry", default=None, metavar="DIR", help=f"the telemetry root the browsers read (default: ${TELEMETRY_ENV})")
    _ = parser.add_argument("--persist", default=None, metavar="DIR", help="where layouts and choices are saved (default: the per-user config directory)")

    return parser

main

main(argv: Sequence[str] | None = None) -> None

Parses the command line and shows the chosen workbench, or supervises the server under --dev.

Source code in packages/femforge-workbench/src/femforge_workbench/cli.py
def main(argv: Sequence[str] | None = None) -> None:
    """Parses the command line and shows the chosen workbench, or supervises the server under --dev."""
    parser = build_parser()
    options, _ = parser.parse_known_args(argv, Options())

    setup_logging(level=LogLevel.INFO)

    if options.dev:
        supervise(CLI_MODULE, sys.argv[1:] if argv is None else argv)
        return

    try:
        spec = select(options.name, options.module)
        show(spec, window=options.window, port=options.port, telemetry=options.telemetry, persist=options.persist, watch=options.watch)
    except (CliError, NotAWorkbench, TelemetryRootMissing) as error:
        parser.exit(2, f"femforge-workbench: {error}\n")

femforge_workbench.app

The workbench shell: a React header over a dock whose panels are trame templates, with persisted layouts and a worker-thread marshal.

Classes:

  • WidgetLibrary –

    A trame widget package whose initialize() enables its client module on a server.

  • Cls –

    The shell's Tailwind compositions; the strings appear literally so the stylesheet build scans them.

  • Window –

    Where a served workbench opens.

  • Template –

    Templates rendered outside the panels.

  • StateVar –

    The shell's own state variables.

  • Hook –

    Controller slots panels fill: the launcher takes OPEN_WORKBENCH, the field viewer FIELD_OPEN.

  • OpenedKey –

    The fields of the persisted opened document: the cards open in the dock, in opening order.

  • CardKey –

    The fields of a catalogue card in the state.

  • Opened –

    A Split in the running dock: the card it serves (None for a bare Split), the panel it was anchored to and the header widgets it added.

  • Watch –

    A card rebuilt on save and the reloader watching its target's roots.

  • Reporting –

    Anything with a status line: the shell and its panels.

  • Workbench –

    One trame server with the header, the dock and the panels of a named workbench.

Functions:

  • check_unique –

    Rejects a panel id used twice in one Split.

  • guard –

    Prints a handler's exception with rich and reports it in the status line instead of letting it escape into wslink.

Attributes:

  • TITLE_VAR –

    The state variable trame's client shows as the page title.

  • PANEL_PARAM –

    The render-prop parameter carrying a dock panel's params to its server template.

  • WIDGET_LIBRARIES (tuple[WidgetLibrary, ...]) –

    Every widget library a panel may use: a client loads only the modules enabled before it connected, and a card can open such panels later.

  • QUIET_LIBRARIES –

    Stdlib loggers gated to WARNING by every shell, whichever entry point built it: the access log alone writes thousands of lines per page load.

TITLE_VAR module-attribute

TITLE_VAR = 'trame__title'

The state variable trame's client shows as the page title.

PANEL_PARAM module-attribute

PANEL_PARAM = 'panel'

The render-prop parameter carrying a dock panel's params to its server template.

WIDGET_LIBRARIES module-attribute

WIDGET_LIBRARIES: tuple[WidgetLibrary, ...] = (widgets, rca, vtklocal)

Every widget library a panel may use: a client loads only the modules enabled before it connected, and a card can open such panels later.

QUIET_LIBRARIES module-attribute

QUIET_LIBRARIES = ('trame_server', 'trame_client', 'trame_common', 'trame_rca', 'trame_vtklocal', 'wslink', 'aiohttp', 'aiohttp.access', 'watchdog', 'asyncio')

Stdlib loggers gated to WARNING by every shell, whichever entry point built it: the access log alone writes thousands of lines per page load.

WidgetLibrary

Bases: Protocol

A trame widget package whose initialize() enables its client module on a server.

Methods:

  • initialize –

    Enables the library's client module on server.

initialize

initialize(server: Server) -> None

Enables the library's client module on server.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def initialize(self, server: Server) -> None:
    """Enables the library's client module on `server`."""
    ...

Cls

Bases: StrEnum

The shell's Tailwind compositions; the strings appear literally so the stylesheet build scans them.

Window

Bases: StrEnum

Where a served workbench opens.

Attributes:

  • BROWSER –

    A tab in the default browser.

  • APP –

    A Chrome/Edge app window; closing it stops the server.

  • WEBVIEW –

    pywebview, from the app extra.

BROWSER class-attribute instance-attribute

BROWSER = 'browser'

A tab in the default browser.

APP class-attribute instance-attribute

APP = 'app'

A Chrome/Edge app window; closing it stops the server.

WEBVIEW class-attribute instance-attribute

WEBVIEW = 'webview'

pywebview, from the app extra.

Template

Bases: StrEnum

Templates rendered outside the panels.

StateVar

Bases: StrEnum

The shell's own state variables.

Hook

Bases: StrEnum

Controller slots panels fill: the launcher takes OPEN_WORKBENCH, the field viewer FIELD_OPEN.

OpenedKey

Bases: StrEnum

The fields of the persisted opened document: the cards open in the dock, in opening order.

CardKey

Bases: StrEnum

The fields of a catalogue card in the state.

Opened dataclass

Opened(root: Split, reference: str | None, name: str | None = None, header: tuple[object, ...] = ())

A Split in the running dock: the card it serves (None for a bare Split), the panel it was anchored to and the header widgets it added.

Watch dataclass

Watch(name: str, reloader: Reloader[Split])

A card rebuilt on save and the reloader watching its target's roots.

Reporting

Bases: Protocol

Anything with a status line: the shell and its panels.

Methods:

  • status –

    Writes message to the status line.

status

status(message: str) -> None

Writes message to the status line.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def status(self, message: str) -> None:
    """Writes `message` to the status line."""
    ...

Workbench

Workbench(name: str, *, palette: Palette | None = None, title: str = 'femforge workbench', server: Server | None = None, persist: UiStateStore | None = None, user: str | None = None, catalogue: Sequence[WorkbenchSpec] | None = None)

Bases: TrameApp

One trame server with the header, the dock and the panels of a named workbench.

Parameters:

  • name (str) –

    The workbench name, which names the trame server and scopes the persisted state.

  • palette (Palette | None, default: None ) –

    The colours the shell and its panels paint in; None loads the dist's dark palette.

  • title (str, default: 'femforge workbench' ) –

    The title the header and the browser tab show.

  • server (Server | None, default: None ) –

    The trame server to run on; None makes one named after name.

  • persist (UiStateStore | None, default: None ) –

    Where layouts and opened cards persist; None keeps them in JSON files under the per-user config directory.

  • user (str | None, default: None ) –

    The user the persisted state is scoped to; None takes the OS user.

  • catalogue (Sequence[WorkbenchSpec] | None, default: None ) –

    The cards the title menu and a launcher panel list; None lists the discovered ones.

Methods:

  • load_catalogue –

    The discovered specs, empty with a warning when an entry point is broken.

  • build_ui –

    The theme switch, the header and the dock, whose panel bodies are the server templates the placements name.

  • theme –

    The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too.

  • build_switcher –

    The title as a menu over the catalogue.

  • add –

    Builds every panel's component and template; the dock arranges them once the client reports ready.

  • open –

    Adds a Split's panels to the running dock, building those not built yet, and activates the first.

  • open_spec –

    Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it.

  • opened_first –

    The first panel id of the open card of that name, None when it is not open.

  • close –

    Takes an opened card out of the dock and the shell and stops rebuilding it on save.

  • close_all –

    Closes every opened card.

  • discard –

    Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten.

  • register –

    Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken).

  • retire –

    Disposes the built panels of those ids and drops them from the shell.

  • drop_header –

    Removes a Split's widgets from the header, those still there.

  • watch –

    Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

  • unwatch –

    Stops rebuilding a card on save.

  • rebuild –

    Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

  • replace_root –

    Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement.

  • replace_card –

    Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

  • swap –

    Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

  • find_spec –

    The catalogue card of that name, None when the catalogue has none.

  • restore_opened –

    Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split.

  • remember_opened –

    Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards.

  • prune_opened –

    Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last.

  • place –

    add_panel for every placement of a Split, its first panel within reference's group when given; ids already in the dock are left alone.

  • panel –

    The built component of a panel, checked to be of kind.

  • status –

    Writes the status line, flushing so it also lands from outside a handler.

  • marshal –

    Runs fn(*args) on the server loop from any thread.

  • when_ready –

    Calls fn on the server loop: now when the server runs, else once it is ready.

  • authored_layout –

    The dockview layout of the authored Split; RuntimeError before add.

  • on_dock_ready –

    Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap.

  • on_layout_changed –

    Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel.

  • on_panel_removed –

    Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again.

  • rebuild_closed –

    Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live.

  • on_panel_visibility –

    Records that the dock showed or hid a panel and tells that panel.

  • reset_layout –

    Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window.

  • open_workbench –

    Hands a catalogue selection to the open_workbench hook, or names the command that opens it.

  • on_server_ready –

    Takes the server loop and opens the app window when one was asked for.

  • on_server_exited –

    Stops every reloader and flushes the persisted state.

  • open_app_window –

    Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab.

  • install_break_handler –

    Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way.

  • on_break –

    Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks.

  • run –

    Serves until stopped, or until an app window closes.

Attributes:

  • authored (Opened | None) –

    The Split add() built, named after the spec it serves when it is to be rebuilt on save.

  • opened (dict[str, Opened]) –

    Splits opened into the running dock by their first panel id.

  • pending (set[str]) –

    Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close.

  • on_screen (set[str]) –

    Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here.

  • catalogue –

    The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own.

  • header –

    The header's slot for a workbench's own controls: build widgets inside with app.header: and flush_content() after a later change.

  • root (Split | None) –

    The authored Split, None before add.

  • panel_ids (list[str]) –

    The ids of the built panels, in build order.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def __init__(
    self,
    name: str,
    *,
    palette: Palette | None = None,
    title: str = "femforge workbench",
    server: Server | None = None,
    persist: UiStateStore | None = None,
    user: str | None = None,
    catalogue: Sequence[WorkbenchSpec] | None = None,
) -> None:
    """Builds the shell: the header, the dock and the shared state, before any panel is added.

    Args:
        name: The workbench name, which names the trame server and scopes the persisted state.
        palette: The colours the shell and its panels paint in; None loads the dist's dark palette.
        title: The title the header and the browser tab show.
        server: The trame server to run on; None makes one named after `name`.
        persist: Where layouts and opened cards persist; None keeps them in JSON files under the per-user config directory.
        user: The user the persisted state is scoped to; None takes the OS user.
        catalogue: The cards the title menu and a launcher panel list; None lists the discovered ones.
    """
    gate_libraries(dict.fromkeys(QUIET_LIBRARIES, LogLevel.WARNING))
    super().__init__(server=server if server is not None else name, client_type="react")

    self.name = name
    self.title = title
    self.palette = palette if palette is not None else Palette.load()
    self.palette.install()
    self.persist: UiStateStore = persist if persist is not None else LocalJsonStore()
    self.scope = (user if user is not None else default_user(), name)
    self.loop: asyncio.AbstractEventLoop | None = None
    self.window = Window.BROWSER
    self.authored: Opened | None = None
    """The Split add() built, named after the spec it serves when it is to be rebuilt on save."""
    self.panels: list[PanelComponent] = []
    self.opened: dict[str, Opened] = {}
    """Splits opened into the running dock by their first panel id."""
    self.watches: list[Watch] = []
    self.pending: set[str] = set()
    """Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close."""
    self.on_screen: set[str] = set()
    """Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here."""
    self.catalogue = list(catalogue) if catalogue is not None else self.load_catalogue()
    """The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own."""
    self.dock: Dock | None = None
    self.header = DivLayout(self.server, template_name=str(Template.HEADER), classes=str(Cls.HEADER_SLOT))
    """The header's slot for a workbench's own controls: build widgets inside `with app.header:` and flush_content() after a later change."""

    cards = [{CardKey.NAME: spec.name, CardKey.TITLE: spec.title, CardKey.DESCRIPTION: spec.description, CardKey.GROUP: spec.group} for spec in self.catalogue]
    self.state.update(
        {
            StateVar.STATUS: "ready",
            StateVar.TITLE: title,
            StateVar.DARK: self.palette.is_dark,
            StateVar.WORKBENCHES: cards,
            StateVar.OPENED: [],
            TITLE_VAR: title,
        }
    )

    for library in WIDGET_LIBRARIES:
        library.initialize(self.server)

    self.ui = self.build_ui()

authored instance-attribute

authored: Opened | None = None

The Split add() built, named after the spec it serves when it is to be rebuilt on save.

opened instance-attribute

opened: dict[str, Opened] = {}

Splits opened into the running dock by their first panel id.

pending instance-attribute

pending: set[str] = set()

Panel ids handed to the dock since a layout last showed them all; a layout still missing one is a placement in flight, not a close.

on_screen instance-attribute

on_screen: set[str] = set()

Panel ids the dock last reported visible; the dock reports a panel only when it mounts, so a panel rebuilt in place learns it from here.

catalogue instance-attribute

catalogue = list(catalogue) if catalogue is not None else self.load_catalogue()

The cards the title menu and a launcher panel list: the discovered ones unless the constructor was given its own.

header instance-attribute

header = DivLayout(self.server, template_name=str(Template.HEADER), classes=str(Cls.HEADER_SLOT))

The header's slot for a workbench's own controls: build widgets inside with app.header: and flush_content() after a later change.

root property

root: Split | None

The authored Split, None before add.

panel_ids property

panel_ids: list[str]

The ids of the built panels, in build order.

load_catalogue

load_catalogue() -> list[WorkbenchSpec]

The discovered specs, empty with a warning when an entry point is broken.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def load_catalogue(self) -> list[WorkbenchSpec]:
    """The discovered specs, empty with a warning when an entry point is broken."""
    try:
        return discover()
    except (TypeError, ValueError, ImportError):
        log.opt(exception=True).warning("Workbench catalogue unavailable.")
        return []

build_ui

build_ui() -> DivLayout

The theme switch, the header and the dock, whose panel bodies are the server templates the placements name.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def build_ui(self) -> DivLayout:
    """The theme switch, the header and the dock, whose panel bodies are the server templates the placements name."""
    with DivLayout(self.server, classes=str(Cls.ROOT)) as layout:
        _ = self.theme()
        if os.environ.get(DEV_ENV):
            client.Script(reload_script())

        with html.Header(classes=str(Cls.HEADER)):
            if self.catalogue:
                self.build_switcher()
            else:
                html.Span([react.Bind(StateVar.TITLE)], classes=str(Cls.TITLE))

            client.ServerTemplate(name=str(Template.HEADER))
            with react.If(f"{StateVar.OPENED}.length"):
                Button("Close All", variant="destructive", size="sm", title="Close every opened card", on_click=react.Callback(self.close_all))

            with Button(
                size="icon-sm",
                variant="ghost",
                title="Reset the layout",
                aria_label="Reset the layout",
                on_click=react.Callback(self.reset_layout),
            ):
                Icon("layout-grid")

            html.Span([react.Bind(StateVar.STATUS)], classes=str(Cls.STATUS))

        with html.Main(classes=str(Cls.MAIN)):
            with react.Slot(params=[PANEL_PARAM]) as render_panel:
                client.ServerTemplate(name=react.Bind(f"{PANEL_PARAM}.{dock.Key.TEMPLATE_NAME}"))

            self.dock = Dock(
                render_panel=render_panel,
                dark=react.Bind(StateVar.DARK),
                popout_url=asset_url(POPOUT_PAGE),
                floating_group_bounds="boundedWithinViewport",
                on_ready=react.Callback(self.on_dock_ready),
                on_layout_change=react.Callback(self.on_layout_changed, args="[$event]"),
                on_visibility=react.Callback(self.on_panel_visibility, args="[$event]"),
                on_panel_removed=react.Callback(self.on_panel_removed, args="[$event]"),
            )

    return layout

theme

theme() -> Theme

The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def theme(self) -> Theme:
    """The scheme switch bound to the dark state; mounted in the root and in every panel, so a standalone ?ui= page and a popout paint it too."""
    return Theme(dark=react.Bind(StateVar.DARK))

build_switcher

build_switcher() -> None

The title as a menu over the catalogue.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def build_switcher(self) -> None:
    """The title as a menu over the catalogue."""
    with DropdownMenu():
        with DropdownMenuTrigger(as_child=True), Button(variant="ghost", size="sm", classes=str(Cls.TITLE)):
            html.Span([react.Bind(StateVar.TITLE)])
            Icon("chevron-down")

        with DropdownMenuContent(align="start"), react.For(items=StateVar.WORKBENCHES, name="card"):
            with DropdownMenuItem(on_select=react.Callback(self.open_workbench, args=f"[card.{CardKey.NAME}]"), classes=str(Cls.MENU_ITEM)):
                html.Span([react.Bind(f"card.{CardKey.TITLE}")])
                html.Span([react.Bind(f"card.{CardKey.DESCRIPTION}")], classes=str(Cls.MENU_DESCRIPTION))

add

add(root: Split, *, spec: WorkbenchSpec | None = None, watch: Sequence[Path] = ()) -> None

Builds every panel's component and template; the dock arranges them once the client reports ready.

Parameters:

  • root (Split) –

    The Split to serve.

  • spec (WorkbenchSpec | None, default: None ) –

    The card the Split was built from; given, the Split is rebuilt on every save.

  • watch (Sequence[Path], default: () ) –

    More directories whose .py saves rebuild the Split, beside the spec's own roots.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def add(self, root: dock.Split, *, spec: WorkbenchSpec | None = None, watch: Sequence[Path] = ()) -> None:
    """Builds every panel's component and template; the dock arranges them once the client reports ready.

    Args:
        root: The Split to serve.
        spec: The card the Split was built from; given, the Split is rebuilt on every save.
        watch: More directories whose .py saves rebuild the Split, beside the spec's own roots.
    """
    leaves = dock.panels(root)
    ids = [leaf.panel_id for leaf in leaves]
    check_unique(ids)

    header_before = len(self.header.root.children)
    self.panels = []
    _ = self.register(leaves)
    self.authored = Opened(root, None, spec.name if spec is not None else None, tuple(self.header.root.children[header_before:]))

    log.bind(workbench=self.name, panels=ids).info("Panels built.")

    self.restore_opened()
    if spec is not None:
        self.watch(spec, extra=watch)

open

open(root: Split, *, reference: str | None = None, name: str | None = None) -> None

Adds a Split's panels to the running dock, building those not built yet, and activates the first.

Parameters:

  • root (Split) –

    The Split whose panels to add.

  • reference (str | None, default: None ) –

    The panel whose group the first panel joins; None lets the dock place it.

  • name (str | None, default: None ) –

    The card the Split comes from, remembered so the next start reopens it; None for a bare Split.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open(self, root: dock.Split, *, reference: str | None = None, name: str | None = None) -> None:
    """Adds a Split's panels to the running dock, building those not built yet, and activates the first.

    Args:
        root: The Split whose panels to add.
        reference: The panel whose group the first panel joins; None lets the dock place it.
        name: The card the Split comes from, remembered so the next start reopens it; None for a bare Split.
    """
    leaves = dock.panels(root)
    ids = [leaf.panel_id for leaf in leaves]
    check_unique(ids)

    known = self.panel_ids
    fresh = [leaf for leaf in leaves if leaf.panel_id not in known]
    header_before = len(self.header.root.children)
    _ = self.register(fresh)
    self.header.flush_content()
    self.opened[ids[0]] = Opened(root, reference, name, tuple(self.header.root.children[header_before:]))

    self.place(root, reference)
    if self.dock is not None:
        self.dock.active_panel(ids[0])

    if name is not None:
        self.remember_opened()

    log.bind(workbench=self.name, panels=ids, built=[leaf.panel_id for leaf in fresh]).info("Panels opened.")

open_spec

open_spec(spec: WorkbenchSpec, *, reference: str | None = None) -> None

Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open_spec(self, spec: WorkbenchSpec, *, reference: str | None = None) -> None:
    """Opens a dock card: its target builds the Split, remembered under the card's name so the next start rebuilds it."""
    if spec.kind is not WorkbenchKind.PANEL:
        raise ValueError(f"{spec.name} is not a dock")

    factory = resolve(spec.target)
    if not callable(factory):
        raise TypeError(f"{spec.target} is not callable")

    root = factory()
    if not isinstance(root, dock.Split):
        raise TypeError(f"{spec.target}() returned {type(root).__name__}, not a Split")

    self.open(root, reference=reference, name=spec.name)
    self.watch(spec, reference=reference)

opened_first

opened_first(name: str) -> str | None

The first panel id of the open card of that name, None when it is not open.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def opened_first(self, name: str) -> str | None:
    """The first panel id of the open card of that name, None when it is not open."""
    return next((first for first, opened in self.opened.items() if opened.name == name), None)

close

close(name: str) -> None

Takes an opened card out of the dock and the shell and stops rebuilding it on save.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def close(self, name: str) -> None:
    """Takes an opened card out of the dock and the shell and stops rebuilding it on save."""
    self.discard(name)
    self.unwatch(name)

close_all

close_all() -> None

Closes every opened card.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def close_all(self) -> None:
    """Closes every opened card."""
    names = [opened.name for opened in list(self.opened.values()) if opened.name is not None]
    for name in names:
        self.close(name)

    self.status(f"{len(names)} cards closed")

discard

discard(name: str) -> None

Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def discard(self, name: str) -> None:
    """Takes an opened card out of the dock and the shell: its panels are removed and disposed, its header widgets dropped, the card forgotten."""
    first = self.opened_first(name)
    if first is None:
        raise KeyError(f"{name} is not open")

    opened = self.opened.pop(first)
    ids = dock.panel_ids(opened.root)
    self.pending.difference_update(ids)
    self.on_screen.difference_update(ids)
    if self.dock is not None:
        for panel_id in ids:
            self.dock.remove_panel(panel_id)

    self.retire(ids)
    self.drop_header(opened.header)
    self.header.flush_content()
    self.remember_opened()
    log.bind(workbench=self.name, card=name, panels=ids).info("Card closed.")

register

register(leaves: Sequence[Panel]) -> list[PanelComponent]

Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken).

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def register(self, leaves: Sequence[dock.Panel]) -> list[PanelComponent]:
    """Builds each leaf and registers it before the next is built, so a panel can see the ones already standing (a second scene panel must know a mirror is taken)."""
    built: list[PanelComponent] = []
    try:
        for leaf in leaves:
            panel = leaf.build(self, leaf.panel_id, leaf.name)
            built.append(panel)
            self.panels.append(panel)
    except Exception:
        for panel in built:
            panel.dispose()
            self.panels.remove(panel)

        raise

    return built

retire

retire(ids: Sequence[str]) -> None

Disposes the built panels of those ids and drops them from the shell.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def retire(self, ids: Sequence[str]) -> None:
    """Disposes the built panels of those ids and drops them from the shell."""
    for panel in [panel for panel in self.panels if panel.id in ids]:
        panel.dispose()
        self.panels.remove(panel)

drop_header

drop_header(widgets: Sequence[object]) -> None

Removes a Split's widgets from the header, those still there.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def drop_header(self, widgets: Sequence[object]) -> None:
    """Removes a Split's widgets from the header, those still there."""
    for widget in widgets:
        if widget in self.header.root.children:
            self.header.root.children.remove(widget)

watch

watch(spec: WorkbenchSpec, *, reference: str | None = None, extra: Sequence[Path] = ()) -> None

Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

Parameters:

  • spec (WorkbenchSpec) –

    The card whose target builds the Split.

  • reference (str | None, default: None ) –

    The panel whose group the rebuilt card joins when it is no longer open.

  • extra (Sequence[Path], default: () ) –

    More directories to watch, beside the femforge core, the workbench package and the spec's own source root.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def watch(self, spec: WorkbenchSpec, *, reference: str | None = None, extra: Sequence[Path] = ()) -> None:
    """Rebuilds the spec's Split after every .py save under its roots once the server runs; a spec of the shell core is left to the dev supervisor.

    Args:
        spec: The card whose target builds the Split.
        reference: The panel whose group the rebuilt card joins when it is no longer open.
        extra: More directories to watch, beside the femforge core, the workbench package and the spec's own source root.
    """
    self.unwatch(spec.name)
    if persistent(spec.module):
        return

    roots = [*default_roots(spec.module), *extra]

    def apply(root: dock.Split, seconds: float) -> None:
        self.rebuild(spec.name, root, reference=reference, seconds=seconds)

    def failed(error: Exception) -> None:
        self.status(f"{type(error).__name__}: {error}")

    reloader = Reloader(spec.target, roots, apply, kind=dock.Split, marshal=self.marshal, failed=failed)
    watch = Watch(spec.name, reloader)
    self.watches.append(watch)
    if self.loop is not None:
        reloader.start()
        return

    def start(**_: object) -> None:
        if watch in self.watches:
            reloader.start()

    _ = self.ctrl.on_server_ready.add(start)

unwatch

unwatch(name: str) -> None

Stops rebuilding a card on save.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def unwatch(self, name: str) -> None:
    """Stops rebuilding a card on save."""
    for watch in [watch for watch in self.watches if watch.name == name]:
        watch.reloader.stop()
        self.watches.remove(watch)

rebuild

rebuild(name: str, root: Split, *, reference: str | None = None, seconds: float = 0.0) -> None

Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

Parameters:

  • name (str) –

    The card to rebuild.

  • root (Split) –

    The fresh build of its Split.

  • reference (str | None, default: None ) –

    The panel whose group the card joins when it is no longer open.

  • seconds (float, default: 0.0 ) –

    The seconds the build took, added to the rebuild time the status line reports.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def rebuild(self, name: str, root: dock.Split, *, reference: str | None = None, seconds: float = 0.0) -> None:
    """Swaps a card's panels for a fresh build under the same ids, the dock untouched; other ids close and reopen the card, and a card no longer open reopens.

    Args:
        name: The card to rebuild.
        root: The fresh build of its Split.
        reference: The panel whose group the card joins when it is no longer open.
        seconds: The seconds the build took, added to the rebuild time the status line reports.
    """
    started = time.perf_counter()
    # Delivered through call_soon_threadsafe, outside any trame handler, so the state transaction is explicit for the flush.
    with self.state:
        if self.authored is not None and self.authored.name == name:
            self.replace_root(root)
        else:
            self.replace_card(name, root, reference)

    elapsed = seconds + time.perf_counter() - started
    log.bind(workbench=self.name, card=name, seconds=round(elapsed, 2)).info("Card rebuilt.")
    self.status(f"rebuilt {name} in {elapsed:.1f} s")

replace_root

replace_root(root: Split) -> None

Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def replace_root(self, root: dock.Split) -> None:
    """Rebuilds the authored Split in place; changed ids re-lay the dock out from the authored arrangement."""
    authored = self.authored
    if authored is None:
        raise RuntimeError("add() a Split before rebuilding it")

    retired = dock.panel_ids(authored.root)
    header = self.swap(root, retired, authored.header)
    self.authored = replace(authored, root=root, header=header)
    if dock.panel_ids(root) != retired:
        self.reset_layout()

replace_card

replace_card(name: str, root: Split, reference: str | None) -> None

Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

Parameters:

  • name (str) –

    The card to rebuild.

  • root (Split) –

    The fresh build of its Split.

  • reference (str | None) –

    The panel whose group the card joins when it is not open.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def replace_card(self, name: str, root: dock.Split, reference: str | None) -> None:
    """Rebuilds an open card in place; changed ids close and reopen it, and a card not open opens.

    Args:
        name: The card to rebuild.
        root: The fresh build of its Split.
        reference: The panel whose group the card joins when it is not open.
    """
    first = self.opened_first(name)
    if first is None:
        self.open(root, reference=reference, name=name)
        return

    opened = self.opened[first]
    retired = dock.panel_ids(opened.root)
    if dock.panel_ids(root) != retired:
        self.discard(name)
        self.open(root, reference=opened.reference, name=name)
        return

    header = self.swap(root, retired, opened.header)
    self.opened[first] = replace(opened, root=root, header=header)

swap

swap(root: Split, retired: Sequence[str], header: Sequence[object]) -> tuple[object, ...]

Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

Parameters:

  • root (Split) –

    The Split whose leaves to build.

  • retired (Sequence[str]) –

    The ids of the panels to dispose first.

  • header (Sequence[object]) –

    The header widgets the retired panels added, dropped with them.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def swap(self, root: dock.Split, retired: Sequence[str], header: Sequence[object]) -> tuple[object, ...]:
    """Disposes the retired panels, then builds the Split's leaves, whose templates overwrite those of the same ids; returns the header widgets the build added.

    Args:
        root: The Split whose leaves to build.
        retired: The ids of the panels to dispose first.
        header: The header widgets the retired panels added, dropped with them.
    """
    leaves = dock.panels(root)
    check_unique([leaf.panel_id for leaf in leaves])
    previous = [panel for panel in self.panels if panel.id in retired]
    self.retire(retired)
    self.drop_header(header)
    header_before = len(self.header.root.children)

    try:
        built = self.register(leaves)
    except Exception:
        del self.header.root.children[header_before:]
        self.header.flush_content()
        raise

    for panel in built:
        predecessor = next((old for old in previous if old.id == panel.id), None)
        if predecessor is not None:
            panel.succeed(predecessor)

        if panel.id in self.on_screen:
            panel.on_visible(True)

    self.header.flush_content()

    return tuple(self.header.root.children[header_before:])

find_spec

find_spec(name: str) -> WorkbenchSpec | None

The catalogue card of that name, None when the catalogue has none.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def find_spec(self, name: str) -> WorkbenchSpec | None:
    """The catalogue card of that name, None when the catalogue has none."""
    return next((spec for spec in self.catalogue if spec.name == name), None)

restore_opened

restore_opened() -> None

Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def restore_opened(self) -> None:
    """Reopens the cards the last session left in the dock, dropping any whose card is gone or whose target no longer builds a Split."""
    saved = self.persist.load(self.scope, StateKey.OPENED) or {}
    raw = saved.get(OpenedKey.ENTRIES)
    entries = cast("list[object]", raw) if isinstance(raw, list) else []

    for entry in entries:
        record = cast("dict[str, object]", entry) if isinstance(entry, dict) else {}
        name = str(record.get(OpenedKey.NAME))
        reference = record.get(OpenedKey.REFERENCE)
        spec = self.find_spec(name)
        if spec is None or spec.kind is not WorkbenchKind.PANEL:
            log.bind(workbench=self.name, card=name).warning("Opened card is no longer a dock in the catalogue.")
            continue

        anchor = reference if isinstance(reference, str) and reference in self.panel_ids else None
        try:
            self.open_spec(spec, reference=anchor)
        except Exception:  # noqa: BLE001 - a card that fails to build costs its panels, never the start
            log.bind(workbench=self.name, card=name).opt(exception=True).warning("Opened card failed to rebuild.")

    self.remember_opened()

remember_opened

remember_opened() -> None

Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def remember_opened(self) -> None:
    """Persists the cards open in the dock, in opening order, and names them in the state for the launcher's cards."""
    cards = [opened for opened in self.opened.values() if opened.name is not None]
    entries = [{str(OpenedKey.NAME): opened.name, str(OpenedKey.REFERENCE): opened.reference} for opened in cards]

    self.persist.save(self.scope, StateKey.OPENED, {str(OpenedKey.ENTRIES): entries})
    with self.state:
        self.state[StateVar.OPENED] = [opened.name for opened in cards]

prune_opened

prune_opened(layout: Mapping[str, object]) -> None

Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def prune_opened(self, layout: Mapping[str, object]) -> None:
    """Forgets an opened Split whose first panel is missing from a settled layout, one that shows every panel handed to the dock since the last."""
    panels = layout.get(dock.Key.PANELS)
    present = set(cast("Mapping[str, object]", panels).keys()) if isinstance(panels, Mapping) else set[str]()
    if self.pending:
        if not self.pending <= present:
            return

        self.pending.clear()

    closed = [first for first in self.opened if first not in present]
    if not closed:
        return

    for first in closed:
        del self.opened[first]

    self.remember_opened()
    log.bind(workbench=self.name, panels=closed).info("Opened panels closed.")

place

place(root: Split, reference: str | None = None) -> None

add_panel for every placement of a Split, its first panel within reference's group when given; ids already in the dock are left alone.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def place(self, root: dock.Split, reference: str | None = None) -> None:
    """add_panel for every placement of a Split, its first panel within `reference`'s group when given; ids already in the dock are left alone."""
    if self.dock is None:
        return

    for placement in dock.placements(root):
        anchored = replace(placement, reference=reference, direction=dock.PlacementDirection.WITHIN) if placement.reference is None and reference is not None else placement
        self.dock.add_panel(anchored.panel.panel_id, anchored.panel.name, dock.template_name(anchored.panel.panel_id), **anchored.add_on())
        self.pending.add(anchored.panel.panel_id)

panel

panel(panel_id: str, kind: type[P]) -> P

The built component of a panel, checked to be of kind.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def panel[P: PanelComponent](self, panel_id: str, kind: type[P]) -> P:
    """The built component of a panel, checked to be of `kind`."""
    found = next((panel for panel in self.panels if panel.id == panel_id), None)
    if not isinstance(found, kind):
        raise KeyError(f"no {kind.__name__} with id {panel_id!r}")

    return found

status

status(message: str) -> None

Writes the status line, flushing so it also lands from outside a handler.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def status(self, message: str) -> None:
    """Writes the status line, flushing so it also lands from outside a handler."""
    with self.state:
        self.state[StateVar.STATUS] = message

marshal

marshal(fn: Callable[..., object], *args: object) -> None

Runs fn(*args) on the server loop from any thread.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def marshal(self, fn: Callable[..., object], *args: object) -> None:
    """Runs `fn(*args)` on the server loop from any thread."""
    if self.loop is None:
        raise RuntimeError("the server is not running")

    _ = self.loop.call_soon_threadsafe(fn, *args)

when_ready

when_ready(fn: Callable[[], None]) -> None

Calls fn on the server loop: now when the server runs, else once it is ready.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def when_ready(self, fn: Callable[[], None]) -> None:
    """Calls `fn` on the server loop: now when the server runs, else once it is ready."""
    if self.loop is not None:
        fn()
        return

    def ready(**_: object) -> None:
        fn()

    _ = self.ctrl.on_server_ready.add(ready)

authored_layout

authored_layout() -> Layout

The dockview layout of the authored Split; RuntimeError before add.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def authored_layout(self) -> dock.Layout:
    """The dockview layout of the authored Split; RuntimeError before `add`."""
    if self.root is None:
        raise RuntimeError("add() a Split before the dock is ready")

    return dock.layout_json(self.root)

on_dock_ready

on_dock_ready() -> None

Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_dock_ready(self) -> None:
    """Restores the saved arrangement when it names the authored panels, else applies the authored one; add_panel then fills any gap."""
    if self.dock is None or self.root is None:
        return

    saved = dock.restorable(self.persist.load(self.scope, StateKey.LAYOUT), self.panel_ids)
    self.dock.restore_layout(saved if saved is not None else self.authored_layout())

    self.place(self.root)
    for opened in self.opened.values():
        self.place(opened.root, opened.reference)

    log.bind(workbench=self.name, restored=saved is not None).info("Dock ready.")

on_layout_changed

on_layout_changed(layout: Mapping[str, object]) -> None

Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_layout_changed(self, layout: Mapping[str, object]) -> None:
    """Persists the layout the dock reports, forgets the cards it no longer shows and tells every panel."""
    self.persist.save(self.scope, StateKey.LAYOUT, layout)
    self.prune_opened(layout)

    for panel in self.panels:
        panel.on_layout_changed()

on_panel_removed

on_panel_removed(event: Mapping[str, object]) -> None

Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_panel_removed(self, event: Mapping[str, object]) -> None:
    """Disposes the panel whose tab the user closed; the dock reports only a user close here, never its own teardown, and Reset layout builds a closed panel again."""
    panel_id = str(event.get("id"))
    if panel_id not in self.panel_ids:
        return

    self.pending.discard(panel_id)
    self.on_screen.discard(panel_id)
    self.retire([panel_id])
    log.bind(workbench=self.name, panel=panel_id).info("Panel closed.")

rebuild_closed

rebuild_closed() -> None

Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def rebuild_closed(self) -> None:
    """Rebuilds every card a closed tab left incomplete, header widgets included, so an arrangement naming its panels finds them live."""
    cards: list[tuple[str | None, Opened]] = [(first, opened) for first, opened in self.opened.items()]
    if self.authored is not None:
        cards.insert(0, (None, self.authored))

    for first, card in cards:
        ids = dock.panel_ids(card.root)
        if all(panel_id in self.panel_ids for panel_id in ids):
            continue

        rebuilt = replace(card, header=self.swap(card.root, ids, card.header))
        if first is None:
            self.authored = rebuilt
        else:
            self.opened[first] = rebuilt

        log.bind(workbench=self.name, panels=ids).info("Closed panels rebuilt.")

on_panel_visibility

on_panel_visibility(change: Mapping[str, object]) -> None

Records that the dock showed or hid a panel and tells that panel.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def on_panel_visibility(self, change: Mapping[str, object]) -> None:
    """Records that the dock showed or hid a panel and tells that panel."""
    panel_id = str(change.get("id"))
    visible = bool(change.get("visible"))
    if visible:
        self.on_screen.add(panel_id)
    else:
        self.on_screen.discard(panel_id)

    for panel in self.panels:
        if panel.id == panel_id:
            panel.on_visible(visible)

reset_layout

reset_layout() -> None

Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def reset_layout(self) -> None:
    """Rebuilds the authored arrangement, which also recovers a group orphaned by a killed popout window."""
    if self.dock is None:
        return

    self.rebuild_closed()
    self.dock.restore_layout(self.authored_layout())
    for opened in self.opened.values():
        self.place(opened.root, opened.reference)

    self.status("layout reset")

open_workbench

open_workbench(name: str) -> None

Hands a catalogue selection to the open_workbench hook, or names the command that opens it.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@guard
def open_workbench(self, name: str) -> None:
    """Hands a catalogue selection to the open_workbench hook, or names the command that opens it."""
    spec = self.find_spec(name)
    if spec is None:
        self.status(f"unknown workbench {name}")
        return

    if spec.name == self.name:
        self.status(f"{spec.title} is open")
        return

    hook = self.ctrl[Hook.OPEN_WORKBENCH]
    if hook.exists():
        _ = hook(spec)
        return

    self.status(f"open with: {' '.join(spec.command())}")
    log.bind(workbench=spec.name, command=spec.command()).info("Workbench selected.")

on_server_ready

on_server_ready(**_: object) -> None

Takes the server loop and opens the app window when one was asked for.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@life_cycle.server_ready
def on_server_ready(self, **_: object) -> None:
    """Takes the server loop and opens the app window when one was asked for."""
    self.loop = asyncio.get_running_loop()
    if self.window is Window.APP:
        self.open_app_window()

on_server_exited

on_server_exited(**_: object) -> None

Stops every reloader and flushes the persisted state.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
@life_cycle.server_exited
def on_server_exited(self, **_: object) -> None:
    """Stops every reloader and flushes the persisted state."""
    for watch in self.watches:
        watch.reloader.stop()

    self.persist.flush()

open_app_window

open_app_window() -> None

Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def open_app_window(self) -> None:
    """Chrome (Edge fallback) as an app window on its own profile; the server stops when the window process exits. Without either, a system browser tab."""
    url = f"http://localhost:{self.server.port}/"
    browser = next((path for path in BROWSERS if path.exists()), None)
    if browser is None:
        log.bind(url=url).warning("No Chrome or Edge found; opening the system browser.")
        _ = webbrowser.open(url)
        return

    profile = Path(platformdirs.user_cache_dir(APP_NAME, appauthor=False)) / APP_PROFILE
    profile.mkdir(parents=True, exist_ok=True)
    command = [
        str(browser),
        f"--app={url}",
        f"--user-data-dir={profile}",
        "--no-first-run",
        "--no-default-browser-check",
        f"--window-size={APP_WINDOW_SIZE}",
    ]
    process = subprocess.Popen(command)
    log.bind(browser=browser.name, pid=process.pid).info("App window opened.")

    def stop_with_window() -> None:
        code = process.wait()
        log.bind(returncode=code).info("App window closed; stopping the server.")
        if self.loop is not None:
            _ = asyncio.run_coroutine_threadsafe(self.server.stop(), self.loop)

    threading.Thread(target=stop_with_window, name="app-window", daemon=True).start()

install_break_handler

install_break_handler() -> None

Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def install_break_handler(self) -> None:
    """Makes a break or termination signal a clean stop, so the exit hooks flush the persisted state; the --dev supervisor stops its child this way."""
    _ = signal.signal(signal.SIGTERM, self.on_break)
    if sys.platform == "win32":
        _ = signal.signal(signal.SIGBREAK, self.on_break)

on_break

on_break(signum: int, frame: FrameType | None) -> None

Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def on_break(self, signum: int, frame: FrameType | None) -> None:
    """Schedules the server's stop on its loop, which ends the main loop and runs the exit hooks."""
    loop = self.loop
    if loop is None:
        return

    def stop() -> None:
        _ = asyncio.ensure_future(self.server.stop())

    loop.call_soon_threadsafe(stop)

run

run(*, port: int = 8080, window: Window = BROWSER) -> None

Serves until stopped, or until an app window closes.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def run(self, *, port: int = 8080, window: Window = Window.BROWSER) -> None:
    """Serves until stopped, or until an app window closes."""
    self.window = window
    # trame's own --app flag means pywebview; the app window is ours.
    sys.argv = [arg for arg in sys.argv if arg != APP_FLAG]
    self.install_break_handler()

    log.bind(workbench=self.name, port=port, window=str(window)).info("Starting the workbench.")
    _ = self.server.start(port=port, exec_mode="desktop" if window is Window.WEBVIEW else "main", open_browser=window is Window.BROWSER, timeout=0)

check_unique

check_unique(ids: Sequence[str]) -> None

Rejects a panel id used twice in one Split.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def check_unique(ids: Sequence[str]) -> None:
    """Rejects a panel id used twice in one Split."""
    repeated = sorted({panel_id for panel_id in ids if ids.count(panel_id) > 1})
    if repeated:
        raise ValueError(f"panel ids repeat: {repeated}")

guard

guard(handler: Callable[Concatenate[S, P], None]) -> Callable[Concatenate[S, P], None]

Prints a handler's exception with rich and reports it in the status line instead of letting it escape into wslink.

Source code in packages/femforge-workbench/src/femforge_workbench/app.py
def guard[S: Reporting, **P](handler: Callable[Concatenate[S, P], None]) -> Callable[Concatenate[S, P], None]:
    """Prints a handler's exception with rich and reports it in the status line instead of letting it escape into wslink."""

    @wraps(handler)
    def guarded(self: S, *args: P.args, **kwargs: P.kwargs) -> None:
        try:
            handler(self, *args, **kwargs)
        except Exception as error:
            get_console().print_exception()
            self.status(f"{type(error).__name__}: {error}")

    return guarded

femforge_workbench.launcher

The built-in launcher dock: the catalogue cards on the left, the scene, the outputs and the log as tabs on the right.

Functions:

workbench

workbench() -> Split

The launcher dock.

Source code in packages/femforge-workbench/src/femforge_workbench/launcher.py
def workbench() -> Split:
    """The launcher dock."""

    def launcher(app: "Workbench", id: str, title: str) -> LauncherPanel:
        return LauncherPanel(app, id, title)

    def scene(app: "Workbench", id: str, title: str) -> SceneViewerPanel:
        return SceneViewerPanel(app, id, title)

    def outputs(app: "Workbench", id: str, title: str) -> WorkbenchOutputsPanel:
        return WorkbenchOutputsPanel(app, id, title, catalogue=app.catalogue)

    def log(app: "Workbench", id: str, title: str) -> LogPanel:
        return LogPanel(app, id, title)

    viewers = Tabs(
        [
            Panel("Scene", scene, id=LauncherPanelId.SCENE),
            Panel("Outputs", outputs, id=LauncherPanelId.OUTPUTS),
            Panel("Log", log, id=LauncherPanelId.LOG),
        ],
        weight=VIEWERS_WEIGHT,
    )

    return Split([Panel("Launcher", launcher, id=LauncherPanelId.LAUNCHER, weight=CARDS_WEIGHT), viewers])

femforge_workbench.demo

The built-in demo workbench: a sphere in the scene, a line figure and a grouped table, exercising the shell without lab data.

Classes:

  • DemoPanel –

    The demo's panel ids.

  • DemoVar –

    The demo's state variables.

  • Sphere –

    A Renderable: a faceted sphere beside a smaller smooth one, painted from the palette.

Functions:

  • signal_figure –

    Two damped sinusoids named for export through layout.meta.title.

  • rows_frame –

    A hundred rows over five series with a group label carrying the count and the peak y.

  • series_rows –

    The grouped rows of one series, or all of them.

  • workbench –

    The demo dock: the sphere on the left, the figure over the table on the right, with a series filter in the app bar.

Attributes:

  • SPEC –

    The catalogue card the workbench entry point exposes.

SPEC module-attribute

SPEC = DEMO

The catalogue card the workbench entry point exposes.

DemoPanel

Bases: StrEnum

The demo's panel ids.

DemoVar

Bases: StrEnum

The demo's state variables.

Sphere

Sphere(palette: Palette)

A Renderable: a faceted sphere beside a smaller smooth one, painted from the palette.

Methods:

  • render –

    Draws the two spheres into plotter and fits the camera.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def __init__(self, palette: Palette) -> None:
    """Takes the palette the spheres are painted from."""
    self.palette = palette

render

render(plotter: Plotter) -> None

Draws the two spheres into plotter and fits the camera.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def render(self, plotter: pv.Plotter) -> None:
    """Draws the two spheres into `plotter` and fits the camera."""
    big, small = self.palette.colorway(2)
    plotter.clear()
    _ = plotter.add_mesh(pv.Sphere(radius=1.0), color=big, show_edges=True, edge_color=self.palette.grid)  # pyright: ignore[reportUnknownMemberType]  # pyvista's add_mesh and Sphere are partially untyped
    _ = plotter.add_mesh(pv.Sphere(radius=0.4, center=(1.6, 0.0, 0.0)), color=small, smooth_shading=True)  # pyright: ignore[reportUnknownMemberType]
    plotter.renderer.ResetCamera()

signal_figure

signal_figure() -> Figure

Two damped sinusoids named for export through layout.meta.title.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def signal_figure() -> go.Figure:
    """Two damped sinusoids named for export through layout.meta.title."""
    x = np.linspace(0.0, 4.0 * np.pi, SAMPLES)
    damping = np.exp(-x / 8.0)
    figure = go.Figure([go.Scatter(x=x, y=np.sin(x) * damping, name="sin"), go.Scatter(x=x, y=np.cos(x) * damping, name="cos")])
    _ = figure.update_layout(meta={"title": "Damped signal"}, uirevision="demo", legend={"orientation": "h"})

    return figure

rows_frame

rows_frame() -> DataFrame

A hundred rows over five series with a group label carrying the count and the peak y.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def rows_frame() -> pl.DataFrame:
    """A hundred rows over five series with a group label carrying the count and the peak y."""
    rng = np.random.default_rng(SEED)
    index = np.arange(ROWS)
    x = index / 10.0
    frame = pl.DataFrame(
        {
            "index": index,
            SERIES_COLUMN: [SERIES[i % len(SERIES)] for i in range(ROWS)],
            "x": x,
            "y": np.sin(x) + rng.normal(0.0, 0.1, ROWS),
            "z": x * x,
        }
    )

    return group_rows(frame, SERIES_COLUMN, max_=["y"])

series_rows

series_rows(frame: DataFrame, selected: str) -> DataFrame

The grouped rows of one series, or all of them.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def series_rows(frame: pl.DataFrame, selected: str) -> pl.DataFrame:
    """The grouped rows of one series, or all of them."""
    if selected == ALL_SERIES:
        return frame

    return frame.filter(pl.col(SERIES_COLUMN) == selected)

workbench

workbench() -> Split

The demo dock: the sphere on the left, the figure over the table on the right, with a series filter in the app bar.

Source code in packages/femforge-workbench/src/femforge_workbench/demo.py
def workbench() -> Split:
    """The demo dock: the sphere on the left, the figure over the table on the right, with a series filter in the app bar."""
    frame = rows_frame()

    def sphere(app: "Workbench", id: str, title: str) -> ScenePanel:
        return ScenePanel(app, id, title, renderable=Sphere(app.palette))

    def signal(app: "Workbench", id: str, title: str) -> FigurePanel:
        return FigurePanel(app, id, title, signal_figure())

    def rows(app: "Workbench", id: str, title: str) -> TablePanel:
        table = TablePanel(app, id, title, frame, grouped=True, filterable=True, lead=(SERIES_COLUMN,))
        _ = app.state.setdefault(DemoVar.SERIES, ALL_SERIES)

        with app.header, labelled("Series"), Select(model=DemoVar.SERIES):
            with SelectTrigger(size="sm", classes="w-28"):
                SelectValue()

            with SelectContent():
                for name in (ALL_SERIES, *SERIES):
                    SelectItem(name, value=name)

        @table.listen(str(DemoVar.SERIES))
        def on_series(**_: object) -> None:
            table.set_rows(series_rows(frame, str(app.state[DemoVar.SERIES])))

        return table

    return Split(
        [
            Panel("Sphere", sphere, id=DemoPanel.SPHERE, weight=1.2),
            Split([Panel("Signal", signal, id=DemoPanel.SIGNAL), Panel("Rows", rows, id=DemoPanel.ROWS)]),
        ]
    )

femforge_workbench.settings

Environment switches the workbench reads, each spelled once.

Classes:

  • Setting –

    The environment variables that switch optional workbench behaviour on.

Functions:

  • enabled –

    Whether the switch is set to a true value.

Setting

Bases: StrEnum

The environment variables that switch optional workbench behaviour on.

Attributes:

  • OCP_VSCODE –

    Also push every CompoundRenderable to the OCP CAD Viewer in VS Code.

OCP_VSCODE class-attribute instance-attribute

OCP_VSCODE = 'FEMFORGE_OCP_VSCODE'

Also push every CompoundRenderable to the OCP CAD Viewer in VS Code.

enabled

enabled(setting: Setting) -> bool

Whether the switch is set to a true value.

Source code in packages/femforge-workbench/src/femforge_workbench/settings.py
def enabled(setting: Setting) -> bool:
    """Whether the switch is set to a true value."""
    return os.environ.get(setting, "").strip().lower() in TRUE_VALUES

femforge_workbench.reload

Watchdog-driven module reload: a .py save under the roots purges and re-imports a workbench target, builds its result on a worker thread and hands it to the server loop.

Type Aliases:

  • Apply –

    Receives the fresh result and the seconds the purge, import and factory took.

Classes:

  • Watched –

    What the shared purge needs of a reloader: the roots it watches.

  • Purge –

    The last eviction the reloaders shared: when it started and the roots it covered.

  • SaveWatcher –

    Calls on_save once per burst of .py saves under the roots (only files when given), debounce seconds after the last one; a save is an event whose file's mtime changed.

  • Reloader –

    Rebuilds the target on a worker thread after each burst of saves under the roots, one rebuild at a time with at most one more queued.

Functions:

  • package_root –

    The directory of the top-level package a module belongs to.

  • source_root –

    The directory to watch for a module: the src directory of a src layout, so its sibling packages reload with it, else the package itself.

  • persistent –

    Whether a module stays loaded across a purge.

  • shell_files –

    The source files of the shell core.

  • default_roots –

    The femforge core's source root, the workbench package's (its panels, widgets and theme rebuild in process) and the module's own.

  • outermost –

    The resolved roots without repeats and without any lying under another, since each is watched recursively.

  • is_source_save –

    The saved .py path an event reports, None for anything else; an editor saves as a modification, or as a create or move of a temp file.

Attributes:

  • DEBOUNCE_SECONDS –

    Seconds after the last save of a burst before it is acted on: an editor writes one file in several events, a save-all several files.

  • REBUILD_LOCK –

    Serializes purge-and-import across the reloaders of every open card: two threads evicting and importing the same modules at once corrupt each other.

  • WATCHING_LOCK –

    Guards the watching list alone, never held across a rebuild, so stopping a reloader does not wait for one.

  • SHELL_MODULES –

    The shell core, relative to the workbench package: never purged in process, restarted by the --dev supervisor instead.

  • PERSISTENT_PACKAGES –

    Modules the purge leaves alive: the shell core and the logging singleton it shares with the reloaded code.

  • WATCHING (list[Watched]) –

    The reloaders of the open cards, whose roots the shared purge covers.

  • PURGED –

    Read and written under REBUILD_LOCK only.

DEBOUNCE_SECONDS module-attribute

DEBOUNCE_SECONDS = 0.3

Seconds after the last save of a burst before it is acted on: an editor writes one file in several events, a save-all several files.

REBUILD_LOCK module-attribute

REBUILD_LOCK = threading.Lock()

Serializes purge-and-import across the reloaders of every open card: two threads evicting and importing the same modules at once corrupt each other.

WATCHING_LOCK module-attribute

WATCHING_LOCK = threading.Lock()

Guards the watching list alone, never held across a rebuild, so stopping a reloader does not wait for one.

SHELL_MODULES module-attribute

SHELL_MODULES = ('app', 'cli', 'dev', 'dock', 'persist', 'registry', 'reload', 'runner', 'serve', 'settings', 'widgets.element', 'widgets.module')

The shell core, relative to the workbench package: never purged in process, restarted by the --dev supervisor instead.

PERSISTENT_PACKAGES module-attribute

PERSISTENT_PACKAGES = (*(f'{femforge_workbench.__name__}.{name}' for name in SHELL_MODULES), 'femforge.common.log')

Modules the purge leaves alive: the shell core and the logging singleton it shares with the reloaded code.

WATCHING module-attribute

WATCHING: list[Watched] = []

The reloaders of the open cards, whose roots the shared purge covers.

PURGED module-attribute

PURGED = Purge()

Read and written under REBUILD_LOCK only.

Apply

Apply = Callable[[R, float], None]

Receives the fresh result and the seconds the purge, import and factory took.

Watched

Bases: Protocol

What the shared purge needs of a reloader: the roots it watches.

Attributes:

roots property

roots: list[Path]

The directories the reloader watches.

Purge dataclass

Purge(at: float = 0.0, roots: tuple[Path, ...] = ())

The last eviction the reloaders shared: when it started and the roots it covered.

Methods:

  • covers –

    Whether this purge followed the save at requested and evicted every one of those roots.

covers

covers(roots: Sequence[Path], requested: float) -> bool

Whether this purge followed the save at requested and evicted every one of those roots.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def covers(self, roots: Sequence[Path], requested: float) -> bool:
    """Whether this purge followed the save at `requested` and evicted every one of those roots."""
    if self.at < requested:
        return False

    return all(any(root.is_relative_to(purged) for purged in self.roots) for root in roots)

SaveWatcher

SaveWatcher(roots: Sequence[Path], on_save: Callable[[], None], *, debounce: float = DEBOUNCE_SECONDS, files: Sequence[Path] | None = None)

Bases: FileSystemEventHandler

Calls on_save once per burst of .py saves under the roots (only files when given), debounce seconds after the last one; a save is an event whose file's mtime changed.

Parameters:

  • roots (Sequence[Path]) –

    The directories watched recursively.

  • on_save (Callable[[], None]) –

    Called once per burst of saves.

  • debounce (float, default: DEBOUNCE_SECONDS ) –

    Seconds after the last save of a burst before on_save runs.

  • files (Sequence[Path] | None, default: None ) –

    The only files whose saves count; None counts every .py file under the roots.

Methods:

  • watched –

    The files whose saves count: files when given, else every source file under the roots.

  • changed –

    Whether the file's mtime differs from the last one seen; a file never seen counts as changed, a vanished one as unchanged.

  • start –

    Records every watched file's mtime and starts observing the roots.

  • stop –

    Cancels a pending call and stops observing.

  • on_any_event –

    Every save restarts the timer, so only the last of a burst fires.

  • fire –

    Calls on_save for the burst that just settled.

Attributes:

  • stamps (dict[Path, int]) –

    The mtime last seen per file: watchdog also reports access-time updates, which NTFS writes on reads.

  • observed –

    When the last save of the current burst was seen, which every reloader watching it reads before any purge.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def __init__(self, roots: Sequence[Path], on_save: Callable[[], None], *, debounce: float = DEBOUNCE_SECONDS, files: Sequence[Path] | None = None) -> None:
    """Sets the watcher up stopped; `start` begins observing.

    Args:
        roots: The directories watched recursively.
        on_save: Called once per burst of saves.
        debounce: Seconds after the last save of a burst before `on_save` runs.
        files: The only files whose saves count; None counts every .py file under the roots.
    """
    super().__init__()
    self.roots = outermost(roots)
    self.files = {file.resolve() for file in files} if files is not None else None
    self.on_save = on_save
    self.debounce = debounce
    self.stamps: dict[Path, int] = {}
    """The mtime last seen per file: watchdog also reports access-time updates, which NTFS writes on reads."""
    self.timer: threading.Timer | None = None
    self.lock = threading.Lock()
    self.observer: BaseObserver | None = None
    self.observed = 0.0
    """When the last save of the current burst was seen, which every reloader watching it reads before any purge."""

stamps instance-attribute

stamps: dict[Path, int] = {}

The mtime last seen per file: watchdog also reports access-time updates, which NTFS writes on reads.

observed instance-attribute

observed = 0.0

When the last save of the current burst was seen, which every reloader watching it reads before any purge.

watched

watched() -> list[Path]

The files whose saves count: files when given, else every source file under the roots.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def watched(self) -> list[Path]:
    """The files whose saves count: `files` when given, else every source file under the roots."""
    if self.files is not None:
        return sorted(self.files)

    return [path for root in self.roots for path in root.rglob(f"*{SOURCE_SUFFIX}") if PYCACHE not in path.parts]

changed

changed(path: Path) -> bool

Whether the file's mtime differs from the last one seen; a file never seen counts as changed, a vanished one as unchanged.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def changed(self, path: Path) -> bool:
    """Whether the file's mtime differs from the last one seen; a file never seen counts as changed, a vanished one as unchanged."""
    try:
        stamp = path.stat().st_mtime_ns
    except OSError:
        return False

    if self.stamps.get(path) == stamp:
        return False

    self.stamps[path] = stamp

    return True

start

start() -> None

Records every watched file's mtime and starts observing the roots.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def start(self) -> None:
    """Records every watched file's mtime and starts observing the roots."""
    for path in self.watched():
        _ = self.changed(path)

    observer: BaseObserver = Observer()
    for root in self.roots:
        _ = observer.schedule(self, str(root), recursive=True)

    observer.start()
    self.observer = observer
    log.bind(roots=[str(root) for root in self.roots]).info("Watching for saves.")

stop

stop() -> None

Cancels a pending call and stops observing.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def stop(self) -> None:
    """Cancels a pending call and stops observing."""
    with self.lock:
        if self.timer is not None:
            self.timer.cancel()
            self.timer = None

    if self.observer is None:
        return

    self.observer.stop()
    self.observer.join()
    self.observer = None

on_any_event

on_any_event(event: FileSystemEvent) -> None

Every save restarts the timer, so only the last of a burst fires.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def on_any_event(self, event: FileSystemEvent) -> None:
    """Every save restarts the timer, so only the last of a burst fires."""
    path = is_source_save(event)
    if path is None:
        return

    resolved = Path(path).resolve()
    if self.files is not None and resolved not in self.files:
        return

    if not self.changed(resolved):
        return

    self.observed = time.monotonic()
    with self.lock:
        if self.timer is not None:
            self.timer.cancel()

        self.timer = threading.Timer(self.debounce, self.fire)
        self.timer.daemon = True
        self.timer.start()

    log.bind(path=path).debug("File changed.")

fire

fire() -> None

Calls on_save for the burst that just settled.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def fire(self) -> None:
    """Calls `on_save` for the burst that just settled."""
    with self.lock:
        self.timer = None

    self.on_save()

Reloader

Reloader(target: str, roots: Sequence[Path], apply: Apply[R], *, kind: type[R], marshal: Marshal, failed: Failed | None = None)

Rebuilds the target on a worker thread after each burst of saves under the roots, one rebuild at a time with at most one more queued.

Parameters:

  • target (str) –

    The pkg.module:attr factory rebuilt after every save.

  • roots (Sequence[Path]) –

    The directories whose .py saves trigger a rebuild.

  • apply (Apply[R]) –

    Receives each result and the seconds it took, on the server loop.

  • kind (type[R]) –

    The type the factory must return.

  • marshal (Marshal) –

    Runs a callable on the server loop.

  • failed (Failed | None, default: None ) –

    Receives a failed build's error on the server loop; None drops it.

Methods:

  • start –

    Watches the roots; request builds now.

  • stop –

    Stops watching and leaves the shared purge.

  • request –

    Starts a rebuild, or queues one more when a rebuild is running.

  • work –

    Rebuilds until no request is queued.

  • rebuild –

    Builds once and delivers the result, or the failure, to the server loop.

  • deliver –

    Runs outcome on the server loop, dropping it when the loop is gone.

  • report –

    Hands a failed build's error to failed.

  • build –

    Purges the watched modules, imports the target afresh and calls it, one reloader at a time; a build without a save behind it always purges.

  • purge –

    Drops every loaded module under the watched roots except the persistent packages, once per save however many cards are open.

Attributes:

  • requested –

    When the save that asked for the pending rebuild was seen; a purge older than that is not this rebuild's.

  • roots (list[Path]) –

    The directories the watcher observes.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def __init__(self, target: str, roots: Sequence[Path], apply: Apply[R], *, kind: type[R], marshal: Marshal, failed: Failed | None = None) -> None:
    """Sets the reloader up stopped; `start` watches the roots.

    Args:
        target: The `pkg.module:attr` factory rebuilt after every save.
        roots: The directories whose .py saves trigger a rebuild.
        apply: Receives each result and the seconds it took, on the server loop.
        kind: The type the factory must return.
        marshal: Runs a callable on the server loop.
        failed: Receives a failed build's error on the server loop; None drops it.
    """
    self.target = target
    self.kind = kind
    self.apply = apply
    self.failed = failed
    self.marshal = marshal
    self.watcher = SaveWatcher(roots, self.request)
    self.running = False
    self.queued = False
    self.requested = 0.0
    """When the save that asked for the pending rebuild was seen; a purge older than that is not this rebuild's."""
    self.lock = threading.Lock()

requested instance-attribute

requested = 0.0

When the save that asked for the pending rebuild was seen; a purge older than that is not this rebuild's.

roots property

roots: list[Path]

The directories the watcher observes.

start

start() -> None

Watches the roots; request builds now.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def start(self) -> None:
    """Watches the roots; `request` builds now."""
    with WATCHING_LOCK:
        if self not in WATCHING:
            WATCHING.append(self)

    self.watcher.start()

stop

stop() -> None

Stops watching and leaves the shared purge.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def stop(self) -> None:
    """Stops watching and leaves the shared purge."""
    with WATCHING_LOCK:
        if self in WATCHING:
            WATCHING.remove(self)

    self.watcher.stop()

request

request() -> None

Starts a rebuild, or queues one more when a rebuild is running.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def request(self) -> None:
    """Starts a rebuild, or queues one more when a rebuild is running."""
    self.requested = self.watcher.observed or time.monotonic()
    with self.lock:
        if self.running:
            self.queued = True
            return

        self.running = True

    threading.Thread(target=self.work, name="workbench-reload", daemon=True).start()

work

work() -> None

Rebuilds until no request is queued.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def work(self) -> None:
    """Rebuilds until no request is queued."""
    while True:
        self.rebuild()

        with self.lock:
            if not self.queued:
                self.running = False
                return

            self.queued = False

rebuild

rebuild() -> None

Builds once and delivers the result, or the failure, to the server loop.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def rebuild(self) -> None:
    """Builds once and delivers the result, or the failure, to the server loop."""
    started = time.perf_counter()
    requested = self.requested or time.monotonic()
    try:
        result = self.build(requested)
    except Exception as error:
        get_console().print_exception()
        failure = error
        self.deliver(lambda: self.report(failure))
        return

    seconds = time.perf_counter() - started
    log.bind(target=self.target, seconds=round(seconds, 2)).info("Target rebuilt.")
    self.deliver(lambda: self.apply(result, seconds))

deliver

deliver(outcome: Callable[[], None]) -> None

Runs outcome on the server loop, dropping it when the loop is gone.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def deliver(self, outcome: Callable[[], None]) -> None:
    """Runs `outcome` on the server loop, dropping it when the loop is gone."""
    try:
        self.marshal(outcome)
    except RuntimeError:
        log.bind(target=self.target).warning("No server loop to deliver the rebuild to; dropped.")

report

report(error: Exception) -> None

Hands a failed build's error to failed.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def report(self, error: Exception) -> None:
    """Hands a failed build's error to `failed`."""
    if self.failed is not None:
        self.failed(error)

build

build(requested: float | None = None) -> R

Purges the watched modules, imports the target afresh and calls it, one reloader at a time; a build without a save behind it always purges.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def build(self, requested: float | None = None) -> R:
    """Purges the watched modules, imports the target afresh and calls it, one reloader at a time; a build without a save behind it always purges."""
    with REBUILD_LOCK:
        self.purge(time.monotonic() if requested is None else requested)

        factory = resolve(self.target)
        if not callable(factory):
            raise TypeError(f"{self.target} is not callable")

        result = factory()
    if not isinstance(result, self.kind):
        raise TypeError(f"{self.target}() returned {type(result).__name__}, not a {self.kind.__name__}")

    return result

purge

purge(requested: float) -> None

Drops every loaded module under the watched roots except the persistent packages, once per save however many cards are open.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def purge(self, requested: float) -> None:
    """Drops every loaded module under the watched roots except the persistent packages, once per save however many cards are open."""
    global PURGED
    with WATCHING_LOCK:
        roots = outermost([*self.roots, *(root for watched in WATCHING for root in watched.roots)])

    # One purge per card would evict what the card before it just imported, leaving every card on its own class
    # generation; the roots of every open card are covered here, so the cards that follow skip their own.
    if PURGED.covers(roots, requested):
        log.bind(module=split_target(self.target)[0]).debug("Modules already purged for this save.")
        return

    def evictable(name: str, module: ModuleType) -> bool:
        if name == MAIN_MODULE or persistent(name):
            return False

        file = getattr(module, "__file__", None)
        if not isinstance(file, str):
            return False

        path = Path(file).resolve()

        return any(path.is_relative_to(root) for root in roots)

    at = time.monotonic()
    stale = [name for name, module in list(sys.modules.items()) if evictable(name, module)]
    for name in stale:
        _ = sys.modules.pop(name, None)

    linecache.checkcache()
    PURGED = Purge(at, tuple(roots))
    log.bind(module=split_target(self.target)[0], purged=len(stale), roots=len(roots)).debug("Modules purged.")

package_root

package_root(module_name: str) -> Path

The directory of the top-level package a module belongs to.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def package_root(module_name: str) -> Path:
    """The directory of the top-level package a module belongs to."""
    top = importlib.import_module(module_name.split(".", 1)[0])
    file = top.__file__
    if file is None:
        raise ValueError(f"{module_name!r} has no package directory to watch")

    return Path(file).resolve().parent

source_root

source_root(module_name: str) -> Path

The directory to watch for a module: the src directory of a src layout, so its sibling packages reload with it, else the package itself.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def source_root(module_name: str) -> Path:
    """The directory to watch for a module: the `src` directory of a src layout, so its sibling packages reload with it, else the package itself."""
    package = package_root(module_name)
    parent = package.parent
    if parent.name == SRC_LAYOUT and not (parent / PACKAGE_MARKER).exists():
        return parent

    return package

persistent

persistent(name: str) -> bool

Whether a module stays loaded across a purge.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def persistent(name: str) -> bool:
    """Whether a module stays loaded across a purge."""
    return any(name == package or name.startswith(package + ".") for package in PERSISTENT_PACKAGES)

shell_files

shell_files() -> list[Path]

The source files of the shell core.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def shell_files() -> list[Path]:
    """The source files of the shell core."""
    root = package_root(femforge_workbench.__name__)

    return [root.joinpath(*name.split(".")).with_suffix(SOURCE_SUFFIX) for name in SHELL_MODULES]

default_roots

default_roots(workbench_module: str) -> list[Path]

The femforge core's source root, the workbench package's (its panels, widgets and theme rebuild in process) and the module's own.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def default_roots(workbench_module: str) -> list[Path]:
    """The femforge core's source root, the workbench package's (its panels, widgets and theme rebuild in process) and the module's own."""
    roots = [source_root(femforge.__name__), source_root(femforge_workbench.__name__)]
    own = source_root(workbench_module)
    if own not in roots:
        roots.append(own)

    return roots

outermost

outermost(roots: Sequence[Path]) -> list[Path]

The resolved roots without repeats and without any lying under another, since each is watched recursively.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def outermost(roots: Sequence[Path]) -> list[Path]:
    """The resolved roots without repeats and without any lying under another, since each is watched recursively."""
    resolved = list(dict.fromkeys(root.resolve() for root in roots))

    return [root for root in resolved if not any(root != other and root.is_relative_to(other) for other in resolved)]

is_source_save

is_source_save(event: FileSystemEvent) -> str | None

The saved .py path an event reports, None for anything else; an editor saves as a modification, or as a create or move of a temp file.

Source code in packages/femforge-workbench/src/femforge_workbench/reload.py
def is_source_save(event: FileSystemEvent) -> str | None:
    """The saved .py path an event reports, None for anything else; an editor saves as a modification, or as a create or move of a temp file."""
    if event.event_type not in SAVE_EVENTS:
        return None

    path = os.fsdecode(event.dest_path or event.src_path)
    if event.is_directory or not path.endswith(SOURCE_SUFFIX) or PYCACHE in path:
        return None

    return path

femforge_workbench.dev

The dev loop for the shell core: a supervisor that reruns the server after every save of a core module, and the page script that follows it back and reloads on a rebuilt dist.

Classes:

  • Supervisor –

    Runs a command as a child process and restarts it after every burst of .py saves under the roots, or of the files when given.

Functions:

  • reload_script –

    Probes the server and the dist each second: the page reloads when the server comes back after a restart, when version.json changes, and when the library failed to install.

  • without_dev –

    The arguments without the dev flag.

  • child_command –

    The server as python -m module, given the arguments without the dev flag.

  • supervise –

    Serves python -m module argv under a supervisor restarting it on a save of a shell core module, until Ctrl-C; every other save rebuilds in process.

Attributes:

  • DEV_ENV –

    Set in the supervised server's environment; the shell then serves the self-reload script.

  • POLL_SECONDS –

    How often the supervisor looks at its child between saves; a bare wait would hold Ctrl-C on Windows.

  • PROBE_PAGE –

    Probed beside the page's own URL: the root answers HEAD with 403, the page itself with 200.

  • LIBRARY_GRACE_MS –

    How long after load the page waits for the widget library before reloading once more.

  • STOP_GRACE –

    Seconds a child gets to exit on its own after the break, so its exit hooks flush the persisted state.

  • NEW_GROUP –

    The child gets its own console group, so the break reaches it and not the supervisor.

DEV_ENV module-attribute

DEV_ENV = 'FEMFORGE_WORKBENCH_DEV'

Set in the supervised server's environment; the shell then serves the self-reload script.

POLL_SECONDS module-attribute

POLL_SECONDS = 0.25

How often the supervisor looks at its child between saves; a bare wait would hold Ctrl-C on Windows.

PROBE_PAGE module-attribute

PROBE_PAGE = 'index.html'

Probed beside the page's own URL: the root answers HEAD with 403, the page itself with 200.

LIBRARY_GRACE_MS module-attribute

LIBRARY_GRACE_MS = 2000

How long after load the page waits for the widget library before reloading once more.

STOP_GRACE module-attribute

STOP_GRACE = 5.0

Seconds a child gets to exit on its own after the break, so its exit hooks flush the persisted state.

NEW_GROUP module-attribute

The child gets its own console group, so the break reaches it and not the supervisor.

Supervisor

Supervisor(command: Sequence[str], roots: Sequence[Path], *, files: Sequence[Path] | None = None)

Runs a command as a child process and restarts it after every burst of .py saves under the roots, or of the files when given.

Parameters:

  • command (Sequence[str]) –

    The server command the child runs.

  • roots (Sequence[Path]) –

    The directories whose .py saves restart the child.

  • files (Sequence[Path] | None, default: None ) –

    The only files whose saves restart the child; None counts every .py file under the roots.

Methods:

  • spawn –

    Starts the child in its own process group and tree, flagged as supervised.

  • stop_child –

    Asks the child to exit, so its exit hooks flush the persisted state, and kills its tree when it does not within the grace.

  • release –

    Closes the child's tree once the child is gone, which also ends anything the child left behind.

  • await_save –

    Returns at the next save or stop, noting once when the child exits on its own meanwhile.

  • serve –

    Runs the child until Ctrl-C or stop, restarting it after each save.

  • stop –

    Ends serve, which then stops the child.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def __init__(self, command: Sequence[str], roots: Sequence[Path], *, files: Sequence[Path] | None = None) -> None:
    """Sets the supervisor up; `serve` runs it.

    Args:
        command: The server command the child runs.
        roots: The directories whose .py saves restart the child.
        files: The only files whose saves restart the child; None counts every .py file under the roots.
    """
    self.command = list(command)
    self.wake = threading.Event()
    self.stopping = threading.Event()
    self.watcher = SaveWatcher(roots, self.wake.set, files=files)
    self.child: subprocess.Popen[bytes] | None = None
    self.tree: ProcessTree | None = None
    self.starts = 0

spawn

spawn() -> Popen[bytes]

Starts the child in its own process group and tree, flagged as supervised.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def spawn(self) -> subprocess.Popen[bytes]:
    """Starts the child in its own process group and tree, flagged as supervised."""
    self.release()
    child = subprocess.Popen(
        self.command,
        env={**os.environ, DEV_ENV: "1"},
        creationflags=NEW_GROUP | CREATION_FLAGS,
        start_new_session=NEW_SESSION,
    )
    self.tree = ProcessTree.adopt(child.pid)
    self.child = child
    self.starts += 1
    log.bind(pid=child.pid, command=self.command, starts=self.starts).info("Server started.")

    return child

stop_child

stop_child(child: Popen[bytes]) -> None

Asks the child to exit, so its exit hooks flush the persisted state, and kills its tree when it does not within the grace.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def stop_child(self, child: subprocess.Popen[bytes]) -> None:
    """Asks the child to exit, so its exit hooks flush the persisted state, and kills its tree when it does not within the grace."""
    child.send_signal(BREAK_SIGNAL)
    try:
        _ = child.wait(timeout=STOP_GRACE)
    except subprocess.TimeoutExpired:
        if self.tree is not None:
            _ = self.tree.kill()

        log.bind(pid=child.pid).warning("Server did not stop on the break; killed.")

release

release() -> None

Closes the child's tree once the child is gone, which also ends anything the child left behind.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def release(self) -> None:
    """Closes the child's tree once the child is gone, which also ends anything the child left behind."""
    if self.tree is not None:
        self.tree.close()
        self.tree = None

await_save

await_save(child: Popen[bytes]) -> None

Returns at the next save or stop, noting once when the child exits on its own meanwhile.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def await_save(self, child: subprocess.Popen[bytes]) -> None:
    """Returns at the next save or stop, noting once when the child exits on its own meanwhile."""
    self.wake.clear()
    exited = False
    while not self.wake.wait(POLL_SECONDS):
        if self.stopping.is_set():
            return

        if not exited and child.poll() is not None:
            exited = True
            log.bind(returncode=child.returncode).warning("Server exited; waiting for a save.")

serve

serve() -> None

Runs the child until Ctrl-C or stop, restarting it after each save.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def serve(self) -> None:
    """Runs the child until Ctrl-C or `stop`, restarting it after each save."""
    self.watcher.start()
    try:
        while not self.stopping.is_set():
            child = self.spawn()
            self.await_save(child)
            if child.poll() is None:
                self.stop_child(child)

            self.release()
    except KeyboardInterrupt:
        pass
    finally:
        if self.child is not None and self.child.poll() is None:
            self.stop_child(self.child)

        self.release()
        self.watcher.stop()
        log.info("Supervisor stopped.")

stop

stop() -> None

Ends serve, which then stops the child.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def stop(self) -> None:
    """Ends `serve`, which then stops the child."""
    self.stopping.set()
    self.wake.set()

reload_script

reload_script() -> str

Probes the server and the dist each second: the page reloads when the server comes back after a restart, when version.json changes, and when the library failed to install.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def reload_script() -> str:
    """Probes the server and the dist each second: the page reloads when the server comes back after a restart, when version.json changes, and when the library failed to install."""
    version_url = f"/{ROUTE}/{VERSION_FILE}"

    return (
        "(function follow() {"
        " let down = false;"
        " let version = null;"
        f" const assets = {json.dumps(assets())};"
        " async function refresh() {"
        "  await Promise.all(assets.map((asset) => fetch(asset, { cache: 'reload' })));"
        "  location.reload();"
        " }"
        " setInterval(async () => {"
        "  try {"
        f"   const response = await fetch(new URL('{PROBE_PAGE}', location.href), {{ method: 'HEAD', cache: 'no-store' }});"
        "   if (!response.ok) { throw new Error(String(response.status)); }"
        "   if (down) { location.reload(); return; }"
        f"   const probe = await fetch('{version_url}', {{ cache: 'no-store' }});"
        "   if (!probe.ok) { return; }"
        "   const current = await probe.text();"
        "   if (version === null) { version = current; }"
        "   else if (current !== version) { version = current; await refresh(); }"
        "  } catch (error) { down = true; }"
        f" }}, {PROBE_MS});"
        f" setTimeout(() => {{ if (!window.Ff) {{ refresh(); }} }}, {LIBRARY_GRACE_MS});"
        "})();"
    )

without_dev

without_dev(argv: Sequence[str]) -> list[str]

The arguments without the dev flag.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def without_dev(argv: Sequence[str]) -> list[str]:
    """The arguments without the dev flag."""
    return [arg for arg in argv if arg != DEV_FLAG]

child_command

child_command(module: str, argv: Sequence[str]) -> list[str]

The server as python -m module, given the arguments without the dev flag.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def child_command(module: str, argv: Sequence[str]) -> list[str]:
    """The server as `python -m module`, given the arguments without the dev flag."""
    return [sys.executable, "-m", module, *without_dev(argv)]

supervise

supervise(module: str, argv: Sequence[str]) -> None

Serves python -m module argv under a supervisor restarting it on a save of a shell core module, until Ctrl-C; every other save rebuilds in process.

Source code in packages/femforge-workbench/src/femforge_workbench/dev.py
def supervise(module: str, argv: Sequence[str]) -> None:
    """Serves `python -m module argv` under a supervisor restarting it on a save of a shell core module, until Ctrl-C; every other save rebuilds in process."""
    Supervisor(child_command(module, argv), [package_root(femforge_workbench.__name__)], files=shell_files()).serve()