Source code for signal_dataset._internal.policy

"""Configurable resource limits."""

from dataclasses import dataclass




@dataclass(frozen=True, slots=True)
class ResourcePolicy:
    """Bounds untrusted tensor-bearing records and their descriptors."""

    max_encoded_record_bytes: int = 64 * 1024**3
    max_descriptor_bytes: int = 16 * 1024**2
    max_fields: int = 10_000
    max_axes_per_field: int = 128
    max_coordinate_values: int = 10_000_000
    max_tensor_bytes: int = 16 * 1024**3
    max_total_tensor_bytes: int = 64 * 1024**3
    max_metadata_bytes: int = 16 * 1024**2
    max_json_depth: int = 128
    max_json_nodes: int = 1_000_000

    def __post_init__(self) -> None:
        for name in self.__dataclass_fields__:
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                raise ValueError(f"{name} must be a nonnegative integer")





@dataclass(frozen=True, slots=True)
class ControlPolicy:
    """Bounds untrusted dataset roots, snapshots, manifests, and shard references."""

    max_document_bytes: int = 64 * 1024**2
    max_json_depth: int = 128
    max_json_nodes: int = 1_000_000

    def __post_init__(self) -> None:
        for name in self.__dataclass_fields__:
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError(f"{name} must be a positive integer")