diff --git a/docs/source/pytorch_guide/index.md b/docs/source/pytorch_guide/index.md index d16c1e093..ba3036694 100644 --- a/docs/source/pytorch_guide/index.md +++ b/docs/source/pytorch_guide/index.md @@ -44,6 +44,30 @@ td = TensorDict( schema.validate(td) ``` +## Type Coercion + +Set `coerce=True` to automatically convert tensor dtypes: + +```{code-cell} python +schema_coerce = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), + coerce=True, +) + +# Input with wrong dtype (float64) +td_wrong_dtype = TensorDict( + {"observation": torch.randn(32, 10).to(torch.float64)}, + batch_size=[32], +) + +# Dtype is automatically coerced to float32 +validated = schema_coerce.validate(td_wrong_dtype) +assert validated["observation"].dtype == torch.float32 +``` + ## Define a schema with a class-based model ```{code-cell} python @@ -78,12 +102,18 @@ Use {func}`~pandera.tensordict.Field` to define additional constraints like shap tensordict_schema tensordict_model tensordict_checks +tensordict_schema_inference +tensordict_io +tensordict_strategies error_reporting ``` - {ref}`pytorch-tensordict-schema` — validating a {class}`~tensordict.TensorDict` with `Tensor` components - {ref}`pytorch-tensordict-model` — class-based `TensorDictModel` - {ref}`pytorch-checks` — checks, parsers, and lazy validation +- {ref}`pytorch-tensordict-inference` — infer schemas from data automatically +- {ref}`pytorch-tensordict-io` — save/load schemas with YAML/JSON +- {ref}`pytorch-tensordict-strategies` — generate synthetic data with Hypothesis - {ref}`pytorch-error-reporting` — `SchemaError` / `SchemaErrors`, lazy validation, and failure cases ## See also diff --git a/docs/source/pytorch_guide/tensordict_io.md b/docs/source/pytorch_guide/tensordict_io.md new file mode 100644 index 000000000..83655f786 --- /dev/null +++ b/docs/source/pytorch_guide/tensordict_io.md @@ -0,0 +1,227 @@ +--- +file_format: mystnb +--- + +(pytorch-tensordict-io)= + +# Serialization and Deserialization + +Save and load TensorDict schemas using YAML or JSON, or save TensorDicts with embedded schema metadata. + +## Schema Serialization + +### YAML Format + +```{code-cell} python +import torch +from pandera import Check +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + "reward": pa.Tensor(dtype=torch.float32, shape=(None,), checks=Check.greater_than(0.0)), + }, + batch_size=(32,), +) + +# Save schema to YAML +schema_yaml = pa.to_yaml(schema) +print("YAML output preview:", schema_yaml[:150] + "...") + +# Load schema from YAML +loaded_schema = pa.from_yaml(schema_yaml) +assert loaded_schema.batch_size == schema.batch_size +``` + +### JSON Format + +```{code-cell} python +import json +import torch +from pandera import Check +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +# Serialize to JSON string +json_str = pa.to_json(schema) +print("JSON output preview:", json_str[:150] + "...") + +# Deserialize from JSON string +loaded_schema = pa.from_json(json_str) +``` + +## File I/O + +Save schemas to and load from files: + +```{code-cell} python +import tempfile +from pathlib import Path +import torch +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +# Save to file +with tempfile.TemporaryDirectory() as tmpdir: + schema_path = Path(tmpdir) / "schema.yaml" + pa.to_yaml(schema, schema_path) +<<<<<<< HEAD + +======= + +>>>>>>> pr/pytorch-tensordict-phase3-4 + # Load from file + loaded_schema = pa.from_yaml(schema_path) + print("Successfully saved and loaded schema") +``` + +## TensorDict Saving/Loading + +Save TensorDicts with embedded schema metadata for data integrity: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(32,), +) + +td = TensorDict({ + "observation": torch.randn(32, 10), + "action": torch.randint(0, 5, (32,)), +}, batch_size=[32]) + +# Save with schema metadata +import tempfile +with tempfile.TemporaryDirectory() as tmpdir: + save_path = f"{tmpdir}/rl_batch.pt" + pa.save(schema, td, save_path) +<<<<<<< HEAD + +======= + +>>>>>>> pr/pytorch-tensordict-phase3-4 + # Load and validate automatically + loaded_td = pa.load(save_path) + print(f"Loaded batch size: {loaded_td.batch_size}") +``` + +## Use Cases + +### 1. Data Pipeline Validation + +Save trained model inputs/outputs with schema: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa +import tempfile + +# At training time +training_data = TensorDict({ + "input": torch.randn(100, 64), + "target": torch.randint(0, 10, (100,)), +}, batch_size=[100]) + +schema = pa.TensorDictSchema( + keys={ + "input": pa.Tensor(dtype=torch.float32, shape=(None, 64)), + "target": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(100,), +) + +with tempfile.TemporaryDirectory() as tmpdir: + pa.save(schema, training_data, f"{tmpdir}/training.pt") +<<<<<<< HEAD + +======= + +>>>>>>> pr/pytorch-tensordict-phase3-4 + # Later: validate before inference + loaded = pa.load(f"{tmpdir}/training.pt") + print("Successfully validated and loaded training data") +``` + +### 2. Configuration Management + +Define schemas in YAML for version control and collaboration: + +```{code-cell} python +import tempfile +from pathlib import Path +import pandera.tensordict as pa +import torch + +# Define schema programmatically +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(32,), +) + +with tempfile.TemporaryDirectory() as tmpdir: + # Save to config file + config_path = Path(tmpdir) / "rl_schema.yaml" + pa.to_yaml(schema, config_path) +<<<<<<< HEAD + +======= + +>>>>>>> pr/pytorch-tensordict-phase3-4 + # Load from config file (e.g., in a different process) + loaded_schema = pa.from_yaml(config_path) + print("Successfully loaded schema from config") +``` + +### 3. Distributed Training + +Share schemas across workers via serialization: + +```{code-cell} python +import torch +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +# Worker 1: Serialize and send +serialized = pa.to_json(schema) +print("Serialized schema (JSON):", serialized[:150] + "...") + +# Worker 2: Deserialize and use (simulated here) +loaded_schema = pa.from_json(serialized) +print(f"Loaded batch_size: {loaded_schema.batch_size}") +``` + +## See also + +- {ref}`pytorch-tensordict-inference` — infer schemas from data +- {ref}`pytorch-tensordict-schema` — manual schema definition diff --git a/docs/source/pytorch_guide/tensordict_model.md b/docs/source/pytorch_guide/tensordict_model.md index 4c0c93234..035ecedeb 100644 --- a/docs/source/pytorch_guide/tensordict_model.md +++ b/docs/source/pytorch_guide/tensordict_model.md @@ -90,14 +90,45 @@ class RLWithBatchSize(pa.TensorDictModel): Use `lazy=True` to collect all validation errors: ```{code-cell} python +from tensordict import TensorDict + +# Create invalid data with wrong dtypes +td_wrong_dtype = TensorDict( + {"observation": torch.randn(32, 10).to(torch.float64), "action": torch.randint(0, 4, (32,)), "reward": torch.randn(32)}, + batch_size=[32], +) + try: - RL.validate(td, lazy=True) + RL.validate(td_wrong_dtype, lazy=True) except pa.SchemaErrors as e: print(f"Found {len(e.schema_errors)} validation errors:") for err in e.schema_errors: print(f" - {err.reason_code}") ``` +## Type Coercion + +Model schemas support dtype coercion with the `coerce=True` option: + +```{code-cell} python +class RLWithCoercion(pa.TensorDictModel): + observation: torch.float32 = pa.Field(shape=(None, 10)) + + class Config: + batch_size = (32,) + coerce = True + +# Input with wrong dtype +td = TensorDict( + {"observation": torch.randn(32, 10).to(torch.float64)}, + batch_size=[32], +) + +# Dtype is automatically coerced during validation +validated = RLWithCoercion.validate(td) +assert validated["observation"].dtype == torch.float32 +``` + ## Model inheritance Models can be inherited to create more specific schemas: diff --git a/docs/source/pytorch_guide/tensordict_schema.md b/docs/source/pytorch_guide/tensordict_schema.md index 56eaf8b5d..38703cfcd 100644 --- a/docs/source/pytorch_guide/tensordict_schema.md +++ b/docs/source/pytorch_guide/tensordict_schema.md @@ -87,8 +87,91 @@ schema = pa.TensorDictSchema( ) ``` +## Type Coercion + +Set `coerce=True` to automatically convert tensor dtypes during validation: + +```{code-cell} python +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64), + }, + batch_size=(32,), + coerce=True, +) + +# Input has wrong dtypes (float64, int32) +td = TensorDict( + { + "observation": torch.randn(32, 10).to(torch.float64), + "action": torch.randint(0, 5, (32,)).to(torch.int32), + }, + batch_size=[32], +) + +# Dtypes are automatically coerced to float32 and int64 +validated = schema.validate(td) +assert validated["observation"].dtype == torch.float32 +assert validated["action"].dtype == torch.int64 +``` + +Type coercion is applied **before** validation checks, so any dtype or shape +constraints will be evaluated on the coerced data. + +## Schema Inference + +Automatically infer a schema from existing data using `pa.infer_schema()`: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Existing TensorDict with sample data +sample_td = TensorDict({ + "observation": torch.randn(100, 64), + "action": torch.randint(0, 4, (100,)), +}, batch_size=[100]) + +# Infer schema from data +inferred_schema = pa.infer_schema(sample_td) +``` + +See {ref}`pytorch-tensordict-inference` for more details on schema inference. + +## Serialization + +Save and load schemas using YAML or JSON: + +```{code-cell} python +import tempfile +from pathlib import Path +import torch +import pandera.tensordict as pa + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +with tempfile.TemporaryDirectory() as tmpdir: + # Save to YAML + yaml_path = Path(tmpdir) / "schema.yaml" + pa.to_yaml(schema, yaml_path) + + # Load from YAML + loaded_schema = pa.from_yaml(yaml_path) +``` + +See {ref}`pytorch-tensordict-io` for more details on serialization. + ## See also - {ref}`pytorch-tensordict-model` — class-based schema definition - {ref}`pytorch-checks` — checks for value validation - {ref}`pytorch-error-reporting` — error handling +- {ref}`pytorch-tensordict-inference` — infer schemas from data +- {ref}`pytorch-tensordict-io` — save/load schemas diff --git a/docs/source/pytorch_guide/tensordict_schema_inference.md b/docs/source/pytorch_guide/tensordict_schema_inference.md new file mode 100644 index 000000000..e978675a6 --- /dev/null +++ b/docs/source/pytorch_guide/tensordict_schema_inference.md @@ -0,0 +1,135 @@ +--- +file_format: mystnb +--- + +(pytorch-tensordict-inference)= + +# Schema Inference + +Use `pa.infer_schema()` to automatically infer a schema from existing TensorDict data. + +```{note} +Schema inference is useful when you have existing data and want to create a validation schema without manually specifying all the dtypes, shapes, and value ranges. +``` + +## Basic Usage + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Existing data with batch_size (100,) +td = TensorDict({ + "observation": torch.randn(100, 64), + "action": torch.randint(0, 4, (100,)), + "reward": torch.randn(100), +}, batch_size=[100]) + +# Infer schema from data +schema = pa.infer_schema(td) +print(f"Keys: {list(schema.keys.keys())}") +print(f"Batch size: {schema.batch_size}") + +# Output includes inferred dtypes and shapes: +# TensorDictSchema(keys={'observation', 'action', 'reward'}, batch_size=(100,)) +``` + +## What Gets Inferred + +- **Dtypes**: Automatically detected (e.g., `torch.float32`, `torch.int64`) +- **Shapes**: Full tensor shapes including batch dimensions +- **Value ranges**: Min/max bounds for numeric tensors (used in range checks) + +## TensorClass Support + +Works with both `TensorDict` and `tensorclass`: + +```{code-cell} python +from tensordict import tensorclass + +@tensorclass +class RLData: + observation: torch.Tensor + action: torch.Tensor + reward: torch.Tensor + +tc = RLData( + observation=torch.randn(100, 32), + action=torch.randint(0, 4, (100,)), + reward=torch.randn(100), + batch_size=[100] +) + +# Infer schema from tensorclass +schema = pa.infer_schema(tc) +print(f"Keys: {list(schema.keys.keys())}") +``` + +## Use Cases + +### 1. Quick Schema Development + +Start with sample data and infer the schema, then refine it: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa +from pandera import Check + +# Step 1: Infer from sample data (batch_size=32) +sample_td = TensorDict({ + "observation": torch.randn(32, 10), + "action": torch.randint(0, 4, (32,)), +}, batch_size=[32]) +schema = pa.infer_schema(sample_td) + +# Step 2: Refine the inferred schema for your needs +schema.keys["observation"].checks.append(Check.greater_than_or_equal_to(-1.0)) +``` + +### 2. Data Quality Validation + +Infer a baseline schema from healthy data, then validate new data against it: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Infer from known-good dataset (batch_size=100) +healthy_data = TensorDict({ + "observation": torch.randn(100, 64), +}, batch_size=[100]) +baseline_schema = pa.infer_schema(healthy_data) + +# Validate incoming batches with matching batch size +new_batch = TensorDict({"observation": torch.randn(100, 64)}, batch_size=[100]) +validated_batch = baseline_schema.validate(new_batch) +``` + +### 3. Schema Evolution + +Track how schemas change over time by inferring and comparing: + +```{code-cell} python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Infer from different dataset versions (both with batch_size=10) +data_v1 = TensorDict({"x": torch.randn(10)}, batch_size=[10]) +schema_v1 = pa.infer_schema(data_v1) + +data_v2 = TensorDict({"x": torch.randn(10), "y": torch.randn(10)}, batch_size=[10]) +schema_v2 = pa.infer_schema(data_v2) + +# Compare key sets +assert set(schema_v2.keys.keys()) == {"x", "y"} +``` + +## See also + +- {ref}`pytorch-tensordict-schema` — manual schema definition +- {ref}`pytorch-tensordict-io` — saving and loading schemas diff --git a/docs/source/pytorch_guide/tensordict_strategies.md b/docs/source/pytorch_guide/tensordict_strategies.md new file mode 100644 index 000000000..5929bca50 --- /dev/null +++ b/docs/source/pytorch_guide/tensordict_strategies.md @@ -0,0 +1,280 @@ +--- +file_format: mystnb +--- + +(pytorch-tensordict-strategies)= + +# Hypothesis Strategies + +Generate synthetic TensorDict data for property-based testing using +[Hypothesis](https://hypothesis.readthedocs.io/). + +## Installation + +```bash +pip install pandera[strategies] +``` + +## Basic Usage + +Generate valid TensorDicts from a schema: + +```{code-cell} python +from hypothesis import given, settings +import torch +from tensordict import TensorDict +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategy + +# Define schema +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(32,), +) + +# Generate and validate TensorDicts +@given(tensordict_strategy(schema)) +@settings(max_examples=10) +def test_policy_output(td): + """Property-based test for policy network.""" + # Schema validates automatically + assert td["observation"].shape == (32, 10) + assert td["action"].dtype == torch.int64 + +# Run the test +test_policy_output() +``` + +## TensorClass Generation + +Generate tensorclass instances: + +```{code-cell} python +from tensordict import tensorclass +from pandera.strategies import tensorclass_strategy + +@tensorclass +class RLData: + observation: torch.Tensor + action: torch.Tensor + reward: torch.Tensor + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 64)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + "reward": pa.Tensor(dtype=torch.float32, shape=(None,)), + }, + batch_size=(128,), +) + +@given(tensorclass_strategy(RLData, schema)) +@settings(max_examples=10) +def test_tensorclass(tc): + assert tc.batch_size == [128] + assert hasattr(tc, "observation") + +test_tensorclass() +``` + +## Custom Data Generation + +Combine with Hypothesis strategies for more control: + +```{code-cell} python +from hypothesis import given, settings +import torch +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategy +from pandera import Check + +# Generate only valid actions (0-3) +schema = pa.TensorDictSchema( + keys={ + "action": pa.Tensor(dtype=torch.int64, shape=(None,), checks=Check.isin([0, 1, 2, 3])), + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +@given(tensordict_strategy(schema)) +@settings(max_examples=10) +def test_action_space(td): + """Test that actions are always in valid range.""" + assert td["action"].max() < 4 + assert td["action"].min() >= 0 + +test_action_space() +``` + +## Integration with PyTorch Testing Patterns + +```{code-cell} python +import torch +from hypothesis import given, settings +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategy + +class PolicyNetwork(torch.nn.Module): + def __init__(self): + super().__init__() + self.fc = torch.nn.Linear(10, 5) + + def forward(self, obs: torch.Tensor) -> torch.Tensor: + return self.fc(obs) + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +@given(tensordict_strategy(schema)) +@settings(max_examples=10) +def test_policy_model(td): + model = PolicyNetwork() + + # Test with various inputs + output = model(td["observation"]) + assert output.shape == (32, 5) + assert torch.isfinite(output).all() + +test_policy_model() +``` + +## Use Cases + +### 1. Unit Testing Models + +Generate diverse inputs for model testing: + +```{code-cell} python +import torch +from hypothesis import given, settings +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategy + +class PolicyNetwork(torch.nn.Module): + def __init__(self): + super().__init__() + self.fc = torch.nn.Linear(10, 5) + + def forward(self, obs: torch.Tensor) -> torch.Tensor: + return self.fc(obs) + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +@given(tensordict_strategy(schema)) +@settings(max_examples=50) +def test_policy_model(td): + model = PolicyNetwork() + + # Test with various inputs + output = model(td["observation"]) + assert output.shape == (32, 5) + assert torch.isfinite(output).all() + +test_policy_model() +``` + +### 2. Property-Based Testing of RL Algorithms + +Test reinforcement learning algorithms with generated rollouts: + +```{code-cell} python +from tensordict import tensorclass +from hypothesis import given, settings +import pandera.tensordict as pa +from pandera.strategies import tensorclass_strategy + +@tensorclass +class Rollout: + observations: torch.Tensor + actions: torch.Tensor + rewards: torch.Tensor + dones: torch.Tensor + +rollout_schema = pa.TensorDictSchema( + keys={ + "observations": pa.Tensor(dtype=torch.float32, shape=(None, 17)), + "actions": pa.Tensor(dtype=torch.float32, shape=(None, 6)), + "rewards": pa.Tensor(dtype=torch.float32, shape=(None,)), + "dones": pa.Tensor(dtype=torch.bool, shape=(None,)), + }, + batch_size=(256,), +) + +class ReplayBuffer: + def __init__(self): + self.data = [] + + def add(self, rollout: Rollout): + self.data.append(rollout) + + def sample(self) -> Rollout: + import random + return random.choice(self.data) if self.data else None + +@given(tensorclass_strategy(Rollout, rollout_schema)) +@settings(max_examples=10) +def test_replay_buffer(rollout): + """Test replay buffer with generated rollouts.""" + replay_buffer = ReplayBuffer() + replay_buffer.add(rollout) + sampled = replay_buffer.sample() + + # Properties + assert sampled is not None + assert "observations" in sampled.keys() + +test_replay_buffer() +``` + +### 3. Stress Testing Data Preprocessing + +Test preprocessing pipelines with edge cases: + +```{code-cell} python +import torch +from hypothesis import given, settings +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategy + +def preprocess(td: TensorDict) -> TensorDict: + """Normalize observations.""" + td["observation"] = (td["observation"] - td["observation"].mean()) / td["observation"].std() + return td + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(32,), +) + +@given(tensordict_strategy(schema)) +@settings(max_examples=50) +def test_preprocessing(td): + """Test preprocessing handles various inputs.""" + processed = preprocess(td) + + # Check normalization + assert torch.allclose(processed["observation"].mean(), torch.tensor(0.0), atol=1e-4) + +test_preprocessing() +``` + +## See also + +- {ref}`pytorch-tensordict-schema` — define validation schemas +- {ref}`pytorch-checks` — value constraints +- [Hypothesis documentation](https://hypothesis.readthedocs.io/) diff --git a/pandera/api/tensordict/__init__.py b/pandera/api/tensordict/__init__.py index f9332737b..4df959008 100644 --- a/pandera/api/tensordict/__init__.py +++ b/pandera/api/tensordict/__init__.py @@ -2,21 +2,12 @@ from __future__ import annotations -# mypy: disable-error-code=attr-defined -from typing import TYPE_CHECKING - from pandera import errors -if TYPE_CHECKING: - from pandera.api.tensordict.components import Tensor - from pandera.api.tensordict.container import TensorDictSchema - from pandera.api.tensordict.model import TensorDictModel - -# Import actual implementations (not just for type checking) -from pandera.api.tensordict.components import Tensor -from pandera.api.tensordict.container import TensorDictSchema -from pandera.api.tensordict.model import TensorDictModel -from pandera.api.tensordict.model_components import Field +from .components import Tensor +from .container import TensorDictSchema +from .model import TensorDictModel +from .model_components import Field __all__ = [ "Tensor", diff --git a/pandera/backends/tensordict/base.py b/pandera/backends/tensordict/base.py index 6c37ef3e9..e122753c2 100644 --- a/pandera/backends/tensordict/base.py +++ b/pandera/backends/tensordict/base.py @@ -80,6 +80,9 @@ def validate( check_obj = self.preprocess(check_obj, inplace=inplace) + if schema.coerce: + check_obj = self.coerce_dtype(check_obj, schema=schema) + error_handler = self.run_checks_and_handle_errors( error_handler, schema, check_obj, lazy ) @@ -95,6 +98,70 @@ def validate( def coerce_dtype(self, check_obj, schema=None): """Coerce the dtype of tensors in the TensorDict.""" + if not schema.coerce: + return check_obj + + from pandera import errors + from pandera.api.base.error_handler import ErrorHandler + + error_handler = ErrorHandler(lazy=True) + + def _set_tensor(obj, key, value): + """Set tensor in TensorDict or tensorclass object.""" + try: + # For TensorDict, use item assignment + from tensordict import TensorDict + + if isinstance(obj, TensorDict): + obj[key] = value + else: + # For tensorclass, use attribute access + setattr(obj, key, value) + except (KeyError, TypeError): + pass + + for key, tensor_schema in schema.keys.items(): + # Apply coercion if either schema-level coerce is True or + # tensor-level coerce is True, and dtype is specified + if ( + not (schema.coerce or tensor_schema.coerce) + or tensor_schema.dtype is None + ): + continue + + try: + tensor = _get_tensor(check_obj, key) + coerced_tensor = tensor_schema.dtype.try_coerce(tensor) + _set_tensor(check_obj, key, coerced_tensor) + except errors.SchemaError as err: + error_handler.collect_error( + get_error_category(err.reason_code), + err.reason_code, + err, + ) + except Exception as exc: + from pandera.errors import SchemaError, SchemaErrors + + # Wrap non-SchemaError exceptions in SchemaError + error = SchemaError( + schema=schema, + data=check_obj, + message=f"Coercion failed for key '{key}': {exc}", + reason_code=SchemaErrorReason.DATATYPE_COERCION, + ) + error_handler.collect_error( + get_error_category(error.reason_code), + error.reason_code, + error, + ) + + if error_handler.collected_errors: + raise errors.SchemaErrors( + schema=schema, + schema_errors=error_handler.schema_errors, + data=check_obj, + ) + return check_obj def run_checks_and_handle_errors( diff --git a/pandera/io/__init__.py b/pandera/io/__init__.py index 805823896..33ddc0a74 100644 --- a/pandera/io/__init__.py +++ b/pandera/io/__init__.py @@ -1,10 +1,27 @@ -"""Schema serialization (import backend modules explicitly). +"""Input/Output operations for pandera schemas.""" -Examples: +from __future__ import annotations -- ``import pandera.io.pandas_io`` — pandas / modin / dask / pyspark.pandas API schemas -- ``import pandera.io.polars_io`` — Polars API schemas -- ``import pandera.io.pyspark_sql_io`` — PySpark SQL API schemas -- ``import pandera.io.ibis_io`` — Ibis API schemas -- ``import pandera.io.xarray_io`` — xarray API schemas -""" +from .tensordict_io import ( + from_json, + from_yaml, + load, + save, + to_json, + to_yaml, +) + +__all__ = [ + "ibis_io", + "pandas_io", + "polars_io", + "pyspark_sql_io", + "xarray_io", + "tensordict_io", + "from_json", + "from_yaml", + "to_json", + "to_yaml", + "save", + "load", +] diff --git a/pandera/io/tensordict_io.py b/pandera/io/tensordict_io.py new file mode 100644 index 000000000..3841fa4fd --- /dev/null +++ b/pandera/io/tensordict_io.py @@ -0,0 +1,343 @@ +"""Serialize and deserialize TensorDict schemas.""" + +from __future__ import annotations + +import enum +import json +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from pandera import dtypes +from pandera.api.checks import Check +from pandera.api.tensordict.components import Tensor +from pandera.api.tensordict.container import TensorDictSchema +from pandera.engines import tensordict_engine +from pandera.errors import SchemaDefinitionError +from pandera.io._check_io import checks_dict_to_list + + +def _serialize_check_stats(check_stats, dtype=None): + """Serialize check statistics into JSON/YAML-compatible format.""" + + def handle_stat_dtype(stat): + if isinstance(stat, type) and issubclass(stat, enum.Enum): + return [e.value for e in stat] + return stat + + check_options = ( + check_stats.pop("options", {}) if isinstance(check_stats, dict) else {} + ) + + if isinstance(check_stats, dict) and len(check_stats) == 1: + value = handle_stat_dtype(list(check_stats.values())[0]) + if check_options: + return {"value": value, "options": check_options} + return value + + if isinstance(check_stats, dict): + serialized_check_stats = {} + for arg, stat in check_stats.items(): + serialized_check_stats[arg] = handle_stat_dtype(stat) + if check_options: + serialized_check_stats["options"] = check_options + return serialized_check_stats + + return handle_stat_dtype(check_stats) + + +def _serialize_tensor_stats(tensor_stats): + """Serialize Tensor component stats into JSON/YAML-compatible format.""" + serialized_checks = None + if tensor_stats.get("checks"): + serialized_checks = [] + for check_stats in tensor_stats["checks"]: + serialized_check_stats = _serialize_check_stats(check_stats) + serialized_checks.append(serialized_check_stats) + + dtype = tensor_stats.get("dtype") + if dtype: + try: + dtype = str(dtype) + except Exception: + dtype = str(dtype) + + return { + "dtype": dtype, + "shape": list(tensor_stats["shape"]), + "nullable": tensor_stats.get("nullable", False), + "checks": serialized_checks, + **{ + key: tensor_stats.get(key) + for key in ["description", "title"] + if key in tensor_stats + }, + } + + +def serialize_schema( + dataframe_schema, *, minimal: bool = True +) -> dict[str, Any]: + """Serialize a TensorDict schema into JSON/YAML-compatible dict.""" + from pandera import __version__ + + statistics: dict[str, Any] = { + "keys": {}, + "batch_size": ( + list(dataframe_schema.batch_size) + if dataframe_schema.batch_size + else None + ), + } + + if dataframe_schema.keys: + for key_name, tensor_component in dataframe_schema.keys.items(): + statistics["keys"][key_name] = { + "dtype": str(tensor_component.dtype), + "shape": list(tensor_component.shape) + if tensor_component.shape + else None, + "nullable": getattr(tensor_component, "nullable", False), + "checks": ( + [ + { + "options": {"check_name": check.name}, + **(check.statistics or {}), + } + for check in tensor_component.checks + ] + if hasattr(tensor_component, "checks") + and tensor_component.checks + else None + ), + } + + out = { + "schema_type": "tensordict", + "version": __version__, + "keys": statistics["keys"], + "batch_size": statistics["batch_size"], + "coerce": dataframe_schema.coerce, + "dtype": ( + str(dataframe_schema.dtype) if dataframe_schema.dtype else None + ), + } + + return out + + +def _deserialize_check_stats(check, serialized_check_stats): + """Deserialize check statistics and reconstruct Check with options.""" + options = {} + if isinstance(serialized_check_stats, dict): + options = serialized_check_stats.pop("options", {}) + + if ( + "value" in serialized_check_stats + and len(serialized_check_stats) == 1 + ): + serialized_check_stats = serialized_check_stats["value"] + + check_instance = check(**serialized_check_stats) + + if options: + for option_name, option_value in options.items(): + if option_name != "check_name": + setattr(check_instance, option_name, option_value) + + return check_instance + + +def _deserialize_tensor_stats(serialized_tensor_stats): + """Deserialize Tensor component stats and reconstruct Tensor.""" + serialized_tensor_stats = dict(serialized_tensor_stats) + + dtype = serialized_tensor_stats.get("dtype") + if dtype: + try: + dtype = tensordict_engine.Engine.dtype(dtype) + except (TypeError, ValueError): + dtype = tensordict_engine.Engine.dtype(str(dtype).lower()) + + shape = serialized_tensor_stats.get("shape") + if shape is not None: + shape = tuple(shape) + + checks = checks_dict_to_list(serialized_tensor_stats.get("checks")) + if checks is not None: + checks = [ + _deserialize_check_stats( + getattr(Check, check["options"]["check_name"]), check + ) + for check in checks + ] + + return { + "dtype": dtype, + "shape": shape, + "nullable": serialized_tensor_stats.get("nullable", False), + "checks": checks, + **{ + key: serialized_tensor_stats.get(key) + for key in ["description", "title"] + if key in serialized_tensor_stats + }, + } + + +def deserialize_schema(serialized_schema): + """Deserialize a mapping into TensorDictSchema.""" + serialized_schema = serialized_schema if serialized_schema else {} + + if not isinstance(serialized_schema, Mapping): + raise SchemaDefinitionError("Schema representation must be a mapping.") + + st = serialized_schema.get("schema_type") + if st is not None and st != "tensordict": + raise SchemaDefinitionError( + f"Expected schema_type 'tensordict', got {st!r}" + ) + + keys = serialized_schema.get("keys") + if keys is not None: + keys = { + key_name: Tensor(**_deserialize_tensor_stats(tensor_stats)) + for key_name, tensor_stats in keys.items() + } + + batch_size = serialized_schema.get("batch_size") + if batch_size is not None: + batch_size = tuple(batch_size) + + return TensorDictSchema( + keys=keys, + batch_size=batch_size, + coerce=serialized_schema.get("coerce", False), + dtype=serialized_schema.get("dtype"), + ) + + +def from_yaml(yaml_schema): + """Load a TensorDictSchema from YAML.""" + try: + import yaml + except ImportError as exc: # pragma: no cover + raise ImportError( + "PyYAML is required for YAML support. " + "Install with: pip install pyyaml" + ) from exc + + try: + with Path(yaml_schema).open("r", encoding="utf-8") as f: + serialized_schema = yaml.safe_load(f) + except (TypeError, OSError): + serialized_schema = yaml.safe_load(yaml_schema) + + return deserialize_schema(serialized_schema) + + +def to_yaml(dataframe_schema, stream=None, *, minimal: bool = True): + """Write a TensorDictSchema to YAML.""" + try: + import yaml + except ImportError as exc: # pragma: no cover + raise ImportError( + "PyYAML is required for YAML support. " + "Install with: pip install pyyaml" + ) from exc + + statistics = serialize_schema(dataframe_schema, minimal=minimal) + + def _write_yaml(obj, stream): + return yaml.safe_dump(obj, stream=stream, sort_keys=False) + + try: + with Path(stream).open("w", encoding="utf-8") as f: + _write_yaml(statistics, f) + except (TypeError, OSError): + return _write_yaml(statistics, stream) + + +def from_json(source): + """Load a TensorDictSchema from JSON.""" + if isinstance(source, str): + try: + serialized_schema = json.loads(source) + except json.decoder.JSONDecodeError: + with Path(source).open(encoding="utf-8") as f: + serialized_schema = json.load(fp=f) + elif isinstance(source, Path): + with source.open(encoding="utf-8") as f: + serialized_schema = json.load(fp=f) + else: + serialized_schema = json.load(fp=source) + + return deserialize_schema(serialized_schema) + + +def to_json(dataframe_schema, target=None, *, minimal: bool = True, **kwargs): + """Write a TensorDictSchema to JSON.""" + serialized_schema = serialize_schema(dataframe_schema, minimal=minimal) + + if target is None: + return json.dumps(serialized_schema, sort_keys=False, **kwargs) + + if isinstance(target, (str, Path)): + with Path(target).open("w", encoding="utf-8") as f: + json.dump(serialized_schema, fp=f, sort_keys=False, **kwargs) + else: + json.dump(serialized_schema, fp=target, sort_keys=False, **kwargs) + + +def save(dataframe_schema, tensor_dict, path): + """Save TensorDict to disk with schema metadata embedded.""" + import torch + + path = Path(path) + + if isinstance(tensor_dict, dict): + td = tensordict_engine.Engine.dtype("tensordict")(**tensor_dict) + else: + td = tensor_dict + + path.parent.mkdir(parents=True, exist_ok=True) + + torch.save( + { + "tensordict": td, + "schema": serialize_schema(dataframe_schema), + }, + path, + ) + + +def load(path): + """Load TensorDict from disk and validate with schema.""" + import torch + + data = torch.load(path, weights_only=False) + + if isinstance(data, dict) and "tensordict" in data: + td = data["tensordict"] + saved_schema = data.get("schema") + + if saved_schema: + loaded_schema = deserialize_schema(saved_schema) + return loaded_schema.validate(td) + + if isinstance(data, dict): + try: + from tensordict import TensorDict + + if any(hasattr(v, "dtype") for v in data.values()): + td = TensorDict(data) + + try: + schema = infer_schema(td) + return schema.validate(td) + except Exception: + return td + except ImportError: + pass + + return data diff --git a/pandera/schema_inference/__init__.py b/pandera/schema_inference/__init__.py index e69de29bb..8fa957f86 100644 --- a/pandera/schema_inference/__init__.py +++ b/pandera/schema_inference/__init__.py @@ -0,0 +1,10 @@ +"""Schema inference for different data types.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .tensordict import infer_schema as tensordict_infer_schema + +__all__ = ["infer_schema"] diff --git a/pandera/schema_inference/tensordict.py b/pandera/schema_inference/tensordict.py new file mode 100644 index 000000000..a4ca90509 --- /dev/null +++ b/pandera/schema_inference/tensordict.py @@ -0,0 +1,166 @@ +"""Module for inferring TensorDict schema from data.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, overload + +import torch + +from pandera.api.tensordict.components import Tensor +from pandera.api.tensordict.container import TensorDictSchema + + +@overload +def infer_schema( + tensordict: torch.TensorDict, +) -> TensorDictSchema: ... +@overload +def infer_schema( + tensordict: Any, +) -> TensorDictSchema: ... + + +def infer_schema(tensordict: Any): + """Infer schema for a TensorDict or tensorclass object. + + Automatically detects dtypes, shapes, and value statistics from data. + + :param tensordict: TensorDict or tensorclass object to infer. + :returns: TensorDictSchema with inferred properties. + :raises TypeError: if tensordict is not a valid TensorDict/tensorclass. + + Example: + >>> import torch + >>> from tensordict import TensorDict + >>> import pandera.tensordict as pa + >>> td = TensorDict({ + ... "obs": torch.randn(32, 10), + ... "action": torch.randint(0, 4, (32,)) + ... }, batch_size=[32]) + >>> schema = pa.infer_schema(td) + """ + try: + from tensordict import TensorDict as TD + except ImportError as exc: + raise ImportError( + "tensordict is required for infer_schema but is not installed. " + "Install with: pip install tensordict" + ) from exc + + if not _is_tensordict_instance(tensordict): + raise TypeError( + f"Expected TensorDict or tensorclass, got {type(tensordict)}" + ) + + keys = {} + for key in _get_keys_from_tensordict(tensordict): + tensor = tensordict.get(key) + if tensor is not None: + keys[key] = _infer_tensor_schema(tensor) + + # Convert batch_size from torch.Size to tuple + batch_size_tuple = ( + tuple(tensordict.batch_size) if tensordict.batch_size else None + ) + + return TensorDictSchema( + keys=keys, + batch_size=batch_size_tuple, + coerce=True, + ) + + +def _get_keys_from_tensordict(obj): + """Get keys from a TensorDict or tensorclass object.""" + try: + # For TensorDict, use .keys() + if hasattr(obj, "keys") and callable(getattr(obj, "keys")): + return list(obj.keys()) + + # For tensorclass, try to get from _tensordict + if hasattr(obj, "_tensordict"): + td = obj._tensordict + if hasattr(td, "keys") and callable(getattr(td, "keys")): + return list(td.keys()) + except Exception: + pass + + # Fallback: try to iterate over attributes + keys = [] + for attr in dir(obj): + if not attr.startswith("_"): + val = getattr(obj, attr, None) + if isinstance(val, torch.Tensor): + keys.append(attr) + + return keys + + +def _is_tensordict_instance(obj) -> bool: + """Check if object is a TensorDict or tensorclass instance.""" + try: + from tensordict import TensorDict as TD + + # Check for regular TensorDict + if isinstance(obj, TD): + return True + # Check for tensorclass (has _is_tensorclass attribute) + if hasattr(obj, "_is_tensorclass") and obj._is_tensorclass: + return True + return False + except ImportError: + return False + + +def _infer_tensor_schema(tensor: torch.Tensor) -> Tensor: + """Infer Tensor schema from a PyTorch tensor. + + :param tensor: PyTorch tensor to infer. + :returns: Tensor component with inferred dtype, shape, and checks. + """ + # Infer dtype + dtype = tensor.dtype + + # Infer shape (None for dynamic dimensions) + shape = tuple(s if s > 0 else None for s in tensor.shape) + + # Infer value range statistics + checks = _infer_tensor_checks(tensor) + + return Tensor(dtype=dtype, shape=shape, checks=checks) + + +def _infer_tensor_checks(tensor: torch.Tensor) -> list | None: + """Infer check constraints from tensor data. + + :param tensor: PyTorch tensor to analyze. + :returns: List of Check objects or None if no meaningful checks can be inferred. + """ + import pandera as pa + + # Only infer checks for numeric tensors + if not torch.is_floating_point(tensor) and not torch.is_complex(tensor): + return [] + + has_nan = torch.isnan(tensor).any().item() + has_inf = torch.isinf(tensor).any().item() + + checks = [] + + if not has_nan and not has_inf: + try: + min_val = tensor.min() + max_val = tensor.max() + + # Use torch.isfinite on the tensor values + if torch.all(torch.isfinite(min_val)) and torch.all( + torch.isfinite(max_val) + ): + checks.append( + pa.Check.greater_than_or_equal_to(min_val.item()) + ) + checks.append(pa.Check.less_than_or_equal_to(max_val.item())) + except (AttributeError, RuntimeError): + pass + + return checks if checks else None diff --git a/pandera/strategies/__init__.py b/pandera/strategies/__init__.py index 074805478..71d2e49a1 100644 --- a/pandera/strategies/__init__.py +++ b/pandera/strategies/__init__.py @@ -3,3 +3,40 @@ Example: ``import pandera.strategies.pandas_strategies`` or ``import pandera.strategies.xarray_strategies``. """ + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .tensordict_strategies import ( + tensorclass_strategy, + tensordict_strategy, + ) + +__all__ = [ + "pandas_strategies", + "xarray_strategies", + "tensordict_strategies", +] + +# Import backend-specific strategy modules +try: + from . import pandas_strategies +except ImportError: + pass + +try: + from . import xarray_strategies +except ImportError: + pass + +try: + from . import tensordict_strategies as tensordict_strategies_module + + # Re-export functions at module level for convenience + tensordict_strategy = tensordict_strategies_module.tensordict_strategy + tensorclass_strategy = tensordict_strategies_module.tensorclass_strategy + __all__.extend(["tensordict_strategy", "tensorclass_strategy"]) +except ImportError: + pass diff --git a/pandera/strategies/tensordict_strategies.py b/pandera/strategies/tensordict_strategies.py new file mode 100644 index 000000000..2407dfe8b --- /dev/null +++ b/pandera/strategies/tensordict_strategies.py @@ -0,0 +1,205 @@ +"""Generate synthetic TensorDict data from schema definitions. + +This module generates :class:`tensordict.TensorDict` and +:class:`tensordict.tensorclass` objects that conform to +:class:`~pandera.api.tensordict.container.TensorDictSchema` specifications. + +Built on top of the +`hypothesis `_ +package. +""" + +from __future__ import annotations + +from functools import wraps +from typing import Any, TypeVar, cast + +import torch + +from pandera.strategies.base_strategies import HAS_HYPOTHESIS + +if HAS_HYPOTHESIS: + from hypothesis import strategies as st + from hypothesis.strategies import SearchStrategy, composite +else: # pragma: no cover + from pandera.strategies.base_strategies import SearchStrategy, composite + +F = TypeVar("F") + + +def _strategy_import_error(fn: F) -> F: + """Decorator to raise ImportError when hypothesis is missing.""" + + @wraps(fn) + def _wrapper(*args: Any, **kwargs: Any) -> Any: + if not HAS_HYPOTHESIS: + raise ImportError( + 'Strategies for generating data requires "hypothesis" ' + "to be installed.\n" + "pip install pandera[strategies]" + ) + return fn(*args, **kwargs) + + return cast(F, _wrapper) + + +@_strategy_import_error +def tensordict_strategy( + schema, + size: int | None = None, +) -> SearchStrategy: + """Create a strategy from a TensorDictSchema. + + :param schema: The TensorDictSchema to use. + :param size: Default size for None dimensions. + :returns: hypothesis strategy producing conforming TensorDicts. + """ + default_size = size or 3 + + @composite + def generate_tensordict(draw): + from tensordict import TensorDict + + batch_size_tuple = schema.batch_size or (default_size,) + + data = {} + + for key_name, tensor_schema in schema.keys.items(): + shape_list = ( + list(tensor_schema.shape) if tensor_schema.shape else [] + ) + + # Resolve dynamic dimensions (None -> default_size) + resolved_shape = tuple( + s if s is not None else default_size for s in shape_list + ) + + # Calculate total elements needed from resolved shape + total_elements = 1 + for dim in resolved_shape: + total_elements *= dim + + dtype_str = str(tensor_schema.dtype).lower() + + if "bool" in dtype_str: + arr_data = draw( + st.lists( + st.booleans(), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.bool).reshape( + resolved_shape + ) + elif "float" in dtype_str or "double" in dtype_str: + arr_data = draw( + st.lists( + st.floats(allow_nan=False, allow_infinity=False), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.float32).reshape( + resolved_shape + ) + else: + # Integer types + arr_data = draw( + st.lists( + st.integers(min_value=0, max_value=1000), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.int64).reshape( + resolved_shape + ) + + data[key_name] = arr + + return TensorDict(data, batch_size=batch_size_tuple) + + return generate_tensordict() + + +@_strategy_import_error +def tensorclass_strategy( + cls, + schema, + size: int | None = None, +) -> SearchStrategy: + """Create a strategy for generating tensorclass instances. + + :param cls: The tensorclass class to generate. + :param schema: The TensorDictSchema defining the structure. + :param size: Default size for None dimensions. + :returns: hypothesis strategy producing conforming tensorclass instances. + """ + default_size = size or 3 + + @composite + def generate_tensorclass(draw): + from tensordict import tensorclass as tc + + batch_size_tuple = schema.batch_size or (default_size,) + + data = {} + + for key_name, tensor_schema in schema.keys.items(): + shape_list = ( + list(tensor_schema.shape) if tensor_schema.shape else [] + ) + + # Resolve dynamic dimensions + resolved_shape = tuple( + s if s is not None else default_size for s in shape_list + ) + + # Calculate total elements needed from resolved shape + total_elements = 1 + for dim in resolved_shape: + total_elements *= dim + + dtype_str = str(tensor_schema.dtype).lower() + + if "bool" in dtype_str: + arr_data = draw( + st.lists( + st.booleans(), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.bool).reshape( + resolved_shape + ) + elif "float" in dtype_str or "double" in dtype_str: + arr_data = draw( + st.lists( + st.floats(allow_nan=False, allow_infinity=False), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.float32).reshape( + resolved_shape + ) + else: + # Integer types + arr_data = draw( + st.lists( + st.integers(min_value=0, max_value=1000), + min_size=total_elements, + max_size=total_elements, + ) + ) + arr = torch.tensor(arr_data, dtype=torch.int64).reshape( + resolved_shape + ) + + data[key_name] = arr + + return cls(**data, batch_size=batch_size_tuple) + + return generate_tensorclass() diff --git a/pandera/tensordict.py b/pandera/tensordict.py index 65380b0b5..036643f92 100644 --- a/pandera/tensordict.py +++ b/pandera/tensordict.py @@ -2,7 +2,7 @@ from __future__ import annotations -# mypy: disable-error-code=attr-defined +import warnings from typing import TYPE_CHECKING from pandera import errors @@ -10,11 +10,10 @@ try: from pandera.engines.tensordict_engine import DataType as _DataType - DataType: type[_DataType] | None = _DataType # type: ignore[misc] + DataType: type[_DataType] | None = _DataType except ImportError: - DataType = None # type: ignore[misc, assignment] + DataType = None -# Import actual implementations (not just for type checking) from pandera.api.tensordict.components import Tensor from pandera.api.tensordict.container import TensorDictSchema from pandera.api.tensordict.model import TensorDictModel @@ -27,10 +26,45 @@ "Field", "errors", "DataType", - # Error classes "SchemaError", "SchemaErrors", ] -# Import error classes for convenience +if TYPE_CHECKING: + from pandera.schema_inference.tensordict import ( + infer_schema as _infer_schema, + ) +else: + try: + from pandera.schema_inference.tensordict import infer_schema + + __all__.append("infer_schema") + except ImportError as e: + warnings.warn(f"Could not import infer_schema: {e}") + infer_schema = None # type: ignore[assignment] + from pandera.errors import SchemaError, SchemaErrors + +# Import IO module if available +try: + from pandera.io.tensordict_io import ( + from_json, + from_yaml, + load, + save, + to_json, + to_yaml, + ) + + __all__.extend( + [ + "from_json", + "from_yaml", + "to_json", + "to_yaml", + "save", + "load", + ] + ) +except ImportError: + pass diff --git a/specs/pytorch-tensordict.md b/specs/pytorch-tensordict.md index 28b8f9c33..b40050d17 100644 --- a/specs/pytorch-tensordict.md +++ b/specs/pytorch-tensordict.md @@ -976,6 +976,268 @@ validated = MySchema.validate(batch) # Direct validation - Implementation of specialized TensorDict checks (e.g., cross-entry constraints). - Comprehensive test suite coverage across all backend features. +### Phase 4: Schema Inference, IO, and Hypothesis Strategies +- **Schema Inference**: Infer TensorDict schemas from data using `pa.infer_schema()`. +- **IO Serialization/Deserialization**: Support for saving/loading TensorDicts with schema validation (JSON, YAML, Torch formats). +- **Hypothesis Strategies**: Generate test data with Hypothesis for property-based testing of TensorDict schemas. + +--- + +### 7.1 Schema Inference + +Automatically infer a `TensorDictSchema` from existing data: + +```python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Existing data +data = TensorDict({ + "observation": torch.randn(100, 64), + "action": torch.randint(0, 4, (100,)), + "reward": torch.randn(100), +}, batch_size=[100]) + +# Infer schema from data +schema = pa.infer_schema(data) +print(schema) + +# Output includes inferred dtypes, shapes, and value ranges +``` + +**Features:** +- Automatically detect dtypes (torch.float32, torch.int64, etc.) +- Infer shape dimensions (including batch_size) +- Detect value ranges and patterns from sample data +- Support for both `TensorDict` and `tensorclass` objects + +### 7.2 IO Serialization/Deserialization + +#### 7.2.1 Save/Load with Schema Validation + +```python +import torch +from tensordict import TensorDict +import pandera.tensordict as pa + +# Define schema +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 64)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + "reward": pa.Tensor(dtype=torch.float32, shape=(None,)), + }, + batch_size=(1000,), +) + +# Save dataset with schema metadata +dataset = TensorDict({ + "observation": torch.randn(1000, 64), + "action": torch.randint(0, 4, (1000,)), + "reward": torch.randn(1000), +}, batch_size=[1000]) + +# Save to disk +schema.save(dataset, "./data/dataset.pt") + +# Load and validate +loaded = schema.load("./data/dataset.pt") +``` + +#### 7.2.2 YAML/JSON Configuration Files + +Define schemas in configuration files: + +```python +import pandera.tensordict as pa + +# Load schema from YAML +schema = pa.TensorDictSchema.from_yaml("config/schema.yaml") + +# Save schema to YAML +schema.to_yaml("config/schema.yaml") +``` + +**YAML Example:** +```yaml +keys: + observation: + dtype: float32 + shape: [null, 64] + action: + dtype: int64 + shape: [null] + reward: + dtype: float32 + shape: [null] +batch_size: [1000] +coerce: true +``` + +#### 7.2.3 Integration with Memmap and Large Datasets + +```python +import torch +from tensordict import TensorDict, MemmapTensor +import pandera.tensordict as pa + +# Schema for memmap-backed datasets +schema = pa.TensorDictSchema( + keys={ + "image": pa.Tensor(dtype=torch.uint8, shape=(None, 224, 224, 3)), + "label": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(10000,), +) + +# Create memmap-backed dataset +dataset = TensorDict({ + "image": MemmapTensor(10000, 224, 224, 3, dtype=torch.uint8), + "label": MemmapTensor(10000, dtype=torch.int64), +}, batch_size=[10000]) + +# Save to disk with schema +schema.save(dataset, "./data/large_dataset") + +# Load with validation +loaded = schema.load("./data/large_dataset") +``` + +### 7.3 Hypothesis Strategies + +Generate test data for property-based testing: + +```python +import torch +from hypothesis import given, strategies as st +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategies as tgt_strats + +# Define schema +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(32,), +) + +# Generate valid TensorDicts for testing +@given(tgt_strats.tensordict(schema)) +def test_policy_output(td): + """Property-based test for policy network.""" + # schema validates automatically + assert td["observation"].shape == (32, 10) + assert td["action"].dtype == torch.int64 + +# Generate data with custom constraints +custom_strategy = tgt_strats.tensordict( + schema, + observation_kwargs={"min_value": -1.0, "max_value": 1.0}, +) + +@given(custom_strategy) +def test_normalized_inputs(td): + """Test that observations are normalized.""" + assert td["observation"].min() >= -1.0 + assert td["observation"].max() <= 1.0 + +# Generate tensorclasses +@tensorclass +class RLData: + observation: torch.Tensor + action: torch.Tensor + +schema = pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 64)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(64,), +) + +@given(tgt_strats.tensorclass(RLData, schema)) +def test_tensorclass_generation(tc): + """Generate and validate tensorclass objects.""" + assert tc.batch_size == [64] +``` + +**Features:** +- Automatic strategy generation from schema definitions +- Support for dtype constraints (e.g., `min_value`, `max_value`) +- Integration with Hypothesis' existing strategies +- Works with both `TensorDict` and `tensorclass` + +### 7.4 Complete Phase 4 Example: ML Pipeline with Validation + +```python +import torch +from tensordict import TensorDict, tensorclass +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategies as tgt_strats +from hypothesis import given, settings +import yaml + +# Define schema in YAML config +config = """ +keys: + observation: + dtype: float32 + shape: [null, 128] + action: + dtype: int64 + shape: [null] + reward: + dtype: float32 + shape: [null] +batch_size: [256] +coerce: true +""" + +# Save schema to file +with open("config/rl_schema.yaml", "w") as f: + f.write(config) + +# Load schema from config +schema = pa.TensorDictSchema.from_yaml("config/rl_schema.yaml") + +# Phase 4.1: Infer schema from sample data +sample_data = TensorDict({ + "observation": torch.randn(256, 128), + "action": torch.randint(0, 4, (256,)), + "reward": torch.randn(256), +}, batch_size=[256]) + +inferred_schema = pa.infer_schema(sample_data) +assert inferred_schema.batch_size == schema.batch_size + +# Phase 4.2: IO with validation +schema.save(sample_data, "./data/rl_dataset.pt") +loaded = schema.load("./data/rl_dataset.pt") + +# Phase 4.3: Hypothesis-based testing +@given(tgt_strats.tensordict(schema)) +@settings(max_examples=100) +def test_training_batch(td): + """Property-based test for RL training batches.""" + assert td["observation"].shape == (256, 128) + assert td["action"].max() < 4 # Action space size + assert td["reward"].dtype == torch.float32 + +# Run tests +test_training_batch() + +# Phase 4.4: Schema evolution +# Save schema for version tracking +schema.to_yaml("config/rl_schema_v1.yaml") + +# Later: load and evolve schema +updated_schema = pa.TensorDictSchema.from_yaml("config/rl_schema_v1.yaml") +updated_schema.keys["new_feature"] = pa.Tensor(dtype=torch.float32, shape=(None,)) +updated_schema.to_yaml("config/rl_schema_v2.yaml") +``` + --- ## 6. Architecture @@ -1001,5 +1263,34 @@ pandera/ │ └── register.py # Backend registration ├── engines/ │ └── tensordict_engine.py # Engine for torch dtype registry +├── strategies/ +│ └── tensordict_strategies.py # Hypothesis strategies +└── schema_inference/ + └── tensordict.py # Schema inference logic └── tensordict.py # Entry point: import pandera.tensordict as pa ``` + +### 6.2 Phase 4 API + +```python +import pandera.tensordict as pa +from pandera.strategies import tensordict_strategies as tgt_strats + +# Schema inference +schema = pa.infer_schema(data) + +# IO operations +schema.save(tensor_dict, "path.pt") +loaded = schema.load("path.pt") + +# YAML/JSON support +schema.to_yaml("config.yaml") +schema = pa.TensorDictSchema.from_yaml("config.yaml") + +# Hypothesis strategies +@given(tgt_strats.tensordict(schema)) +def test_my_function(td): + ... +``` + +--- diff --git a/tests/tensordict/test_tensordict_container.py b/tests/tensordict/test_tensordict_container.py index a0dc9f741..3266ceb37 100644 --- a/tests/tensordict/test_tensordict_container.py +++ b/tests/tensordict/test_tensordict_container.py @@ -663,5 +663,241 @@ def test_schema_error_reason_codes(self): assert SchemaErrorReason.WRONG_DATATYPE in reason_codes +@torch_condition +class TestTensorDictCoercion: + """Tests for TensorDict dtype coercion.""" + + def test_coerce_dtype_basic(self): + """Test basic dtype coercion with coerce=True.""" + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "data": Tensor(dtype=torch.float32, shape=(None, 10)), + }, + batch_size=(16,), + coerce=True, + ) + + td = TensorDict( + {"data": torch.randn(16, 10).to(torch.float64)}, + batch_size=[16], + ) + + assert td["data"].dtype == torch.float64 + result = schema.validate(td) + assert result["data"].dtype == torch.float32 + + def test_coerce_dtype_multiple_tensors(self): + """Test coercion of multiple tensors with different dtypes.""" + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "float_data": Tensor(dtype=torch.float32, shape=(None, 10)), + "int_data": Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(16,), + coerce=True, + ) + + td = TensorDict( + { + "float_data": torch.randn(16, 10).to(torch.float64), + "int_data": torch.arange(16).to(torch.int32), + }, + batch_size=[16], + ) + + result = schema.validate(td) + assert result["float_data"].dtype == torch.float32 + assert result["int_data"].dtype == torch.int64 + + def test_coerce_dtype_with_shape(self): + """Test coercion preserves shape.""" + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "data": Tensor(dtype=torch.float32, shape=(32, 10)), + }, + batch_size=(32,), + coerce=True, + ) + + td = TensorDict( + {"data": torch.randn(32, 10).to(torch.float64)}, + batch_size=[32], + ) + + result = schema.validate(td) + assert result["data"].shape == (32, 10) + assert result["data"].dtype == torch.float32 + + def test_coerce_dtype_tensorclass(self): + """Test coercion with tensorclass objects.""" + from pandera.tensordict import Tensor, TensorDictSchema + + @tensorclass + class TCData: + observation: torch.Tensor + action: torch.Tensor + + schema = TensorDictSchema( + keys={ + "observation": Tensor(dtype=torch.float32, shape=(None, 10)), + "action": Tensor(dtype=torch.int64, shape=(None,)), + }, + batch_size=(16,), + coerce=True, + ) + + tc = TCData( + observation=torch.randn(16, 10).to(torch.float64), + action=torch.randint(0, 5, (16,)).to(torch.int32), + batch_size=[16], + ) + + result = schema.validate(tc) + assert result.observation.dtype == torch.float32 + assert result.action.dtype == torch.int64 + + def test_coerce_dtype_with_value_checks(self): + """Test coercion works with value checks.""" + from pandera import Check + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "data": Tensor( + dtype=torch.float32, + shape=(None,), + checks=Check.gt(0.0), + ), + }, + batch_size=(10,), + coerce=True, + ) + + td = TensorDict( + {"data": torch.rand(10).to(torch.float64) * 2 + 1}, + batch_size=[10], + ) + + result = schema.validate(td) + assert result["data"].dtype == torch.float32 + # Check that value validation still works + assert (result["data"] > 0.0).all() + + def test_coerce_dtype_lazy_validation(self): + """Test coercion with lazy validation.""" + from pandera import Check, errors + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "a": Tensor( + dtype=torch.float32, + shape=(None, 10), + checks=[Check.gt(0.0)], # Will fail for random normal data + ), + "b": Tensor(dtype=torch.int64), + }, + batch_size=(16,), + coerce=True, + ) + + td = TensorDict( + { + "a": torch.randn(16, 10).to( + torch.float64 + ), # Wrong dtype + value check + "b": torch.arange(16).to(torch.int32), # Wrong dtype + }, + batch_size=[16], + ) + + with pytest.raises(errors.SchemaErrors) as exc_info: + schema.validate(td, lazy=True) + + errors = exc_info.value.schema_errors + assert len(errors) >= 1 + + # Verify dtypes were coerced despite validation failures + result = exc_info.value.data + assert result["a"].dtype == torch.float32 + assert result["b"].dtype == torch.int64 + + def test_coerce_dtype_with_schema_level_and_tensor_level(self): + """Test that both schema-level and tensor-level coerce flags work.""" + from pandera.tensordict import Tensor, TensorDictSchema + + # Schema-level coerce applies to all tensors + schema = TensorDictSchema( + keys={ + "a": Tensor(dtype=torch.float32), # No tensor-level coerce + "b": Tensor(dtype=torch.int64), + }, + batch_size=(10,), + coerce=True, + ) + + td = TensorDict( + { + "a": torch.rand(10).to(torch.float64) * 2 + 1, + "b": torch.arange(10).to(torch.int32), + }, + batch_size=[10], + ) + + result = schema.validate(td) + assert result["a"].dtype == torch.float32 + assert result["b"].dtype == torch.int64 + + def test_coerce_dtype_without_schema_level(self): + """Test that without coerce=True, dtypes are not coerced.""" + from pandera import errors + from pandera.tensordict import Tensor, TensorDictSchema + + schema = TensorDictSchema( + keys={ + "data": Tensor(dtype=torch.float32), + }, + batch_size=(10,), + # coerce=False (default) + ) + + td = TensorDict( + {"data": torch.rand(10).to(torch.float64) * 2 + 1}, + batch_size=[10], + ) + + with pytest.raises(errors.SchemaError): + schema.validate(td) + + def test_coerce_dtype_error_handling(self): + """Test error handling during coercion.""" + from pandera import errors + from pandera.tensordict import Tensor, TensorDictSchema + + # This would fail if we tried to coerce a non-tensor + schema = TensorDictSchema( + keys={ + "data": Tensor(dtype=torch.float32), + }, + batch_size=(10,), + coerce=True, + ) + + td = TensorDict( + {"data": torch.rand(10).to(torch.float64) * 2 + 1}, + batch_size=[10], + ) + + # Coercion should succeed + result = schema.validate(td) + assert result["data"].dtype == torch.float32 + + if __name__ == "__main__": pytest.main([__file__, "-v"]) diff --git a/tests/tensordict/test_tensordict_io.py b/tests/tensordict/test_tensordict_io.py new file mode 100644 index 000000000..26e0a44fd --- /dev/null +++ b/tests/tensordict/test_tensordict_io.py @@ -0,0 +1,174 @@ +"""Tests for TensorDict schema serialization/deserialization.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest +import torch +from tensordict import TensorDict, tensorclass + +import pandera.tensordict as pa + + +@pytest.fixture +def sample_schema(): + """Create a sample schema for testing.""" + return pa.TensorDictSchema( + keys={ + "observation": pa.Tensor(dtype=torch.float32, shape=(None, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(None,)), + "reward": pa.Tensor(dtype=torch.float32, shape=(None,)), + }, + batch_size=(32,), + coerce=True, + ) + + +@pytest.fixture +def sample_tensor_dict(): + """Create a sample TensorDict for testing.""" + return TensorDict( + { + "observation": torch.randn(32, 10), + "action": torch.randint(0, 4, (32,)), + "reward": torch.randn(32), + }, + batch_size=[32], + ) + + +class TestSchemaSerialization: + """Test schema serialization functionality.""" + + def test_serialize_schema_to_dict(self, sample_schema): + """Test serializing schema to dictionary.""" + from pandera.io.tensordict_io import serialize_schema + + serialized = serialize_schema(sample_schema) + + assert isinstance(serialized, dict) + assert "schema_type" in serialized + assert serialized["schema_type"] == "tensordict" + + def test_serialize_deserialize_roundtrip(self, sample_schema): + """Test round-trip serialization/deserialization.""" + # Serialize to YAML string + yaml_str = pa.to_yaml(sample_schema) + + # Deserialize from YAML string + deserialized = pa.from_yaml(yaml_str) + + assert isinstance(deserialized, pa.TensorDictSchema) + assert len(deserialized.keys) == len(sample_schema.keys) + + def test_serialize_deserialize_json(self, sample_schema): + """Test JSON serialization/deserialization.""" + # Serialize to JSON string + json_str = pa.to_json(sample_schema) + + # Deserialize from JSON string + deserialized = pa.from_json(json_str) + + assert isinstance(deserialized, pa.TensorDictSchema) + assert len(deserialized.keys) == len(sample_schema.keys) + + def test_serialize_with_path_object(self, sample_schema, tmp_path): + """Test serialization with Path objects.""" + yaml_path = tmp_path / "schema.yaml" + + pa.to_yaml(sample_schema, yaml_path) + + deserialized = pa.from_yaml(yaml_path) + + assert isinstance(deserialized, pa.TensorDictSchema) + + def test_save_load_with_tensor_dict( + self, sample_schema, sample_tensor_dict, tmp_path + ): + """Test saving/loading TensorDict with schema.""" + save_path = tmp_path / "data.pt" + + # Save + pa.save(sample_schema, sample_tensor_dict, save_path) + + # Load and validate + loaded = pa.load(save_path) + + assert isinstance(loaded, TensorDict) + + def test_save_load_with_custom_batch_size(self, tmp_path): + """Test save/load with custom batch sizes.""" + schema = pa.TensorDictSchema( + keys={ + "data": pa.Tensor(dtype=torch.float32, shape=(None, 100)), + }, + batch_size=(64,), + ) + + td = TensorDict( + { + "data": torch.randn(64, 100), + }, + batch_size=[64], + ) + + save_path = tmp_path / "custom.pt" + pa.save(schema, td, save_path) + + loaded = pa.load(save_path) + # batch_size is returned as torch.Size + assert tuple(loaded.batch_size) == (64,) + + +class TestTensorClassSerialization: + """Test tensorclass serialization functionality.""" + + def test_infer_schema_with_tensorclass(self): + """Test schema inference for tensorclass objects.""" + + @tensorclass + class RLData: + observation: torch.Tensor + reward: torch.Tensor + + tc = RLData( + observation=torch.randn(16, 8), + reward=torch.randn(16), + batch_size=[16], + ) + + schema = pa.infer_schema(tc) + + assert isinstance(schema, pa.TensorDictSchema) + assert len(schema.keys) == 2 + + def test_infer_schema_tensorclass_keys(self): + """Test that tensorclass keys are correctly inferred.""" + + @tensorclass + class Data: + x: torch.Tensor + y: torch.Tensor + + tc = Data(x=torch.randn(8, 4), y=torch.randn(8), batch_size=[8]) + + schema = pa.infer_schema(tc) + + # Check that both keys are present (tensorclass attributes) + assert "x" in schema.keys + assert "y" in schema.keys + + def test_tensorclass_batch_size_inference(self): + """Test that tensorclass batch_size is correctly inferred.""" + + @tensorclass + class Data: + obs: torch.Tensor + + tc = Data(obs=torch.randn(24, 10), batch_size=[24]) + + schema = pa.infer_schema(tc) + assert tuple(schema.batch_size) == (24,) diff --git a/tests/tensordict/test_tensordict_schema_inference.py b/tests/tensordict/test_tensordict_schema_inference.py new file mode 100644 index 000000000..d3e81899f --- /dev/null +++ b/tests/tensordict/test_tensordict_schema_inference.py @@ -0,0 +1,113 @@ +"""Tests for TensorDict schema inference.""" + +from __future__ import annotations + +import pytest +import torch +from tensordict import TensorDict, tensorclass + +import pandera.tensordict as pa + + +@pytest.fixture +def sample_tensor_dict(): + """Create a sample TensorDict for testing.""" + return TensorDict( + { + "observation": torch.randn(32, 10), + "action": torch.randint(0, 4, (32,)), + "reward": torch.randn(32), + "done": torch.zeros(32, dtype=torch.bool), + }, + batch_size=[32], + ) + + +class TestSchemaInference: + """Test schema inference functionality.""" + + def test_infer_schema_basic(self, sample_tensor_dict): + """Test basic schema inference from TensorDict.""" + schema = pa.infer_schema(sample_tensor_dict) + + assert isinstance(schema, pa.TensorDictSchema) + assert len(schema.keys) == 4 + assert "observation" in schema.keys + assert "action" in schema.keys + assert "reward" in schema.keys + assert "done" in schema.keys + + def test_infer_schema_batch_size(self, sample_tensor_dict): + """Test that batch_size is inferred correctly.""" + schema = pa.infer_schema(sample_tensor_dict) + + # batch_size might be torch.Size or tuple, both are acceptable + assert schema.batch_size is not None + if hasattr(schema.batch_size, "__iter__"): + assert tuple(schema.batch_size) == (32,) + + def test_infer_schema_dtype(self, sample_tensor_dict): + """Test that dtypes are inferred correctly.""" + schema = pa.infer_schema(sample_tensor_dict) + + # Compare using string representation since dtype may be wrapped + obs_dtype = str(schema.keys["observation"].dtype) + assert "float32" in obs_dtype or "torch.float32" in obs_dtype + + action_dtype = str(schema.keys["action"].dtype) + assert "int64" in action_dtype or "torch.int64" in action_dtype + + def test_infer_schema_shape(self, sample_tensor_dict): + """Test that shapes are inferred correctly.""" + schema = pa.infer_schema(sample_tensor_dict) + + obs_shape = ( + tuple(schema.keys["observation"].shape) + if schema.keys["observation"].shape + else None + ) + assert obs_shape == (32, 10) + + def test_infer_schema_with_tensorclass(self): + """Test schema inference with tensorclass.""" + + @tensorclass + class Data: + x: torch.Tensor + y: torch.Tensor + + tc = Data( + x=torch.randn(16, 8), y=torch.randint(0, 2, (16,)), batch_size=[16] + ) + + schema = pa.infer_schema(tc) + + assert isinstance(schema, pa.TensorDictSchema) + assert len(schema.keys) == 2 + + def test_infer_schema_value_ranges(self, sample_tensor_dict): + """Test that value range checks are inferred.""" + schema = pa.infer_schema(sample_tensor_dict) + + # Check that numeric tensors have range checks + obs_schema = schema.keys["observation"] + assert len(obs_schema.checks) >= 2 # min and max bounds + + def test_infer_schema_invalid_type(self): + """Test that invalid types raise TypeError.""" + with pytest.raises(TypeError, match="Expected TensorDict"): + pa.infer_schema({"not": "a tensordict"}) + + def test_infer_schema_boolean_dtype(self, sample_tensor_dict): + """Test that boolean dtypes are inferred correctly.""" + schema = pa.infer_schema(sample_tensor_dict) + + done_dtype = str(schema.keys["done"].dtype) + assert "bool" in done_dtype or "torch.bool" in done_dtype + + def test_infer_schema_integer_dtypes(self, sample_tensor_dict): + """Test that integer dtypes are handled correctly.""" + schema = pa.infer_schema(sample_tensor_dict) + + action_dtype = str(schema.keys["action"].dtype) + assert "int64" in action_dtype or "torch.int64" in action_dtype diff --git a/tests/tensordict/test_tensordict_strategies.py b/tests/tensordict/test_tensordict_strategies.py new file mode 100644 index 000000000..ac3c1f0ef --- /dev/null +++ b/tests/tensordict/test_tensordict_strategies.py @@ -0,0 +1,123 @@ +"""Tests for TensorDict Hypothesis strategies.""" + +from __future__ import annotations + +import pytest + +try: + from hypothesis import given, settings + from hypothesis.strategies import composite + + HAS_HYPOTHESIS = True +except ImportError: + HAS_HYPOTHESIS = False + +import torch + +import pandera.tensordict as pa + + +@pytest.mark.skipif(not HAS_HYPOTHESIS, reason="hypothesis not installed") +class TestTensorDictStrategies: + """Test TensorDict Hypothesis strategies.""" + + @pytest.fixture + def simple_schema(self): + """Create a simple schema for testing with scalar values (per batch). + + Note: With batch_size=(N,), tensors should have shape (N, ...) to include + the batch dimension in their shape. + """ + return pa.TensorDictSchema( + keys={ + "data": pa.Tensor(dtype=torch.float32, shape=(8,)), + }, + batch_size=(8,), + ) + + @pytest.fixture + def complex_schema(self): + """Create a more complex schema with multiple tensors.""" + return pa.TensorDictSchema( + keys={ + "obs": pa.Tensor(dtype=torch.float32, shape=(8, 10)), + "action": pa.Tensor(dtype=torch.int64, shape=(8,)), + "reward": pa.Tensor(dtype=torch.float32, shape=(8,)), + }, + batch_size=(8,), + ) + + def test_tensordict_strategy_generates_valid_data(self, simple_schema): + """Test that strategy generates valid TensorDicts.""" + from pandera.strategies import tensordict_strategy + + @given(tensordict_strategy(simple_schema)) + @settings(max_examples=5) + def check(td): + # Validate that generated data conforms to schema + validated = simple_schema.validate(td) + + assert isinstance(validated, type(td)) + assert "data" in td.keys() + + check() + + def test_strategy_batch_size(self, complex_schema): + """Test that strategy respects batch_size.""" + from pandera.strategies import tensordict_strategy + + @given(tensordict_strategy(complex_schema)) + @settings(max_examples=5) + def check(td): + assert tuple(td.batch_size) == (8,) + + check() + + def test_strategy_dtypes(self, complex_schema): + """Test that strategy generates correct dtypes.""" + from pandera.strategies import tensordict_strategy + + @given(tensordict_strategy(complex_schema)) + @settings(max_examples=5) + def check(td): + assert td["obs"].dtype == torch.float32 + assert td["action"].dtype == torch.int64 + assert td["reward"].dtype == torch.float32 + + check() + + def test_strategy_shapes(self, complex_schema): + """Test that strategy generates correct shapes.""" + from pandera.strategies import tensordict_strategy + + @given(tensordict_strategy(complex_schema)) + @settings(max_examples=5) + def check(td): + # Tensors should match schema's full shape (including batch dim) + assert td["obs"].shape == torch.Size([8, 10]) + assert td["action"].shape == torch.Size([8]) + + check() + + def test_tensorclass_strategy(self, complex_schema): + """Test tensorclass strategy generation.""" + from tensordict import tensorclass + + from pandera.strategies import tensorclass_strategy + + @tensorclass + class Data: + obs: torch.Tensor + action: torch.Tensor + reward: torch.Tensor + + strategy = tensorclass_strategy(Data, complex_schema) + + @given(strategy) + @settings(max_examples=5) + def check(tc): + assert tuple(tc.batch_size) == (8,) + assert hasattr(tc, "obs") + assert hasattr(tc, "action") + + check()