hyped.core.ops.sequence module¶

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

Each processor is designed to handle a specific sequence transformation or query, such as calculating lengths, slicing or aggregation operations. These processors are intended for use in data processing pipelines, where they can be applied in a batched and efficient manner using Apache Arrow as the backend.

These processors are registered as methods on the SequenceFeature class, allowing them to be applied directly to sequence features.

class hyped.core.ops.sequence.SequenceGetItem(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceGetItemConfig]

Processor to retrieve an item from a sequence based on a specific index.

This processor extracts a single element from a sequence at the position specified by the index argument.

Retrieve an item from a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[ItemType]) – Input sequence from which to extract the item.
index (Int) – Index to get.

Returns:

The item at the specified index in the sequence.

Return type:

ItemType

class hyped.core.ops.sequence.SequenceGetItemConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequenceGetItem 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.sequence.SequenceGetItems(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceGetItemsConfig]

Processor to retrieve an item from a sequence based on a specific index.

This processor extracts a subsequence from a sequence at the positions specified by the index argument.

Retrieve an item from a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[ItemType]) – Input sequence from which to extract the item.
index (Sequence[Int]) – Sequence of indices to get.

Returns:

The items at the specified indices in the sequence.

Return type:

Sequence[ItemType]

class hyped.core.ops.sequence.SequenceGetItemsConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequenceGetItems 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.sequence.SequenceGetSlice(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceGetSliceConfig]

Processor to extract a slice from a sequence.

This processor extracts a sub-sequence from a given sequence using slicing parameters defined in its configuration.

Extract a slice from a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[ItemType]) – Input sequence from which to extract the slice.

Returns:

The sub-sequence defined by the start, stop, and step indices.

Return type:

Sequence[ItemType]

class hyped.core.ops.sequence.SequenceGetSliceConfig(*, start: int, stop: None | int, step: int)[source]¶

Bases: BaseDataAugmentorConfig

Configuration class for the SequenceGetSlice processor.

model_config = {'extra': 'forbid', 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

start: int¶: The starting index of the slice (inclusive).

step: int¶: The step size for the slice.

stop: None | int¶: The ending index of the slice (exclusive).

class hyped.core.ops.sequence.SequenceIndex(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceIndexConfig]

Processor to find the index of a specific value within a sequence.

Find the first index of a given value in a sequence.

This method attempts to locate the first occurrence of val within seq.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[ItemType]) – The input sequence to search within.
val (ItemType) – The value to search for in the sequence.
default (None | ItemType, optional) – The default value to return if val is not found in seq.

Returns:

The zero-based index of the first occurrence of val in seq. If val is not found and a default value was provided, the default value is returned.

Return type:

Int32

Raises:

ValueError – If val is not found in seq and default is None.

class hyped.core.ops.sequence.SequenceIndexConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration for the SequenceIndex 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.sequence.SequenceLength(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceLengthConfig]

Data processor for computing the length of sequences.

process(ctx: RunContext, seq: SequenceFeature[T] | list[T] | list[list[T]] | ListScalar | ListArray) → Int32Feature[source]¶

Computes the length of an input sequence.

Parameters:

ctx (RunContext) – The execution context.
seq (Sequence) – The sequence to compute the length for.

Returns:

The lengths of the input sequences.

Return type:

IntFeature

class hyped.core.ops.sequence.SequenceLengthConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration for the SequenceLength 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.sequence.SequenceMax(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceMaxConfig]

Processor to compute the maximum value of a numeric sequence.

process(ctx: RunContext, seq: SequenceFeature[NumericType] | list[NumericType] | list[list[NumericType]] | ListScalar | ListArray) → NumericType[source]¶

Compute the maximum value of a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[NumericType]) – Input sequence of numeric values.

Returns:

The maximum value in the sequence, or the default value specified in the configuration if the sequence is empty.

Return type:

NumericType

class hyped.core.ops.sequence.SequenceMaxConfig(*, default: None | int | float = None)[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequenceMax processor.

default: None | int | float¶: The default value to return if the sequence is empty.

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.sequence.SequenceMin(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceMinConfig]

Processor to compute the minimum value of a numeric sequence.

process(ctx: RunContext, seq: SequenceFeature[NumericType] | list[NumericType] | list[list[NumericType]] | ListScalar | ListArray) → NumericType[source]¶

Compute the minimum value of a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[NumericType]) – Input sequence of numeric values.

Returns:

The minimum value in the sequence, or the default value specified in the configuration if the sequence is empty.

Return type:

NumericType

class hyped.core.ops.sequence.SequenceMinConfig(*, default: None | int | float = None)[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequenceMin processor.

default: None | int | float¶: The default value to return if the sequence is empty.

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.sequence.SequencePack(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataAugmentor[SequencePackConfig]

Augmentor to reconstruct a sequence from elements and trace indices.

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

Infer the output partition of the pack node.

If the original_partition configuration value is set, the output partition of the node is explicitly set to the original_partition. Otherwise, the output partition is determined by the default logic in the parent class.

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

process(ctx: RunContext, values: ItemType, trace_index: Int32Feature | int | list[int] | Int32Scalar | Int32Array) → tuple[~typing.Annotated[~hyped.core.features.features.SequenceFeature[~hyped.core.ops.sequence.ItemType] | list[~hyped.core.ops.sequence.ItemType] | list[list[~hyped.core.ops.sequence.ItemType]] | ~pyarrow.lib.ListScalar | ~pyarrow.lib.ListArray, ~hyped.core.features.validators.FeatureResolver(func=~hyped.core.features.validators.FeatureResolver.__init__.<locals>.wrapped_resolver, json_schema_input_type=PydanticUndefined)], list[int]][source]¶

Reconstruct a sequence from flattened elements and trace indices.

Parameters:

ctx (RunContext) – Context object containing runtime information.
values (T) – The values to pack into a sequence.
trace_index (Int32) – Indices tracing the original sequence structure.

Returns:

A tuple containing the reconstructed sequence and the offsets for the sequence elements.

Return type:

tuple[Sequence[T], TraceIndexList]

class hyped.core.ops.sequence.SequencePackConfig(*, original_partition: None | str = None, original_length: None | int = None)[source]¶

Bases: BaseDataAugmentorConfig

Configuration class for the SequencePackUnpacked augmentor.

model_config = {'extra': 'forbid', 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

original_length: None | int¶: The length of the sequence before the unpack operation if defined.

original_partition: None | str¶: The partition of unpack node in case the values come from a sequence unpack operation.

class hyped.core.ops.sequence.SequencePad(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequencePadConfig]

Processor to pad sequences to a specified length.

This processor pads input sequences with a specified fill value to either a user-defined target length or, if not provided, the length of the longest sequence in the current batch.

process(ctx: RunContext, seq: SequenceFeature[ItemType] | list[ItemType] | list[list[ItemType]] | ListScalar | ListArray, fill_value: ItemType) → wrapped_resolver, json_schema_input_type=PydanticUndefined)][source]¶

Pad sequences to a uniform length with a fill value.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[ItemType]) – A batch of sequences to be padded.
fill_value (ItemType) – The value to use for padding.

Returns:

A batch of padded sequences. Each sequence will have the same length, either matching the user-specified target length or the longest sequence in the batch if no target length is specified.

Return type:

Sequence[ItemType]

class hyped.core.ops.sequence.SequencePadConfig(*, length: None | int = None)[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequencePad processor.

length: None | int¶

The target length to pad each sequence to.

If None, the processor pads sequences to match the length of the longest sequence in the current batch.

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.sequence.SequenceSum(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceSumConfig]

Processor to compute the sum of numeric values in a sequence.

process(ctx: RunContext, seq: SequenceFeature[NumericType] | list[NumericType] | list[list[NumericType]] | ListScalar | ListArray) → NumericType[source]¶

Compute the sum of a sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[NumericType]) – Input sequence of numeric values.

Returns:

The sum of the values in the sequence, or the default value specified in the configuration if the sequence is empty.

Return type:

NumericType

class hyped.core.ops.sequence.SequenceSumConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration class for the SequenceSum 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.sequence.SequenceUnpack(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataAugmentor[SequenceUnpackConfig]

Augmentor to unpack a sequence.

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

Determine the output partition of the unpack operation.

For fixed-length sequences the output partition is deterministically computed from the node partition and the length of the sequence. This allows to combine elements from different unpacked sequences.

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

process(ctx: RunContext, seq: SequenceFeature[ItemType] | list[ItemType] | list[list[ItemType]] | ListScalar | ListArray) → tuple[ItemType, list[int]][source]¶

Unpack the sequence.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[T]) – Input sequence to be flattened.

Returns:

A tuple containing the flattened sequence with indices and the computed trace indices.

Return type:

tuple[FlatSequenceWithIndex[T], TraceIndexList]

class hyped.core.ops.sequence.SequenceUnpackConfig[source]¶

Bases: BaseDataAugmentorConfig

Configuration class for the SequenceUnpack augmentor.

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.sequence.SequenceUnpackWithIndex(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataAugmentor[SequenceUnpackWithIndexConfig]

Augmentor to unpack a sequence and compute trace indices.

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

Determine the output partition of the unpack operation.

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

process(ctx: RunContext, seq: SequenceFeature[ItemType] | list[ItemType] | list[list[ItemType]] | ListScalar | ListArray) → tuple[SequenceValueWithIndex[ItemType], list[int]][source]¶

Unpack a sequence and compute trace indices.

Parameters:

ctx (RunContext) – Context object containing runtime information.
seq (Sequence[T]) – Input sequence to be flattened.

Returns:

A tuple containing the flattened sequence with indices and the computed trace indices.

Return type:

tuple[FlatSequenceWithIndex[T], TraceIndexList]

class hyped.core.ops.sequence.SequenceUnpackWithIndexConfig[source]¶

Bases: BaseDataAugmentorConfig

Configuration class for the SequenceUnpackWithIndex augmentor.

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.sequence.SequenceValueWithIndex(ref: BaseReference, skip_keys: set[str] = <factory>)[source]¶

Bases: _MappingFeature, Generic[ItemType]

Represents a sequence value and the origin batch index.

index: Int32Feature¶: The index of the batch containing the sequence that the value originates from.

value: ItemType¶: The sequence value.

class hyped.core.ops.sequence.SequenceZip(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceZipConfig]

Data Processor for zipping sequences.

process(ctx: ~hyped.core.nodes.base.RunContext, **seqs: ~typing.Annotated[~hyped.core.features.features.SequenceFeature[~hyped.core.ops.sequence.Annotated[~hyped.core.ops.sequence.SequenceZipValType, ~hyped.core.features.validators.MatchFeatures(func=~hyped.core.features.validators.FeatureValidator.__init__.<locals>.wrapped_validator)]] | list[~hyped.core.ops.sequence.Annotated[~hyped.core.ops.sequence.SequenceZipValType, ~hyped.core.features.validators.MatchFeatures(func=~hyped.core.features.validators.FeatureValidator.__init__.<locals>.wrapped_validator)]] | list[list[~hyped.core.ops.sequence.Annotated[~hyped.core.ops.sequence.SequenceZipValType, ~hyped.core.features.validators.MatchFeatures(func=~hyped.core.features.validators.FeatureValidator.__init__.<locals>.wrapped_validator)]]] | ~pyarrow.lib.ListScalar | ~pyarrow.lib.ListArray, ~hyped.core.features.validators.Len(func=~hyped.core.features.validators.FeatureValidator.__init__.<locals>.wrapped_validator)]) → wrapped_validator)][source]¶

Zip multiple sequences.

Parameters:

ctx (RunContext) – Context object containing runtime information.
**seqs (Sequence[SequenceType]) – Input sequences to zip. Zipped sequences are ordered by their keys, which are converted to integers.

Returns:

The zipped sequences.

Return type:

Sequence[Sequence[SequenceType]]

class hyped.core.ops.sequence.SequenceZipConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration for SequenceZip operation.

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.sequence.SequenceZipMapping(*args: Any, **kwargs: Any)[source]¶

Bases: BaseDataProcessor[SequenceZipMappingConfig]

Data Processor for zipping mapping of sequences to sequence of mapping.

process(ctx: ~hyped.core.nodes.base.RunContext, **seqs: ~typing.Annotated[~hyped.core.features.features.SequenceFeature[~hyped.core.ops.sequence.MixedSequenceType] | list[~hyped.core.ops.sequence.MixedSequenceType] | list[list[~hyped.core.ops.sequence.MixedSequenceType]] | ~pyarrow.lib.ListScalar | ~pyarrow.lib.ListArray, ~hyped.core.features.validators.Len(func=~hyped.core.features.validators.FeatureValidator.__init__.<locals>.wrapped_validator)]) → wrapped_validator)][source]¶

Zip a dict-of-sequences into a sequence-of-dicts.

Parameters:

ctx (RunContext) – Context object containing runtime information.
**seqs (Sequence[MixedSequenceType]) – Input sequences to zip. The output mapping depends on the keys of this dictionary.

Returns:

The zipped sequences.

Return type:

Sequence[Mapping]

class hyped.core.ops.sequence.SequenceZipMappingConfig[source]¶

Bases: BaseDataProcessorConfig

Configuration for SequenceZip operation mapping mixed DTypes.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'validate_default': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Reconstruct a sequence from values and trace indices.

This method leverages the SequencePack augmentor to rebuild a sequence from its flattened components, ensuring the sequence is packed with its associated trace indices. Additionally, it ensures that the sequence is assigned to the appropriate partition based on the node performing the flattening operation.

Parameters:

values (T) – The flattened sequence values to be packed.
trace_index (Int32) – The trace indices associated with the values, mapping them to their original structure.
node (None | ConcreteReference) – A reference to the node performing the flattening operation, used to infer the partition for the packing operation.

Returns:

The packed sequence, reconstructed from the flattened values and trace indices, assigned to the appropriate partition for the unflattening operation.

Return type:

Sequence[T]

Retrieve an item or a subsequence from a sequence using an integer index or a slice.

This method uses the SequenceGetItem and SequenceGetSlice processors to handle both single-item retrieval and slicing. It supports sequences with defined or undefined lengths, but imposes restrictions for negative indices or slices when the sequence length is unknown.

Parameters:

sequence (Sequence[ItemType]) – The input sequence feature.
index (int | slice) – The index or slice used for retrieval. Negative indices are supported only for sequences with defined lengths.

Returns:

The retrieved item or subsequence, based on the provided index.

Return type:

ItemType

Raises:

RuntimeError – If negative indices or slices are used with sequences of undefined length.

hyped.core.ops.sequence.sequence_max(seq: SequenceFeature[NumericType] | list[NumericType] | list[list[NumericType]] | ListScalar | ListArray, default: Any = None) → NumericType[source]¶

Compute the maximum value of a sequence using the SequenceMax processor.

Parameters:

seq (Sequence[NumericType]) – Input sequence feature of numeric values.
default (Any) – Default value to return if the sequence is empty. Defaults to None.

Returns:

The feature representing the maximum value in the sequence, or the provided default value if the sequence is empty.

Return type:

NumericType

hyped.core.ops.sequence.sequence_min(seq: SequenceFeature[NumericType] | list[NumericType] | list[list[NumericType]] | ListScalar | ListArray, default: Any = None) → NumericType[source]¶

Compute the minimum value of a sequence using the SequenceMin processor.

Parameters:

seq (Sequence[NumericType]) – Input sequence feature of numeric values.
default (Any) – Default value to return if the sequence is empty. Defaults to None.

Returns:

The feature representing the minimum value in the sequence, or the provided default value if the sequence is empty.

Return type:

NumericType

Pad the sequence to a specified length with a given fill value.

If length is provided, the sequence is padded to the specified length. If length is None, the sequence is padded to match the length of the longest sequence in the current batch.

Parameters:

seq (Sequence[ItemType]) – The sequence to pad.
fill_value (ItemType) – The value to use for padding.
length (None | int) – The desired length to pad the sequence to. Defaults to None, in which case the longest sequence in the batch is used.

Returns:

A new SequenceFeature instance containing the padded sequence.

Return type:

SequenceFeature[T]

Zip multiple sequences together.

Parameters:

*args (Sequence) – Sequences to zip together. Must all have the same Value type.
**kwargs (Sequence) – Sequences to zip together. Can have mixed value types.

Returns:

The zipped sequences. If the input is: a list of sequences the output will be a Sequence[Sequence] too. If the input is a dict of mixed-type sequences, the output will be a Sequence[Mapping].

Return type:

Sequence[Sequence] | Sequence[Mapping]