scistudio.blocks.process

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

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

ProcessBlock — class

Stability: stable · Since 0.3.1

class ProcessBlock(Block)

Transform every item of a Collection with a deterministic algorithm.

Subclass this for row-by-row or item-by-item transforms (filtering, normalising, feature extraction). The base run streams the primary input Collection one item at a time, so peak memory stays at roughly one item regardless of how large the dataset is.

To implement a block, pick one of two levels of control:

  • Common case -- override process_item with the signature (self, item, config, state=None). The base run does the rest: one setup call, one process_item per item with the shared state, auto-flush of each result, packing into the output Collection, and teardown in a finally block.
  • Full control -- override run directly and use map_items(), parallel_map(), or pack() to build the output Collection yourself.

Ports: reads the primary (first) input Collection and emits one output Collection of the same length on the first output port. Config: this base class reads no config fields of its own; a subclass declares whatever config keys its process_item consumes.

Set the class attribute algorithm to a human-readable identifier for the transform.

Example: >>> class Doubler(ProcessBlock): ... algorithm = "doubler" ... def process_item(self, item, config, state=None): ... return item * 2

Members

  • setup(self, config: 'BlockConfig') -> 'Any' — stable · Since 0.3.1 — Run once at the start of run, before any item is processed.
  • teardown(self, state: 'Any') -> 'None' — stable · Since 0.3.1 — Run once at the end of run, even if an item raised.
  • process_item(self, item: 'Any', config: 'BlockConfig', state: 'Any' = None) -> 'Any' — stable · Since 0.3.1 — Transform a single item; override this for the common case.
  • run(self, inputs: 'dict[str, Collection]', config: 'BlockConfig') -> 'dict[str, Collection]' — stable · Since 0.3.1 — Stream the primary input Collection through process_item.