scistudio.blocks.base

Canonical import root: from scistudio.blocks.base import ...

Self-contained public-API reference — 13 symbols from this module's __all__, with signatures and docstrings inlined (ADR-052 §7). Generated; do not hand-edit.

Block — class

Stability: stable · Since 0.3.1

class Block(ABC)
Block(config: 'dict[str, Any] | None' = None) -> 'None'

Abstract base class every processing block inherits from.

A block is one node in a workflow: it declares typed input and output ports, reads its parameters from a BlockConfig, and turns input data objects into output data objects. This base class defines the contract the runtime relies on and supplies working defaults for everything except the main work.

To write a block, subclass this (or a more specific category such as ProcessBlock), describe it, and implement the work:

  • Declare named inputs and outputs by setting the input_ports and output_ports class attributes (lists of InputPort / OutputPort), and describe the parameters the block accepts in config_schema.
  • Implement run, which receives the inputs keyed by input-port name and the resolved config, and returns the outputs keyed by output-port name. Each output value is a scistudio.core.types.collection.Collection.
  • Optionally override validate (input checks before the run) and postprocess (adjust outputs before delivery); both have working defaults.

The display metadata (name, description, ui_icon, ui_color, …) is how a block describes itself to the workflow editor and registry.

Example: >>> from scistudio.blocks.base import Block, OutputPort >>> class Greeter(Block): # doctest: +SKIP ... name = "Greeter" ... output_ports = [OutputPort(name="message", accepted_types=[])] ... def run(self, inputs, config): ... greeting = config.get("greeting", "hello") ... return {"message": self.pack([greeting])}

Members

  • get_panel_manifest(self) -> 'Any | None' — provisional · Since 0.3.1 — Return this block's interactive panel manifest, or None.
  • get_effective_input_ports(self) -> 'list[InputPort]' — stable · Since 0.3.1 — Return the input ports that actually apply to this block instance.
  • get_effective_output_ports(self) -> 'list[OutputPort]' — stable · Since 0.3.1 — Return the output ports that actually apply to this block instance.
  • validate(self, inputs: 'dict[str, Any]') -> 'bool' — stable · Since 0.3.1 — Check that inputs satisfy the block's input-port contract.
  • run(self, inputs: 'dict[str, Collection]', config: 'BlockConfig') -> 'dict[str, Collection]' — stable · Since 0.3.1 — Execute the block's work and return its outputs. Must be overridden.
  • postprocess(self, outputs: 'dict[str, Collection]') -> 'dict[str, Collection]' — stable · Since 0.3.1 — Adjust the block's outputs before they are handed downstream.
  • process_item(self, item: 'Any', config: 'BlockConfig') -> 'Any' — stable · Since 0.3.1 — Process a single item of the input collection. Override for per-item work.
  • pack(items: 'list[Any]', item_type: 'type | None' = None) -> 'Any' — stable · Since 0.3.1 — Bundle data objects into a Collection, saving any not yet saved.
  • unpack(collection: 'Any') -> 'list[Any]' — stable · Since 0.3.1 — Return a Collection's items as a plain list of data objects.
  • unpack_single(collection: 'Any') -> 'Any' — stable · Since 0.3.1 — Return the single data object from a length-one Collection.
  • map_items(func: 'Any', collection: 'Any') -> 'Any' — stable · Since 0.3.1 — Apply func to each item in turn, saving each result, into a new Collection.
  • parallel_map(func: 'Any', collection: 'Any', max_workers: 'int' = 4) -> 'Any' — stable · Since 0.3.1 — Apply func to each item across worker processes, saving each result.
  • persist_array(self, data_or_iterator: 'Any', shape: 'tuple[int, ...]', dtype: 'Any', output_dir: 'str | None' = None, chunks: 'tuple[int, ...] | None' = None) -> 'StorageReference' — stable · Since 0.3.1 — Write array data to zarr storage and return a reference to it.
  • persist_table(self, table: 'Any', output_dir: 'str | None' = None) -> 'StorageReference' — stable · Since 0.3.1 — Write an Arrow table to parquet storage and return a reference to it.

BlockConfig — class

Stability: stable · Since 0.3.1

class BlockConfig(BaseModel)

Holds the parameters a block reads at run time.

A Pydantic model that allows extra fields, so a block or package can attach its own validated parameters without changing this base class, and the runtime can inject extra keys (such as block_id or project_dir) onto the same object. Reach for get to read a value by name whether it lives in params or was attached as an extra field.

Example: >>> from scistudio.blocks.base import BlockConfig >>> config = BlockConfig(params={"threshold": 0.5}) >>> config.get("threshold") 0.5 >>> config.get("missing", default=0) 0

Members

  • get(self, key: 'str', default: 'Any' = None) -> 'Any' — stable · Since 0.3.1 — Return the value of parameter key, or default when it is not set.

ExecutionMode — enum

Stability: stable · Since 0.3.1

class ExecutionMode(Enum)

How the runtime executes a block.

Set the execution_mode class attribute on a block to one of these values so the engine knows how to run it. Most blocks leave it at the default AUTO. Choose INTERACTIVE only when the block pauses to ask the user a question mid-run, and EXTERNAL only when it hands the work off to a separate desktop application.

Example: >>> from scistudio.blocks.base import ExecutionMode >>> ExecutionMode.AUTO.value 'auto'

INTERACTIVE_RESPONSE_KEY — constant

Stability: unmarked — see the module source / ADR-052

constant — see the module source for the value.

InputPort — class

Stability: stable · Since 0.3.1

class InputPort(Port)
InputPort(*, name: 'str', accepted_types: 'list[type]', is_collection: 'bool' = False, description: 'str' = '', required: 'bool' = True, default: 'Any | None' = None, constraint: 'Callable[[Any], bool] | None' = None, constraint_description: 'str' = '') -> None

A typed input connection point on a block.

Declare one per named input your block reads. The workflow editor uses the accepted types to decide which upstream outputs may connect here, and the runtime uses them (plus the optional constraint) to validate incoming data before the block runs.

Set these fields when constructing a port:

  • name — the port's identifier and the key your block looks up in the inputs mapping passed to Block.run.
  • accepted_types — the data types this port accepts; an empty list means "accept any data object".
  • description — a short human-readable explanation shown in the editor.
  • required — whether a connection is mandatory (defaults to True).
  • is_collection — whether the port expects a multi-item collection.

Example: >>> from scistudio.blocks.base import InputPort >>> port = InputPort(name="image", accepted_types=[], required=True)

InteractiveMixin — class

Stability: provisional · Since 0.3.1

class InteractiveMixin

Mix-in that makes a block interactive.

Inherit this alongside a block category and declare execution_mode = ExecutionMode.INTERACTIVE to turn an ordinary block into one that pauses to ask the user a question. The registry rejects a block that declares one of these without the other. A subclass MUST set the interactive_panel class attribute and SHOULD override prepare_prompt (the default raises). The block's own run — inherited from its category — consumes the user's decision on the compute phase.

Example: >>> from scistudio.blocks.base import ( ... ExecutionMode, ... InteractiveMixin, ... PanelManifest, ... ) >>> class PickOne(InteractiveMixin): # doctest: +SKIP ... execution_mode = ExecutionMode.INTERACTIVE ... interactive_panel = PanelManifest(panel_id="core.interactive.data_router") ... def prepare_prompt(self, inputs, config): ... return {"items": [str(i) for i in inputs]}

Members

  • prepare_prompt(self, inputs: 'dict[str, Any]', config: 'BlockConfig') -> 'InteractivePrompt | dict[str, Any]' — provisional · Since 0.3.1 — Turn the real input data into what the window should show.
  • remap_saved_decision(self, saved_decision: 'dict[str, Any]', saved_signature: 'dict[str, list[str]]', current_signature: 'dict[str, list[str]]') -> 'dict[str, Any] | None' — provisional · Since 0.3.1 — Re-resolve a remembered decision against the current inputs.

InteractivePrompt — class

Stability: provisional · Since 0.3.1

class InteractivePrompt
InteractivePrompt(panel_payload: 'dict[str, Any]', intermediate: 'tuple[StorageReference, ...]' = ()) -> None

What InteractiveMixin.prepare_prompt returns to drive the panel.

Bundles the window-sized view the panel renders together with any heavy intermediate results the block wants to reuse afterwards, without putting those heavy results in memory or on the wire.

Example: >>> prompt = InteractivePrompt(panel_payload={"items": ["a", "b"]})

OutputPort — class

Stability: stable · Since 0.3.1

class OutputPort(Port)
OutputPort(*, name: 'str', accepted_types: 'list[type]', is_collection: 'bool' = False, description: 'str' = '', required: 'bool' = True) -> None

A typed output connection point on a block.

Declare one per named output your block produces. The key you use for each value in the mapping returned by Block.run must match an output port's name. The workflow editor uses the accepted types to decide which downstream inputs this output may connect to.

Set these fields when constructing a port:

  • name — the port's identifier and the key your block writes in the mapping returned by Block.run.
  • accepted_types — the data types this port emits; an empty list means "may emit any data object".
  • description — a short human-readable explanation shown in the editor.

Example: >>> from scistudio.blocks.base import OutputPort >>> port = OutputPort(name="result", accepted_types=[])

PANEL_API_VERSION — constant

Stability: unmarked — see the module source / ADR-052

constant — see the module source for the value.

PackageInfo — class

Stability: stable · Since 0.3.1

class PackageInfo
PackageInfo(name: 'str', description: 'str' = '', author: 'str' = '', version: 'str' = '0.1.0', ota: 'PackageOtaSource | None' = None) -> None

Metadata describing an external block package.

An external block package returns a PackageInfo alongside its block list from its scistudio.blocks entry-point callable. The registry uses it to place the package in the palette as a two-level grouping (package -> category -> block).

Example: >>> info = PackageInfo( ... name="My Imaging Blocks", ... description="Microscopy helpers", ... author="Lab X", ... version="1.2.0", ... )

PackageOtaSource — class

Stability: provisional · Since 0.3.1

class PackageOtaSource
PackageOtaSource(manifest_url: 'str', channel: 'str' = 'stable') -> None

Where a block package publishes its over-the-air (OTA) update manifest.

A package declares its own update source so the core runtime never has to keep a list of packages. When this is set on a PackageInfo, the in-app Package Manager checks the source for newer releases.

Example: >>> source = PackageOtaSource( ... manifest_url="https://example.org/pkg/manifest.json", ... channel="stable", ... )

PanelManifest — class

Stability: provisional · Since 0.3.1

class PanelManifest
PanelManifest(panel_id: 'str', module_url: 'str' = '', export_name: 'str' = 'default', css: 'tuple[str, ...]' = (), version: 'str' = '0', api_version: 'str' = '1', response_schema: 'dict[str, Any] | None' = None, asset_root: 'str | None' = None) -> None

Describes the frontend window component a block opens for interaction.

An interactive block declares one of these as its interactive_panel to name the window the user sees. A built-in (core) panel is resolved by panel_id against the frontend's built-in registry; a package-provided panel is loaded by importing module_url from the backend (same-origin only — remote URLs are rejected). Core panels leave module_url empty because they ship with the app.

Example: >>> manifest = PanelManifest(panel_id="core.interactive.data_router")

Members

  • to_dict(self) -> 'dict[str, Any]' — provisional · Since 0.3.1 — Return the JSON-safe wire form of this manifest sent to the frontend.

load_intermediate — function

Stability: provisional · Since 0.3.1

load_intermediate(config: 'BlockConfig | dict[str, Any]') -> 'tuple[StorageReference, ...]'

Return the intermediate storage references the engine carried across the pause.

Convenience for an interactive block's run (the compute phase) to read back the references its InteractiveMixin.prepare_prompt stored, without re-deriving the wire shape by hand.

Args: config: The block's config — a BlockConfig or a plain dict.

Returns: The stored storage references as a tuple, empty when the block produced no intermediate.