hyped.core.nodes.aggregator module

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

This module defines the base class for data aggregators, which manage the aggregation of data within a data flow graph. It includes generic classes for defining data aggregators with configurable input types and aggregation logic.

Classes:
class hyped.core.nodes.aggregator.BaseDataAggregator(*args: Any, **kwargs: Any)[source]

Bases: BaseNode[C], ABC

Base class for data aggregators.

This class serves as the base for all data aggregators, defining the necessary interfaces and methods for implementing custom aggregators. Subclasses must implement the extract and update methods, which define the logic for retrieving and updating aggregated values in the data flow graph.

Extracted = ~Extract
State = ~State
Value = ~Value
abstractmethod async extract(ctx: RunContext, *args: Feature, **kwargs: Feature) → Extracted[source]

Extract necessary values from the inputs for aggregation.

Parameters:

  • ctx (RunContext) – The run context object.

  • *args (Feature) – Positional input arguments.

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

Returns:

The extracted context values required for aggregation.

Return type:

Extract

abstractmethod seed(ctx: RunContext) → tuple[Value, State][source]

Compute the seed aggregation value and state.

Note that the seed() 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) – The run context object with session=None.

Returns:

The initial value and state for the aggregator.

Return type:

tuple[Value, State]

property signature: Signature

Retrieve the signature for the aggregator node.

This method constructs a signature for the aggregator node by combining the parameters of the extract() method with the return annotation of the update() method. The ctx parameter is excluded from the parameter list, ensuring that the signature reflects only the feature inputs relevant to the aggregation process.

The return annotation of the signature is derived from the update() method, representing the aggregated feature type produced by the aggregator node.

Returns:

A constructed signature for the aggregator node, with parameters from extract() (excluding ctx) and a return annotation based on the feature type from update().

Return type:

inspect.Signature

Raises:

  • TypeError – If the return annotation of update() does not represent

  • a valid aggregated feature type. –

abstractmethod async update(ctx: RunContext, val: Value, state: State, extracted: Extracted) → tuple[Value, State][source]

Update the aggregation value and context.

Parameters:

  • ctx (RunContext) – The run context object.

  • val (Value) – The current aggregation value.

  • state (State) – The current aggregation state.

  • extracted (Extract) – The values extracted from the input batch.

Returns:

The updated aggregation value and state.

Return type:

tuple[Value, State]

class hyped.core.nodes.aggregator.BaseDataAggregatorConfig[source]

Bases: BaseNodeConfig

Base configuration class for data aggregators.

This class serves as the base configuration class for data aggregators. It inherits from BaseConfig, providing basic configuration functionality for data aggregation tasks.

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.aggregator.DataAggregationManager(aggregators: list[BaseDataAggregator], run_contexts: list[RunContext])[source]

Bases: object

Manager for handling data aggregation operations.

This class manages data aggregators, their thread-safe buffers, and synchronization locks to ensure safe concurrent updates during the data processing.

async aggregate(aggregator: BaseDataAggregator, ctx: RunContext, inputs: dict[str, Array]) → None[source]

Perform aggregation for a batch of inputs.

Parameters:

  • aggregator (BaseDataAggregator) – The aggregator object.

  • inputs (Batch) – The batch of input samples.

  • index (IndexList) – The indices associated with the input samples.

  • rank (Rank) – The rank of the processor in a distributed setting.

  • ctx (RunContext) – Context information for the aggregator execution.

property values_proxy: MappingProxyType

Get a read-only view of the aggregation values.

Returns:

A read-only view of the aggregation values.

Return type:

MappingProxyType[str, pa.Array]