'Public ActionCodec2 processor and constrained-generation grammars.' from __future__ import annotations import hashlib import json from collections.abc import Sequence from numbers import Integral from pathlib import Path from typing import Any import numpy as np from transformers.processing_utils import ProcessorMixin from ._validation import active_dimensions from .bpe import BPE_NAME, BPESerializer, SetBPESerializer, serializer_class from .cache import BoundedCache from .configuration import CONFIG_NAME, MODEL_TYPE, ActionCodec2Config from .grammar import ActionCodec2BPEGrammar as ActionCodec2BPEGrammar from .grammar import ActionCodec2Grammar as ActionCodec2Grammar from .hf import ( resolve_pretrained_directory, save_huggingface_metadata, upload_pretrained_directory, ) from .physical_codec import PhysicalCodec from .primitives import PRIMITIVES_NAME, PrimitiveStore class ActionCodec2Processor(ProcessorMixin): 'Hugging Face-style processor for continuous robot action trajectories.' model_type = MODEL_TYPE name = "actioncodec2" is_exact = False set_bpe_is_lossless = True model_input_names = ["input_ids"] attributes = [] def __init__( self, config: ActionCodec2Config, primitives: PrimitiveStore, serializer: SetBPESerializer | BPESerializer, ) -> None: if serializer.scheme != config.set_bpe.scheme: raise ValueError("serializer scheme does not match config.yaml") if tuple(serializer.bins) != tuple(primitives.bins): raise ValueError("BPE bins do not match primitives.npz") if serializer.lookahead != config.set_bpe.lookahead: raise ValueError("BPE lookahead does not match config.yaml") self.config = config self.primitives = primitives self.serializer = serializer self._grammar_cache = BoundedCache(8) self.physical = PhysicalCodec(config, primitives) self.action_dim = config.physical.input_dim self.bins = primitives.bins self.vocab_size = serializer.vocab_size self.lookahead = serializer.lookahead self.requires_previous_first_order = self.physical.requires_previous_first_order zero_cells = [] for dimension, axis in enumerate(primitives.values): matches = np.flatnonzero(np.asarray(axis) == 0) if not len(matches): raise ValueError( "ActionCodec2 artifacts require an exact zero primitive in every axis; " f"dimension {dimension} has none" ) zero_cells.append(int(matches[0])) self._zero_cells = np.asarray(zero_cells, dtype=np.int64) @classmethod def from_pretrained( cls, pretrained_model_name_or_path: str | Path, **kwargs: Any, ) -> ActionCodec2Processor: 'Load a local directory or Hugging Face Hub snapshot.' directory = resolve_pretrained_directory(pretrained_model_name_or_path, **kwargs) config = ActionCodec2Config.from_pretrained(directory) primitives = PrimitiveStore.from_pretrained(directory, input_dim=config.physical.input_dim) serializer = serializer_class(config).from_pretrained( directory, bins=primitives.bins, lookahead=config.set_bpe.lookahead, ) return cls(config, primitives, serializer) def save_pretrained( self, save_directory: str | Path, push_to_hub: bool = False, **kwargs: Any, ) -> list[str]: """Save numerical artifacts plus standard AutoProcessor metadata.""" if not push_to_hub and kwargs: raise TypeError(f"unused save_pretrained keyword(s): {', '.join(sorted(kwargs))}") directory = Path(save_directory) self.config.save_pretrained(directory) self.primitives.save_pretrained(directory) self.serializer.save_pretrained(directory) metadata = save_huggingface_metadata(directory) files = [ directory / CONFIG_NAME, directory / PRIMITIVES_NAME, directory / BPE_NAME, *metadata, ] if push_to_hub: upload_pretrained_directory(directory, **kwargs) return [str(path) for path in files] def to_dict(self, legacy_serialization: bool = True) -> dict[str, object]: """Return lightweight processor metadata, never large primitive/BPE arrays.""" return { "processor_class": self.__class__.__name__, "model_type": self.model_type, "action_dim": self.action_dim, "vocab_size": self.vocab_size, } @property def artifact_fingerprint(self) -> str: """Content hash over config, primitive axes, and ordered BPE rules.""" digest = hashlib.sha256() digest.update( json.dumps(self.config.to_dict(), sort_keys=True, separators=(",", ":")).encode() ) for axis in self.primitives.values: values = np.ascontiguousarray(axis) digest.update(str(values.dtype).encode()) digest.update(np.asarray(values.shape, dtype=np.int64).tobytes()) digest.update(values.tobytes()) digest.update( json.dumps(self.serializer.to_dict(), sort_keys=True, separators=(",", ":")).encode() ) return digest.hexdigest() def __call__( self, actions: np.ndarray, *, previous_first_order: np.ndarray | None = None ) -> dict[str, list[list[int]]]: values = np.asarray(actions) if values.ndim == 2: values = values[None] if previous_first_order is not None: previous_first_order = np.asarray(previous_first_order)[None] return {"input_ids": self.encode_batch(values, previous_first_order=previous_first_order)} def quantize_batch( self, actions: np.ndarray, *, previous_first_order: np.ndarray | None = None ) -> np.ndarray: """Quantize ``[B,T,D]`` using a ``[B,D]`` first-order boundary if required.""" return self.physical.encode_batch(actions, previous_first_order=previous_first_order) def _dense_previous_first_order( self, previous_first_order: np.ndarray | None, dims: tuple[int, ...], batch: int ) -> np.ndarray | None: active = set(dims) for component in self.config.physical.components.values(): present = active.intersection(component.dims) if ( component.primitive_order == 2 and component.space == "so3" and present and present != set(component.dims) ): raise ValueError( "second-order SO(3) requires all three active component dimensions" ) if previous_first_order is None: if any( c.primitive_order == 2 and active.intersection(c.dims) for c in self.config.physical.components.values() ): raise ValueError( f"second-order tokenization requires previous_first_order with shape {(batch, len(dims))} in canonical input units" ) return ( np.zeros((batch, self.action_dim), dtype=np.float32) if self.requires_previous_first_order else None ) values = np.asarray(previous_first_order, dtype=np.float32) if values.shape != (batch, len(dims)) or not np.all(np.isfinite(values)): raise ValueError(f"previous_first_order must be finite with shape {(batch, len(dims))}") dense = np.zeros((batch, self.action_dim), dtype=np.float32) dense[:, dims] = values return dense def tokens_from_cells( self, cells: np.ndarray, *, active_dims: Sequence[int] | None = None, ) -> list[int]: """Apply the lossless tokenizer to one compact or dense cell trajectory.""" return self.serializer.encode(cells, active_dims=active_dims) def configure_encoding( self, *, spatial_search: bool = False, temporal_search: bool = False, search_window: int = 32, search_beam: int = 4, ) -> ActionCodec2Processor: 'Opt into lossless packing search; both stages default to disabled.' self.serializer.configure_encoding( spatial_search=spatial_search, temporal_search=temporal_search, search_window=search_window, search_beam=search_beam, ) return self def cells_from_tokens( self, tokens: Sequence[int], *, active_dims: Sequence[int] | None = None, ) -> np.ndarray: """Losslessly recover dense cells with ``-1`` in absent dimensions.""" return self.serializer.decode(tokens, active_dims=active_dims) def _fill_inactive_cells(self, cells: np.ndarray) -> np.ndarray: values = np.asarray(cells, dtype=np.int64) return np.where(values >= 0, values, self._zero_cells) def encode( self, actions: np.ndarray, *, active_dims: Sequence[int] | None = None, previous_first_order: np.ndarray | None = None, ) -> list[int]: 'Encode ``[T,D_active]`` with an optional compact ``[D_active]`` boundary.' values = np.asarray(actions, dtype=np.float32) if values.ndim != 2 or len(values) < 1 or values.shape[1] < 1: raise ValueError("actions must have non-empty shape [T, D_active]") dims = ( tuple(range(values.shape[1])) if active_dims is None and values.shape[1] < self.action_dim else active_dimensions(active_dims, self.action_dim) ) if len(dims) != values.shape[1]: raise ValueError("active_dims must have one entry per action coordinate") previous = None if previous_first_order is None else np.asarray(previous_first_order)[None] return self.encode_batch(values[None], active_dims=dims, previous_first_order=previous)[0] def encode_batch( self, actions: np.ndarray, *, active_dims: Sequence[int] | None = None, previous_first_order: np.ndarray | None = None, ) -> list[list[int]]: 'Encode a dense or compact ``[B,T,D]`` batch into ragged streams.' values = np.asarray(actions, dtype=np.float32) if values.ndim != 3 or len(values) < 1 or values.shape[1] < 1 or values.shape[2] < 1: raise ValueError("actions must have non-empty shape [B, T, D_active]") dims = active_dimensions(active_dims, self.action_dim) if len(dims) != values.shape[2]: raise ValueError("active_dims must have one entry per action coordinate") full_width = dims == tuple(range(self.action_dim)) if full_width: dense = values else: dense = np.zeros((len(values), values.shape[1], self.action_dim), dtype=np.float32) dense[:, :, dims] = values previous = self._dense_previous_first_order(previous_first_order, dims, len(values)) cells = self.quantize_batch(dense, previous_first_order=previous) if not full_width: cells = np.take(cells, dims, axis=2) return self.serializer.encode_batch(cells, active_dims=dims) def decode( self, tokens: Sequence[int], *, active_dims: Sequence[int] | None = None, previous_first_order: np.ndarray | None = None, ) -> np.ndarray: 'Decode one stream to ``[T,D_active]`` reconstructed actions.' previous = None if previous_first_order is None else np.asarray(previous_first_order)[None] return self.decode_batch([tokens], active_dims=active_dims, previous_first_order=previous)[ 0 ] def decode_batch( self, streams: Sequence[Sequence[int]], *, active_dims: Sequence[int] | None = None, previous_first_order: np.ndarray | None = None, ) -> np.ndarray: 'Decode equal-length streams into one batched physical array.' dims = active_dimensions(active_dims, self.action_dim) stacked_cells = self.serializer.decode_batch(streams, active_dims=dims) dense_cells = self._fill_inactive_cells(stacked_cells) previous = self._dense_previous_first_order(previous_first_order, dims, len(stacked_cells)) decoded = self.physical.decode_batch(dense_cells, previous_first_order=previous) return np.take(decoded, dims, axis=2) def final_first_order( self, streams: Sequence[Sequence[int]], *, previous_first_order: np.ndarray, active_dims: Sequence[int] | None = None, executed_steps: int | None = None, ) -> np.ndarray: 'Return the next compact ``[B,D_active]`` reconstructed boundary.' dims = active_dimensions(active_dims, self.action_dim) cells = [self.cells_from_tokens(stream, active_dims=dims) for stream in streams] if not cells or len({len(row) for row in cells}) != 1: raise ValueError("streams must be non-empty and have equal decoded horizons") if executed_steps is not None: if ( isinstance(executed_steps, bool) or not isinstance(executed_steps, Integral) or not 1 <= executed_steps <= len(cells[0]) ): raise ValueError("executed_steps must be within the decoded horizon") cells = [row[:executed_steps] for row in cells] previous = self._dense_previous_first_order(previous_first_order, dims, len(cells)) result = self.physical.final_first_order( self._fill_inactive_cells(np.stack(cells)), previous_first_order=previous ) return np.take(result, dims, axis=1) def eval(self) -> ActionCodec2Processor: """Return ``self``; the rule processor has no train/eval state.""" return self def to(self, device: object) -> ActionCodec2Processor: """Return ``self``; all runtime arrays are CPU NumPy values.""" del device return self def grammar( self, horizon: int, *, active_dims: Sequence[int] | None = None, ) -> ActionCodec2Grammar | ActionCodec2BPEGrammar: """Build a constrained grammar for one horizon and route support.""" if isinstance(horizon, bool) or not isinstance(horizon, int) or horizon < 1: raise ValueError("horizon must be a positive integer") dims = active_dimensions(active_dims, self.action_dim) key = (horizon, dims) grammar = self._grammar_cache.get(key) if grammar is None: if self.serializer.scheme == "unified_set_bpe": from .unified_set_bpe import UnifiedSetBPEGrammar grammar_class = UnifiedSetBPEGrammar else: grammar_class = ( ActionCodec2BPEGrammar if isinstance(self.serializer, BPESerializer) else ActionCodec2Grammar ) grammar = grammar_class(self.serializer, horizon=horizon, active_dims=dims) self._grammar_cache[key] = grammar return grammar