hyped.core.features.validators module¶

Feature Validators and Resolvers.

This module defines custom validators and resolvers for validating and resolving features using Pydantic’s validation framework. These classes extend Pydantic’s AfterValidator and BeforeValidator to enforce validation and type resolution logic.

class hyped.core.features.validators.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.features.validators.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]¶

class hyped.core.features.validators.Len[source]¶

class hyped.core.features.validators.Len(length: int)

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

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

class hyped.core.features.validators.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]¶

class hyped.core.features.validators.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]¶