hyped.core.nodes.base module

Provides base classes for nodes in a data flow graph.

This module defines base classes for nodes in a data flow graph. It includes a base configuration class (BaseNodeConfig) and a generic base class (BaseNode) for defining nodes with configurable input and output types.

class hyped.core.nodes.base.BaseNode(config: None | U = None, **kwargs)

Bases: BaseConfigurable[C], ABC

Base class for nodes in a data flow graph.

This class serves as a base for defining nodes in a data flow graph. Nodes are the building blocks of the graph, where each node represents a processing unit or transformation in the flow. It provides methods for configuring the node’s signature, and interacting with the underlying graph structure.

Config

alias of BaseNodeConfig

call(*args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array

call(flow: AbstractDataFlow, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array

Call the node, adding it to the underlying data flow.

This method is adds the node in the context of a data flow graph. It validates the node’s signature, processes the input arguments, and adds necessary constants and processor nodes to the graph. Finally, it returns the processed output feature.

Parameters:

• *args (AbstractDataFlow | Feature) – Positional arguments.

• **kwargs (AbstractDataFlow | Feature) – Keyword arguments.

Returns:

The feature resulting from the node’s processing in the graph.

Return type:

_Feature

Raises:

RuntimeError – If the flow cannot be inferred from the arguments.

get_state() → dict[str, Any]

Get the node state.

Returns:

The node’s state, including __dict__ and __slots__ (if present).

Return type:

dict[str, Any]

initialize(ctx: RunContext) → None

Initialize the node for execution.

This method is called before the data flow starts executing. It prepares the node for the upcoming execution by setting up any necessary state. Keep in mind that the node’s state is reset after each execution, and this method is called again for each new run.

The objects created in this method are managed by the RunSession instance, which ensures that objects that cannot be pickled can still be used effectively in multiprocessing settings. In such settings, this method is executed in the child processes.

Parameters:

ctx (RunContext) – The context of the current run.

set_state(state: dict[str, Any]) → None

Set the node state.

Parameters:

state (dict[str, Any]) – A dictionary representing the state to restore, containing __dict__ and __slots__ (if present).

abstract property signature: Signature

Abstract method to define the node’s signature.

The signature method must be implemented by subclasses to return the signature of the node. The signature typically describes the expected input and output types for the node, providing essential metadata for the graph processing.

Returns:

The signature of the node, including its inputs and outputs.

Return type:

Signature

with_state(state: dict[str, Any]) → Iterable[None]

A context manager to temporarily set the node’s state.

Parameters:

state (dict[str, Any]) – A dictionary representing the state to temporarily set.

class hyped.core.nodes.base.BaseNodeConfig

Bases: BaseConfig

Base configuration class for nodes in a data flow graph.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class hyped.core.nodes.base.NodeProtocol(*args, **kwargs)

Bases: Protocol, Generic[Params, Return]

Protocol for node-like objects in a data flow graph.

This protocol defines the interface for nodes within a data flow graph, including methods for calling nodes with or without an explicit data flow. The NodeProtocol is parameterized by Params, defining the arguments, and Return, defining the return type.

call(*args: ~Params, **kwargs: ~Params) → Return

call(flow: AbstractDataFlow, *args: ~Params, **kwargs: ~Params) → Return

class hyped.core.nodes.base.ProcessMode(batched: bool, backend: Literal['python', 'arrow'])

Bases: object

Represents a processing mode for converting data to different formats.

F = ~F

backend: Literal['python', 'arrow']

Specifies the backend used for processing.

batched: bool

Indicates whether the processing mode operates on batched data.

decorate(fn: Callable[[P], R]) → Callable[[P], R]

Decorates a function by associating it with this ProcessMode.

Parameters:

fn (Callable[P, R]) – The function to decorate.

Returns:

The decorated function with the process mode attribute set.

Return type:

Callable[P, R]

finalize(ctx: RunContext, outputs: Iterable[Any]) → Array

Finalizes the data after processing by converting outputs to Arrow.

Parameters:

• ctx (RunContext) – The context describing the current processing state. This argument expects the original context, not the prepared context generated by ProcessMode.prepare.

• outputs (Iterable[Any]) – The processed outputs to convert.

Returns:

The final Arrow array representation of the outputs.

Return type:

pa.Array

from_arrow_converters: ClassVar[dict[ProcessMode, Callable]] = {ProcessMode(batched=False, backend='python'): <function _arrow_to_python_samples>, ProcessMode(batched=True, backend='arrow'): <function _arrow_to_arrow_batch>, ProcessMode(batched=True, backend='python'): <function _arrow_to_python_batch>}

From arrow converter registry

A registry of converter functions for transforming data from PyArrow to the specified processing mode.

classmethod from_decorated_fn(fn: Callable[[P], R]) → ProcessMode

Retrieves the ProcessMode associated with a decorated function.

Parameters:

fn (Callable[P, R]) – The function from which to retrieve the process mode.

Returns:

The associated process mode.

Return type:

ProcessMode

Raises:

RuntimeError – If the function does not have an associated process mode.

prepare(ctx: RunContext, **kwargs: Array) → Iterable[tuple[RunContext, dict[str, Any]]]

Prepares the data for processing in the current mode.

Converts inputs to the required format using the registered from-Arrow converter.

Parameters:

• ctx (RunContext) – The context describing the current processing state.

• **kwargs (pa.Array) – Input data arrays to convert.

Yields:

Iterable[tuple[RunContext, dict[str, Any]]] – A sequence of contexts and converted data.

classmethod register_from_arrow_converter(mode: ProcessMode) → Callable[[F], F]

Decorator to register a function as a from-Arrow converter.

Parameters:

mode (ProcessMode) – The process mode to register the converter for.

Returns:

A callable that registers the function as a converter.

Return type:

Any

classmethod register_to_arrow_converter(mode: ProcessMode) → Callable[[F], F]

Decorator to register a function as a to-Arrow converter.

Parameters:

mode (ProcessMode) – The process mode to register the converter for.

Returns:

A callable that registers the function as a converter.

Return type:

Any

set_default(fn: Callable[[P], R]) → Callable[[P], R]

Sets this ProcessMode as the default for a function.

Parameters:

fn (Callable[P, R]) – The function to set the default process mode for.

Returns:

The function with the default process mode applied.

Return type:

Callable[P, R]

to_arrow_converters: ClassVar[dict[ProcessMode, Callable]] = {ProcessMode(batched=False, backend='python'): <function _python_samples_to_arrow>, ProcessMode(batched=True, backend='arrow'): <function _arrow_batch_to_arrow>, ProcessMode(batched=True, backend='python'): <function _python_batch_to_arrow>}

To arrow converter registry.

A registry of converter functions for transforming data to PyArrow from the specified processing mode.

validate() → ProcessMode

Validates that the ProcessMode is fully supported.

Returns:

The validated process mode.

Return type:

ProcessMode

Raises:

Exception – If the ProcessMode is not supported (missing converters).

class hyped.core.nodes.base.RunContext(session: None | RunSession, node_id: str, index: int | list[int], rank: int, input_dtype: MappingType, output_dtype: None | DType, target_batch_size: None | int)

Bases: object

Context information for the node execution.

Serves as an identifier for the specific call to the node within the data flow graph. This is particularly useful when a single node class is used multiple times in a data flow, as RunContext identifies the specific instance of the node call, i.e., the specific node in the flow graph.

index: int | list[int]

The index or list of indices associated with the processor execution.

This attribute is used to track the position or set of positions for processing data within a specific processor call. It could represent a single index or a list of indices, depending on how the data is partitioned or processed.

input_dtype: MappingType

The expected input data type for the processor.

This attribute defines the type of the data that the processor is designed to handle as input. It typically maps the input data’s structure or schema, providing context for how the data should be processed.

node_id: str

The id of the node in the data flow graph.

This attribute serves as a unique identifier for the context, ensuring each processor call can be distinctly recognized within the flow graph.

output_dtype: None | DType

The type of data the processor will produce as output.

This attribute defines the expected structure or type of the output that the processor will generate. It provides information about the transformation or processing that the input data undergoes and the format of the resulting data.

rank: int

The rank or position of the processor in a parallelized execution.

This attribute identifies the specific rank or position of the processor in a parallelized system (e.g., in distributed or multi-threaded processing). The rank determines the processor’s order or responsibility for a portion of the data during execution.

session: None | RunSession

The run session instance.

target_batch_size: None | int

The expected target batch size.

This attribute defines the batch size of the target partition of the node if set.

class hyped.core.nodes.base.RunSession

Bases: object

Run session for the node execution.

A RunSession instance is shared along all nodes of a data flow and lives for the whole execution of a data flow. It manages the event loop that runs the data flow execution and allows to store context information.

get_context(key: Hashable) → Any

Retrieve the context value associated with a specific key.

Parameters:

key (Hashable) – The key for the context entry to retrieve.

Returns:

The value associated with the key, or None if the key is not found.

Return type:

Any

property session_id: UUID

The unique session ID.

Returns:

A unique identifier for the session.

Return type:

UUID

set_context(key: Hashable, value: Any) → None

Set a context value associated with a specific key.

Parameters:

• key (Hashable) – The key for the context entry.

• value (Any) – The value to associate with the key.

hyped.core.nodes.base.process_mode(batched: bool = False, backend: Literal['python', 'arrow'] = 'python') → Callable[[F], F]

Decorator to specify the processing mode of a data processing function.

This decorator associates a function with a ProcessMode, specifying how the function handles its inputs and outputs during execution. The mode defines whether the function operates in batched or non-batched mode and which backend is used for processing (e.g., “python” or “arrow”).

Use the @process_mode(...) decorator to annotate methods or functions that perform data processing. The decorator ensures the function is tagged with the appropriate processing mode, which can be validated or used during runtime.

Parameters:

• batched (bool) – Indicates if the function processes data in batches.

• backend (Backend) – Specifies the backend used for processing.

Returns:

The decorator function that associates a function with the specified ProcessMode instance.

Return type:

Callable[[F], F]