hyped.core.nodes.augmentor module

Provides base classes for data augmentation in a data flow graph.

This module defines the base classes for data augmentation tasks within a data flow graph framework. Data augmentors are responsible for filtering or generating new data samples from on existing ones.

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

Bases: BaseNode[C], ABC

Base class for data augmentors in a data flow graph.

This class represents a data augmentor node in a data flow graph. Data augmentors modify or generate new samples from existing ones, which can include filtering or creating new data points. Subclasses of BaseDataAugmentor must implement either the process or the batch_process method to define how the augmentation is applied to the input data.

infer_output_partition(ctx: RunContext, partition: str) → str[source]

Determine the output partition of the augmentater.

By default, data augmentors point to their own partition. This method reuses the node ID of the augmentor as the output partition ID.

Note that this function is not part of the data flow execution but of the initialization process. Therefore, it is called before initialize() of the node and the ctx.session will be None.

Parameters:

  • ctx (RunContext) – Execution context for the node.

  • partition (PartitionId) – The ID of the input partition, i.e. the partition that the node is assigned to.

Returns:

The output partition ID, corresponding to the node ID of the augmentor.

Return type:

PartitionId

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

Defines the augmentation logic to be applied.

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

Parameters:

  • ctx (RunContext) – Context information for the data augmentor’s execution.

  • *args (Feature) – Positional input arguments.

  • **kwargs (Feature) – Keyword arguments.

Returns:

  • Iterable[Feature]: If the process function is synchronous, it returns an iterable of augmented output samples, potentially producing multiple outputs per input.

  • AsyncIterable[Feature]: If the process function is asynchronous, it returns an async iterable of augmented output samples, following the same logic as the synchronous mode.

  • tuple[Feature, TraceIndexList]: If the process function operates in batched mode, it returns:

    • Feature: A batch of augmented output samples.

    • TraceIndexList: A list of trace indices mapping each output sample to the corresponding source sample in the input batch. Specifically, the i-th output sample originates from the trace_index[i]-th input example.

Return type:

Union[Iterable[Feature], AsyncIterable[Feature], tuple[Feature, TraceIndexList]]

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

Execute the main processing logic for the data augmentor.

This method serves as the primary entry point for processing data within a data flow graph, returning both the processed outputs and their associated trace indices. 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. Depending on whether the processing is batched: - If batched:

    1. Separate the outputs and their associated trace indices from the process method’s results.

    2. Concatenate the trace indices and finalize the outputs as a PyArrow array.

    • If non-batched: a. Collect all outputs and trace indices from the process method,

      consuming asynchronous or synchronous iterators as appropriate.

      1. Chain the outputs and finalize them as a PyArrow array.

      2. Build a trace index list that maps each output sample back to its corresponding input.

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:

A tuple containing the processed output as a PyArrow array and a list of trace indices mapping the output samples to their respective input sources.

Return type:

tuple[pa.Array, TraceIndexList]

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

Raises:

TypeError – If the return type of process() is not an iterable.

class hyped.core.nodes.augmentor.BaseDataAugmentorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data augmentors.

This class serves as the base configuration for data augmentors, inheriting from BaseNodeConfig to provide configuration functionality specifically for data augmentation tasks.

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

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