hyped.core.typing module¶

Core typing module.

This module defines the type aliases that need to me used to define node interfaces.

class hyped.core.typing.Annotated(*args, **kwargs)[source]¶

Bases: object

Add context-specific metadata to a type.

Example: Annotated[int, runtime_check.Unsigned] indicates to the hypothetical runtime_check module that this type is an unsigned int. Every other consumer of this type can ignore this metadata and treat this type as int.

The first argument to Annotated must be a valid type.

Details:

It’s an error to call Annotated with less than two arguments.

Access the metadata via the __metadata__ attribute:

assert Annotated[int, '$'].__metadata__ == ('$',)

Nested Annotated types are flattened:

assert Annotated[Annotated[T, Ann1, Ann2], Ann3] == Annotated[T, Ann1, Ann2, Ann3]

Instantiating an annotated type is equivalent to instantiating the

underlying type:

assert Annotated[C, Ann1](5) == C(5)

Annotated can be used as a generic type alias:

Optimized: TypeAlias = Annotated[T, runtime.Optimize()]
assert Optimized[int] == Annotated[int, runtime.Optimize()]

OptimizedList: TypeAlias = Annotated[list[T], runtime.Optimize()]
assert OptimizedList[int] == Annotated[list[int], runtime.Optimize()]

Annotated cannot be used with an unpacked TypeVarTuple:
```
Variadic: TypeAlias = Annotated[*Ts, Ann1]  # NOT valid
```
This would be equivalent to:
```
Annotated[T1, T2, T3, ..., Ann1]
```
where T1, T2 etc. are TypeVars, which would be invalid, because only one type should be passed to Annotated.

hyped.core.typing.Bool¶

Type alias for a boolean.

Supported types include:

Feature Types: BoolFeature
Built-in types: bool, list[bool]
PyArrow scalar types: pyarrow.BooleanScalar
PyArrow array types: pyarrow.BooleanArray

Type:: Bool

alias of BoolFeature | bool | list[bool] | BooleanScalar | BooleanArray

hyped.core.typing.ClassLabel¶: alias of ClassLabelFeature

class hyped.core.typing.ExcludeFieldIf(condition: Callable[[BaseConfig, dict[str, Feature], ValidationSession], bool])[source]¶

Bases: AfterValidator

A Pydantic validator that conditionally excludes a field from the mapping.

This class applies a callable condition to determine whether a field should be excluded from the mapping. The condition is evaluated with the node’s configuration (BaseConfig) and the validation session (ValidationSession) as inputs, returning True to exclude the field or False to include it.

For an usage example, see _MappingFeature.

func: Callable[[Any], Any] | Callable[[Any, ValidationInfo[Any]], Any]¶

hyped.core.typing.Feature¶

Type alias for a feature.

Supported types include:

Feature Types: hyped.core.features.features.Feature
Built-in types: Any, list[Any]
PyArrow scalar types: pyarrow.Scalar
PyArrow array types: pyarrow.Array

Type:: Feature

alias of Feature | Any | list[Any] | Scalar | Array

class hyped.core.typing.FeatureResolver(resolver: Callable[[BaseConfig, dict[str, Feature], ValidationSession], Any])[source]¶

Bases: BeforeValidator

A Pydantic BeforeValidator that resolves ambiguous feature types.

This class is designed to resolve union types for features by allowing custom logic to determine the correct feature type before processing begins. It integrates with Pydantic’s BeforeValidator system to resolve any ambiguity in the feature’s type. It is particularly useful when the feature type needs to change based on configuration or other contextual data.

Usage Example:

Below is an example of how to use FeatureResolver to resolve ambiguous feature types based on a configuration or inputs. The validator allows you to specify a custom feature resolution function, which can be used with the Annotated type hint.

Example: Resolving Union Types Based on Configuration

class CustomConfig(BaseDataProcessorConfig):
    value: int

def custom_feature_resolution(
    config: None | CustomConfig,
    inputs: dict[str, Feature],
    session: ValidationSession
) -> Int:
    # implement custom resolution logic here
    return Int if config.value < 4 else Float

class CustomProcessor(BaseDataProcessor[CustomConfig]):

    def process(
        self, ctx: RunContext, a: Int
    ) -> Annotated[Int | Float, FeatureResolver(custom_feature_resolution)]):
        # processing logic
        ...

In this example, the FeatureResolver is used to resolve the return feature type, which can be either Int or Float. The custom feature resolution function custom_feature_resolution chooses the correct type based on the configuration.

func: Callable[[Any], Any] | Callable[[Any, ValidationInfo[Any]], Any]¶

json_schema_input_type: Any¶

class hyped.core.typing.FeatureValidator(validator: Callable[[Any, None | BaseConfig, ValidationSession], Any])[source]¶

Bases: AfterValidator

A Pydantic AfterValidator class that performs feature validation.

This class is used to validate features in data processing pipelines. It integrates with Pydantic’s AfterValidator system to perform custom validation logic after the feature is processed. It is designed to be used with data flow systems where feature validation is required before performing any further operations.

Usage Example:

Below is an example of how to use FeatureValidator to introduce custom feature validation logic into a data flow. The validator allows you to specify a custom validation function, which can be used with the Annotated type hint.

Example 1: Basic Validation without Config

def custom_feature_validation(
    feature: Int,
    config: None,
    session: ValidationSession
) -> Int:
    # implement custom validation logic here
    return feature

class Inputs(Mapping):
    a: Annotated[Int, FeatureValidator(custom_feature_validation)]
    b: Float

flow = DataFlow[Inputs]()

In this example, the feature a is validated using the FeatureValidator with the custom validation function custom_feature_validation. The config is None because the validation are not specific to any node.

Example 2: Validation with Config in Processor Node

class CustomConfig(BaseDataProcessorConfig):
    value: int

def custom_feature_validation(
    feature: Int,
    config: None | CustomConfig,
    session: ValidationSession
) -> Int:
    # implement custom validation logic here
    return feature

class CustomProcessor(BaseDataProcessor[CustomConfig]):
    def process(
        self,
        ctx: RunContext,
        a: Annotated[Int, FeatureValidator(custom_feature_validation)]
    ) -> Float:
        # custom processing logic
        ...

In this example, the feature a is validated using the FeatureValidator, but the validation function now has access to a CustomConfig for the processor node, allowing for validation specific to the processor’s configuration.

func: Callable[[Any], Any] | Callable[[Any, ValidationInfo[Any]], Any]¶

hyped.core.typing.Float¶

Type alias for a floating-point number of varying precision.

Supported types include:

Feature Types: Float32Feature, Float64Feature
Built-in types: float, list[float]
PyArrow scalar types: pyarrow.FloatScalar, pyarrow.DoubleScalar
PyArrow array types: pyarrow.FloatArray, pyarrow.DoubleArray

Type:: Float

hyped.core.typing.Float32¶

Type alias for a 32-bit floating-point number.

Supported types include:

Feature Types: Float32Feature
Built-in types: float, list[float]
PyArrow scalar types: pyarrow.FloatScalar
PyArrow array types: pyarrow.FloatArray

Type:: Float32

alias of Float32Feature | float | list[float] | FloatScalar | FloatArray

hyped.core.typing.Float64¶

Type alias for a 64-bit floating-point number.

Supported types include:

Feature Types: Float64Feature
Built-in types: float, list[float]
PyArrow scalar types: pyarrow.DoubleScalar
PyArrow array types: pyarrow.DoubleArray

Type:: Float64

alias of Float64Feature | float | list[float] | DoubleScalar | DoubleArray

hyped.core.typing.Index¶

An index, usually corresponding to a sample.

Represents a single integer that refers to a specific sample within the dataset. This is often used to retrieve or reference a particular sample from a dataset.

hyped.core.typing.IndexList¶

A list of dataset indices, usually corresponding to a batch.

Contains integer indices that refer to specific samples within the dataset. This is typically used to track which samples are included in a particular batch or subset of the dataset.

alias of list[int]

hyped.core.typing.Int¶

Type alias for a signed integer of varying bit length.

Supported types include:

Feature Types: Int8Feature, Int16Feature, Int32Feature, Int64Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.Int8Scalar, pyarrow.Int16Scalar, pyarrow.Int32Scalar, pyarrow.Int64Scalar
PyArrow array types: pyarrow.Int8Array, pyarrow.Int16Array, pyarrow.Int32Array, pyarrow.Int64Array

Type:: Int

hyped.core.typing.Int16¶

Type alias for a 16-bit signed integer.

Supported types include:

Feature Types: Int16Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.Int16Scalar
PyArrow array types: pyarrow.Int16Array

Type:: Int16

alias of Int16Feature | int | list[int] | Int16Scalar | Int16Array

hyped.core.typing.Int32¶

Type alias for a 32-bit signed integer.

Supported types include:

Feature Types: Int32Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.Int32Scalar
PyArrow array types: pyarrow.Int32Array

Type:: Int32

alias of Int32Feature | int | list[int] | Int32Scalar | Int32Array

hyped.core.typing.Int64¶

Type alias for a 64-bit signed integer.

Supported types include:

Feature Types: Int64Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.Int64Scalar
PyArrow array types: pyarrow.Int64Array

Type:: Int64

alias of Int64Feature | int | list[int] | Int64Scalar | Int64Array

hyped.core.typing.Int8¶

Type alias for an 8-bit signed integer.

Supported types include:

Feature Types: Int8Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.Int8Scalar
PyArrow array types: pyarrow.Int8Array

Type:: Int8

alias of Int8Feature | int | list[int] | Int8Scalar | Int8Array

class hyped.core.typing.Len[source]¶

class hyped.core.typing.Len(length: int)

class hyped.core.typing.Len(length: Callable[[BaseConfig, int | None, ValidationSession], int | None])

class hyped.core.typing.Len(length: None | int | Callable[[BaseConfig, int | None, ValidationSession], int | None], strict: bool)

class hyped.core.typing.Len(length: int | Callable[[BaseConfig, int | None, ValidationSession], int | None], strict: Literal[True])

Bases: FeatureValidator

A TypeValidator that sets/checks the length of a sequence.

The Len class is used to validate or specify the length of a sequence feature. Its behavior changes depending on whether the sequence’s length is explicitly defined, whether the strict mode is enabled, or whether it is used to ensure multiple sequences have the same length.

Usage Examples:

Example 1: Specifying the Length of a Sequence

When the length of the sequence is not predefined, you can use Len to specify it. This example shows how the Len instance is used to enforce the sequence length.

class Inputs(Mapping):
    seq: Annotated[Sequence[Int], Len(2)]

flow = DataFlow[Inputs]()

In this case, the sequence length of seq will be set to 2.

Example 2: Conflict with Predefined Sequence Length

If the sequence length is already defined, the Len instance will throw an exception when its length conflicts with the predefined length. Here’s an example with a dataset where the length of seq is already set to 1.

features = datasets.Features()
features["seq"] = datasets.Sequence(datasets.Value("Int32"), length=1)
ds = datasets.Dataset.from_dict({"seq": [[0], [1]]}, features=features)

class Inputs(Mapping):
    seq: Annotated[Sequence[Int], Len(2)]

flow = DataFlow[Inputs](ds.features)

In this example, the length of seq is already specified as 1 in the dataset. When the Len instance with length 2 is applied, an exception will be raised because the length 2 does not match the predefined length 1.

Example 3: Ensuring Sequences Have the Same Length

You can use the Len instance to ensure that multiple sequences have the same length without explicitly specifying the length. Here, match_length is used to ensure that both seqA and seqB have the same length.

match_length = Len()

features = datasets.Features()
features["seqA"] = datasets.Sequence(datasets.Value("Int32"), length=1)
features["seqB"] = datasets.Sequence(datasets.Value("Int32"), length=1)

class Inputs(Mapping):
    seqA: Annotated[Sequence[Int], match_length]
    seqB: Annotated[Sequence[Float], match_length]

flow = DataFlow[Inputs](features)

In this example, seqA and seqB are validated to ensure they both have the same length. If their lengths would differ, an exception will be raised.

Example 4: Dynamically Specifying the Length of a Sequence in a Processor Node

The Len class can also be used to dynamically specify the length of a return sequence from a node. Here’s an example where the length of the sequence is determined by the Len instance during processing.

match_length = Len()

class CustomConfig(BaseDataProcessorConfig):
    value: int

class CustomProcessor(BaseDataProcessor[CustomConfig]):

    def process(
        self,
        ctx: RunContext,
        a: Annotated[Sequence[Int], match_length]
    ) -> Annotated[Sequence[Float], match_length]:
        # processing logic
        ...

In this example, the length of the sequence a is dynamically captured by the match_length instance, and the same match_length is applied to the return sequence.

Strict Mode:

When the strict=True argument is passed, the behavior of Len changes:

The length of the sequence must be explicitly defined when creating the Len instance (e.g., Len(4)).
The class only checks that the sequence length matches the expected length, and all logic for setting or capturing the length is disabled.
No dynamic length setting or matching logic occurs in strict mode; it is purely for verifying that the sequence length is exactly as expected.

Example: Using Strict Mode to Enforce a Specific Sequence Length

features = datasets.Features()
features["seq"] = datasets.Sequence(datasets.Value("Int32"), length=4)

class Inputs(Mapping):
    seq: Annotated[Sequence[Int], Len(4, strict=True)]

flow = DataFlow[Inputs](features)

In this case, Len(4, strict=True) will ensure that seq must have exactly 4 elements. No other logic will be applied, and no length will be dynamically set or captured.

Behavior:

If the length of a sequence is not defined beforehand, the Len instance can specify the length in non-strict mode.
If the sequence’s length is predefined, the Len instance checks that the length matches the predefined value.
When used across multiple sequences, the Len instance ensures the sequences have the same length in non-strict mode.
In strict mode, the expected length must be explicitly defined, and only length validation occurs.

func: Callable[[Any], Any] | Callable[[Any, ValidationInfo[Any]], Any]¶

hyped.core.typing.Mapping¶: alias of _MappingFeature

class hyped.core.typing.MatchFeatures[source]¶

Bases: FeatureValidator

A MatchFeatures that ensures features have matching data types.

The MatchFeatures validator is designed to enforce that all features annotated with this validator have the same data type (DType) within a validation session. This is particularly useful when processing multiple features that need to maintain consistency in their data types, such as when aggregating or performing operations across features.

MatchingFeatures must be annotated with the same instance of this class.

Usage Example:

Example 1: Ensuring Matching Data Types

In this example, the MatchFeatures validator ensures that two features have the same data type.

MatchFeat = MatchFeatures()

class Inputs(Mapping):
    featureA: Annotated[Int, MatchFeat]
    featureB: Annotated[Float, MatchFeat]

flow = DataFlow[Inputs]()

If featureA and featureB have different data types, a TypeError will be raised during validation.

Example 2: Validation Across Multiple Nodes

The validator can also be used across multiple nodes in a data flow, ensuring that features maintain consistent data types across processing steps. It can also be used together with TypeVars to ensure the features annotated with the TypeVar match, while the output feature is inferred via the TypeVar.

MatchFeat = MatchFeatures()
T = TypeVar("T", bound=Int)

class CustomProcessor(BaseDataProcessor):
    def process(
        self,
        ctx: RunContext,
        a: Annotated[T, MatchFeat],
        b: Annotated[T, MatchFeat],
    ) -> T:
        # Processing logic ensuring all inputs and outputs have the same DType
        ...

Behavior:

During validation, the first feature annotated with MatchFeatures
captures its data type (DType) in the validation session.
Subsequent features are validated against this captured data type. If their
data type does not match, a TypeError is raised.
This behavior ensures that all features annotated with MatchFeatures
are of the same data type.

func: Callable[[Any], Any] | Callable[[Any, ValidationInfo[Any]], Any]¶

hyped.core.typing.NodeId¶: alias of str

hyped.core.typing.PartitionId¶

An identifier for a partition within the data flow graph.

The PartitionId is a string that uniquely identifies these partitions, enabling the tracking and management of different stages within the data flow graph.

A partition in the data flow graph represents a subgraph where each sample from the dataset is processed or transformed independently of others. Partitions are often introduced during data augmentation processes, where new samples are generated or existing samples are filtered out.

hyped.core.typing.Rank¶: alias of int

hyped.core.typing.Sequence¶

Type alias for a sequence of items of a specified type.

Supported types include:

Feature Types: SequenceFeature[T]
Built-in types: list[T], list[list[T]]
PyArrow scalar types: pyarrow.ListScalar
PyArrow array types: pyarrow.ListArray

Type:: Sequence

alias of SequenceFeature[T] | list[T] | list[list[T]] | ListScalar | ListArray

hyped.core.typing.String¶

Type alias for a string.

Supported types include:

Feature Types: StringFeature
Built-in types: str, list[str]
PyArrow scalar types: pyarrow.StringScalar
PyArrow array types: pyarrow.StringArray

Type:: String

alias of StringFeature | str | list[str] | StringScalar | StringArray

hyped.core.typing.TraceIndexList¶

A list of trace indices used to map outputs to their source samples in augmentation processes.

In data augmentation, a single input sample can generate multiple output samples. The TraceIndexList tracks the origin of each output sample by maintaining a list of indices. Each index in this list corresponds to the position of the input sample in the original batch that was used to generate the output sample.

For example, if trace_index[i] = j, it indicates that the i-th output sample was derived from the j-th input sample in the original batch.

Usage Context:

When a batch of input samples undergoes augmentation, this list provides a direct mapping from each output sample back to its corresponding input sample.
This type is commonly returned alongside the augmented batch, enabling users to track which input sample produced which output sample.

alias of list[int]

hyped.core.typing.UInt¶

Type alias for an unsigned integer of varying bit length.

Supported types include:

Feature Types: UInt8Feature, UInt16Feature, UInt32Feature, UInt64Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.UInt8Scalar, pyarrow.UInt16Scalar, pyarrow.UInt32Scalar, pyarrow.UInt64Scalar
PyArrow array types: pyarrow.UInt8Array, pyarrow.UInt16Array, pyarrow.UInt32Array, pyarrow.UInt64Array

Type:: UInt

hyped.core.typing.UInt16¶

Type alias for a 16-bit unsigned integer.

Supported types include:

Feature Types: UInt16Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.UInt16Scalar
PyArrow array types: pyarrow.UInt16Array

Type:: UInt16

alias of UInt16Feature | int | list[int] | UInt16Scalar | UInt16Array

hyped.core.typing.UInt32¶

Type alias for a 32-bit unsigned integer.

Supported types include:

Feature Types: UInt32Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.UInt32Scalar
PyArrow array types: pyarrow.UInt32Array

Type:: UInt32

alias of UInt32Feature | int | list[int] | UInt32Scalar | UInt32Array

hyped.core.typing.UInt64¶

Type alias for a 64-bit unsigned integer.

Supported types include:

Feature Types: UInt64Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.UInt64Scalar
PyArrow array types: pyarrow.UInt64Array

Type:: UInt64

alias of UInt64Feature | int | list[int] | UInt64Scalar | UInt64Array

hyped.core.typing.UInt8¶

Type alias for an 8-bit unsigned integer.

Supported types include:

Feature Types: UInt8Feature
Built-in types: int, list[int]
PyArrow scalar types: pyarrow.UInt8Scalar
PyArrow array types: pyarrow.UInt8Array

Type:: UInt8

alias of UInt8Feature | int | list[int] | UInt8Scalar | UInt8Array