hyped.core.nodes.processor module

Provides base classe for data processors in a data flow graph.

This module defines the base class for data processors, which represent nodes in a data flow graph. It includes generic classes for defining data processors with configurable input and output types.

class hyped.core.nodes.processor.BaseDataProcessor(*args: Any, **kwargs: Any)[source]

Bases: BaseNode[C], ABC

Base class for data processors in a data flow graph.

This class serves as the base for all data processors, representing nodes in a data flow graph. Subclasses of BaseDataProcessor implement specific process functions that map input features to output features. Custom data processors must either override the batch_process() method or the process() method to define their processing logic.

abstractmethod process(ctx: RunContext, *args: Feature | Any | list[Any] | Scalar | Array, **kwargs: Feature | Any | list[Any] | Scalar | Array) → Feature | Any | list[Any] | Scalar | Array[source]

Process function.

This method should be implemented by subclasses to define the processing logic. It may either be synchronous or asynchronous, depending on the subclass.

Parameters:

  • ctx (RunContext) – The context for the current process call.

  • *args (Feature) – Positional feature arguments.

  • **kwargs (Feature) – Keyword feature arguments.

Returns:

The resulting processed feature.

Return type:

Feature

async run(ctx: RunContext, arrays: dict[str, Array]) → Array[source]

Execute the main processing logic for the data processor.

This method serves as the primary entry point for processing data within a data flow graph. It orchestrates the execution of the process method according to the configured ProcessMode, handling input preparation, processing, and output finalization. In detail the workflow is:

  1. Determine the processing mode (ProcessMode) based on the process method’s configuration.

  2. Prepare the input data using the ProcessMode.prepare method.

  3. Apply the process method to the prepared inputs. If the method is asynchronous, the outputs are awaited using asyncio.gather.

  4. Finalize the outputs using the ProcessMode.finalize method, which ensures that the results are correctly formatted as an PyArrow array.

Parameters:

  • ctx (RunContext) – The execution context containing.

  • arrays (dict[str, pa.Array]) – A dictionary mapping input names to PyArrow arrays, representing the input data to be processed.

Returns:

The processed output as a PyArrow array.

Return type:

pa.Array

property signature: Signature

Get the signature of the process() method.

Returns the signature of the process() method with the ctx parameter removed, keeping only the feature inputs modeled in the data flow graph.

Returns:

The signature of the process() method excluding the ctx parameter.

Return type:

inspect.Signature

class hyped.core.nodes.processor.BaseDataProcessorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data processors.

This class serves as the base configuration class for data processors. It inherits from BaseNodeConfig, a Pydantic model, providing basic configuration functionality for data processing tasks.

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

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