hyped.core.ops.numeric module

This module defines a collection of data processors that implement common numeric operations.

Each processor is designed to handle a specific numeric computation, such as addition, subtraction, negation, or computing absolute values. These processors are intended for use in data processing pipelines, enabling efficient and batched execution using Apache Arrow as the backend.

These processors are registered as methods on the respective numeric feature classes, such as IntFeature and FloatFeature, allowing them to be applied directly to numeric features.

class hyped.core.ops.numeric.Abs(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[AbsConfig]

Data Processor for computing absolute values.

T = ~T

process(ctx: RunContext, x: T) → T

Computes the absolute value of the input.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (T) – The numeric input data.

Returns:

The absolute value of the input data.

Return type:

T

class hyped.core.ops.numeric.AbsConfig

Bases: BaseDataProcessorConfig

Configuration for the Abs processor.

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.ops.numeric.Add(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[AddConfig]

Data Processor implementing addition.

This processor takes two numeric inputs, x and y, and computes their sum.

process(ctx: RunContext, x: ScalarType, y: ScalarType) → ScalarType

Perform addition of two numeric inputs.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (Numeric) – The first numeric operand.

• y (Numeric) – The second numeric operand.

Returns:

The result of adding x and y.

Return type:

Numeric

class hyped.core.ops.numeric.AddConfig

Bases: BaseDataProcessorConfig

Configuration for the Add processor.

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.ops.numeric.FloorDiv(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[FloorDivConfig]

Data Processor implementing floor division.

This processor takes two numeric inputs, x and y, and computes their integer quotient (x // y).

process(ctx: RunContext, x: ScalarType, y: ScalarType) → wrapped_resolver, json_schema_input_type=PydanticUndefined)]

Perform floor division of two numeric inputs.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (ScalarType) – The numerator.

• y (ScalarType) – The denominator.

Returns:

The integer quotient of x and y.

Return type:

Float

class hyped.core.ops.numeric.FloorDivConfig

Bases: BaseDataProcessorConfig

Configuration for the floor division processor.

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.ops.numeric.Max(*args: Any, **kwargs: Any)

Bases: BaseDataAggregator[MaxConfig]

Data Aggregator implementing maximum value computation.

async extract(ctx: RunContext, val: ScalarType) → ScalarType

Extract the maximum value from a batch of inputs.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – A batch of numeric values.

Returns:

The maximum value from the input batch.

Return type:

ScalarType

seed(ctx: RunContext) → tuple[ScalarType, bool]

Initialize the aggregator’s state.

Parameters:

ctx (RunContext) – The execution context for the aggregator.

Returns:

A tuple containing the initial state value (default maximum) and a boolean indicating whether the state is initialized.

Return type:

tuple[ScalarType, bool]

async update(ctx: RunContext, val: ScalarType, state: bool, extracted: ScalarType) → tuple[ScalarType, bool]

Update the aggregator’s state with a new value.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – The current state value.

• state (bool) – A boolean indicating whether the state is initialized.

• extracted (ScalarType) – The extracted value from the input.

Returns:

A tuple containing the updated state value (maximum) and a boolean indicating whether the state is initialized.

Return type:

tuple[ScalarType, bool]

class hyped.core.ops.numeric.MaxConfig(*, default: int | float = 0)

Bases: BaseDataAggregatorConfig

Configuration for the Max aggregator.

default: int | float

The default initial value for the maximum.

model_config = {'extra': 'forbid', 'validate_default': True}

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

class hyped.core.ops.numeric.Mean(*args: Any, **kwargs: Any)

Bases: BaseDataAggregator[MeanConfig]

Data Aggregator implementing the mean (average) calculation.

StateType: TypeAlias

alias of tuple[ScalarType, int]

async extract(ctx: RunContext, val: ScalarType) → tuple[ScalarType, int]

Extracts the values to be aggregated.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – The current value to be processed.

Returns:

The current sum and count for aggregation.

Return type:

StateType

seed(ctx: RunContext) → tuple[ScalarType, tuple[ScalarType, int]]

Initializes the state for the mean calculation.

Parameters:

ctx (RunContext) – The execution context for the aggregator.

Returns:

The initial state, which includes the starting sum and count of values.

Return type:

tuple[ScalarType, StateType]

async update(ctx: RunContext, val: ScalarType, state: tuple[ScalarType, int], extracted: ScalarType) → tuple[~typing.Annotated[~hyped.core.features.features.Float64Feature | ~hyped.core.features.features.Float32Feature | float | list[float] | ~pyarrow.lib.DoubleScalar | ~pyarrow.lib.FloatScalar | ~pyarrow.lib.DoubleArray | ~pyarrow.lib.FloatArray, ~hyped.core.features.validators.FeatureResolver(func=~hyped.core.features.validators.FeatureResolver.__init__.<locals>.wrapped_resolver, json_schema_input_type=PydanticUndefined)], tuple[~hyped.core.ops.numeric.ScalarType, int]]

Updates the state by computing the mean after processing a new value.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – The current value to be processed.

• state (StateType) – The current state (sum and count).

• extracted (ScalarType) – The extracted sum and count from the previous step.

Returns:

The updated mean and the new state (sum and count).

Return type:

tuple[Float, StateType]

class hyped.core.ops.numeric.MeanConfig(*, start: int | float = 0, start_count: int = 0)

Bases: BaseDataAggregatorConfig

Configuration for the Mean aggregator.

model_config = {'extra': 'forbid', 'validate_default': True}

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

start: int | float

The default initial value for the mean calculation.

start_count: int

The initial count of values processed, used for mean calculation.

class hyped.core.ops.numeric.Min(*args: Any, **kwargs: Any)

Bases: BaseDataAggregator[MinConfig]

Data Aggregator implementing minimum value computation.

async extract(ctx: RunContext, val: ScalarType) → ScalarType

Extract the minimum value from a batch of inputs.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – A batch of numeric values.

Returns:

The minimum value from the input batch.

Return type:

ScalarType

seed(ctx: RunContext) → tuple[ScalarType, bool]

Initialize the aggregator’s state.

Parameters:

ctx (RunContext) – The execution context for the aggregator.

Returns:

A tuple containing the initial state value (default minimum) and a boolean indicating whether the state is initialized.

Return type:

tuple[ScalarType, bool]

async update(ctx: RunContext, val: ScalarType, state: bool, extracted: ScalarType) → tuple[ScalarType, bool]

Update the aggregator’s state with a new value.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – The current state value.

• state (bool) – A boolean indicating whether the state is initialized.

• extracted (ScalarType) – The extracted value from the input.

Returns:

A tuple containing the updated state value (minimum) and a boolean indicating whether the state is initialized.

Return type:

tuple[ScalarType, bool]

class hyped.core.ops.numeric.MinConfig(*, default: int | float = 0)

Bases: BaseDataAggregatorConfig

Configuration for the Min aggregator.

default: int | float

The default initial value for the minimum.

model_config = {'extra': 'forbid', 'validate_default': True}

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

class hyped.core.ops.numeric.Multiply(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[MultiplyConfig]

Data Processor implementing multiplication.

This processor takes two numeric inputs, x and y, and computes their product (x * y).

process(ctx: RunContext, x: ScalarType, y: ScalarType) → ScalarType

Perform multiplication of two numeric inputs.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (ScalarType) – The first numeric operand.

• y (ScalarType) – The second numeric operand.

Returns:

The product of x and y.

Return type:

ScalarType

class hyped.core.ops.numeric.MultiplyConfig

Bases: BaseDataProcessorConfig

Configuration for the Multiply processor.

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.ops.numeric.Negate(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[NegateConfig]

Data Processor for negating numeric values.

T = ~T

process(ctx: RunContext, x: T) → wrapped_resolver, json_schema_input_type=PydanticUndefined)]

Computes the negation of the input.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (T) – The numeric input data.

Returns:

The negated value of the input data.

Return type:

T

class hyped.core.ops.numeric.NegateConfig

Bases: BaseDataProcessorConfig

Configuration for the Negate processor.

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.ops.numeric.Subtract(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[SubtractConfig]

Data Processor implementing subtraction.

This processor takes two numeric inputs, x and y, and computes their difference (x - y).

process(ctx: RunContext, x: ScalarType, y: ScalarType) → ScalarType

Perform subtraction of two numeric inputs.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (ScalarType) – The first numeric operand.

• y (ScalarType) – The second numeric operand.

Returns:

The result of subtracting y from x.

Return type:

ScalarType

class hyped.core.ops.numeric.SubtractConfig

Bases: BaseDataProcessorConfig

Configuration for the Subtract processor.

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.ops.numeric.Sum(*args: Any, **kwargs: Any)

Bases: BaseDataAggregator[SumConfig]

Data Aggregator implementing sum value computation.

async extract(ctx: RunContext, val: ScalarType) → ScalarType

Extract the sum value from a batch of inputs.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – A batch of numeric values.

Returns:

The sum of the values in the input batch.

Return type:

ScalarType

seed(ctx: RunContext) → tuple[ScalarType, None]

Initialize the aggregator’s state.

Parameters:

ctx (RunContext) – The execution context for the aggregator.

Returns:

A tuple containing the initial state value (default sum) and the unused aggregation state.

Return type:

tuple[ScalarType, None]

async update(ctx: RunContext, val: ScalarType, state: None, extracted: ScalarType) → tuple[ScalarType, None]

Update the aggregator’s state with a new value.

Parameters:

• ctx (RunContext) – The execution context for the aggregator.

• val (ScalarType) – The current state value.

• state (None) – Aggregation state, unused for summation.

• extracted (ScalarType) – The extracted value from the input.

Returns:

A tuple containing the updated value (sum) and the state (None).

Return type:

tuple[ScalarType, None]

class hyped.core.ops.numeric.SumConfig(*, start: int | float = 0)

Bases: BaseDataAggregatorConfig

Configuration for the Sum aggregator.

model_config = {'extra': 'forbid', 'validate_default': True}

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

start: int | float

The default initial value for the sum.

class hyped.core.ops.numeric.TrueDiv(*args: Any, **kwargs: Any)

Bases: BaseDataProcessor[TrueDivConfig]

Data Processor implementing true division.

This processor takes two numeric inputs, x and y, and computes their quotient (x / y).

process(ctx: RunContext, x: ScalarType, y: ScalarType) → wrapped_resolver, json_schema_input_type=PydanticUndefined)]

Perform division of two numeric inputs.

Parameters:

• ctx (RunContext) – The execution context for the processor.

• x (ScalarType) – The numerator.

• y (ScalarType) – The denominator.

Returns:

The quotient of x and y.

Return type:

Float

class hyped.core.ops.numeric.TrueDivConfig

Bases: BaseDataProcessorConfig

Configuration for the true division processor.

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

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

hyped.core.ops.numeric.add_constant(val: Any, candidate_dtype: DType, builder: AbstractDataFlowGraphBuilder) → Feature

Adds a constant value to a data flow graph as a node and returns it as a Feature.

This function evaluates the provided constant value’s data type. If the candidate data type matches the value (e.g., a float value with a float type or an integer value with an integer type), it uses the candidate type. Otherwise, it infers the data type from the value. The constant is then added to the graph as a node.

Parameters:

• val (Any) – The constant value to be added to the graph.

• candidate_dtype (DType) – The candidate data type for the value, expected to be a type like Float32Type or Int64Type.

• builder (AbstractDataFlowGraphBuilder) – The data flow graph builder to add the constant.

Returns:

A Feature object representing the constant added to the graph.

Return type:

Feature

hyped.core.ops.numeric.handle_constant_for_binary_operation(f: Callable[[Feature, Feature], Feature]) → Callable[[Any, Any], Feature]

A decorator that enables binary operations to handle constant values.

This decorator ensures that if one operand in a binary operation is a constant, it is converted into Feature objects before the operation is performed. It wraps the provided binary operation function and seamlessly supports constants as inputs.

Parameters:

f (Callable[[Feature, Feature], Feature]) – The binary operation function that operates on two Feature objects.

Returns:

A wrapped function that handles constants by converting them into Feature objects when necessary, then performs the binary operation.

Return type:

Callable[[Any, Any], Feature]

hyped.core.ops.numeric.mean(feature: Feature, start: int | float = 0, start_count: int = 0) → Feature

Computes the mean (average) value of a feature.

Parameters:

• feature (Feature) – The feature whose mean value is to be computed.

• start (int | float, optional) – The initial sum value to include in the computation. Defaults to 0.

• start_count (int, optional) – The initial count of values to include in the computation. Defaults to 0.

Returns:

A new feature representing the mean value.

Return type:

Feature

hyped.core.ops.numeric.register_all(name: str, types: type[Feature]) → Callable[[Fn], Fn]

Helper function to register a method to multiple feature types.

Parameters:

• name (str) – The name of the method to register.

• types (type[Feature]) – A collection of feature types for which the method will be registered.

Returns:

A decorator that wraps the function, registering it for the

specified feature types.

Return type:

Callable[[Fn], Fn]