File size: 29,029 Bytes

"""User-facing action layouts compiled into shared joint/EEF trajectories."""

from __future__ import annotations

import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path

import numpy as np
import yaml

from ._validation import optional_finite_array, validate_dimension_partition
from .batch_semantics import (
    canonicalize_batch,
    canonicalize_component_batch,
    externalize_canonical_batch,
    resample_canonical_batch,
)
from .configuration import (
    ConfigError,
    _integer,
    _mapping,
    _number,
    _reject_unknown,
    _required,
    _UniqueKeyLoader,
)
from .semantic import wrap_to_pi as wrap_to_pi
from .trajectory import ActiveTrajectory

_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
_SPACES = {"eef", "joint"}
_INPUTS = {"absolute", "delta"}
_ROTATIONS = {"rotation_vector", "euler_xyz", "quaternion_xyzw"}
_JOINT_WRAPS = {"pi", "none"}
_INTERPOLATIONS = {"linear", "pchip", "cubic"}
_GRIPPER_DIM = {"eef": 6, "joint": 7}


def canonical_preset_name(name: str) -> str:
    """Preserve the historical six-coordinate alias without implying a gripper slot."""
    return "single_joint6_abs_no_gripper" if name == "single_joint6_abs_continuous_gripper" else name


def _format_dims(dims: Sequence[int]) -> str:
    values = tuple(dims)
    if len(values) == 1:
        return str(values[0])
    if values == tuple(range(values[0], values[-1] + 1)):
        return f"{values[0]}..{values[-1]}"
    return ",".join(map(str, values))


def _dims(value: object, path: str) -> tuple[int, ...]:
    if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)):
        raise ConfigError(f"{path} must be a list of dimensions")
    output = tuple(_integer(item, f"{path}[{index}]") for index, item in enumerate(value))
    if not output:
        raise ConfigError(f"{path} must not be empty")
    if len(set(output)) != len(output):
        raise ConfigError(f"{path} contains a repeated dimension")
    return output


@dataclass(frozen=True)
class ComponentSpec:
    """One independently encoded arm/body component in an input tensor."""

    name: str
    space: str
    joint_dims: tuple[int, ...] | None = None
    joint_wrap: str | None = None
    position_dims: tuple[int, ...] | None = None
    rotation_dims: tuple[int, ...] | None = None
    rotation_format: str = "rotation_vector"
    gripper_dim: int | None = None
    gripper_closed_value: float = 0.0
    gripper_open_value: float = 1.0
    gripper_threshold: float | None = None
    gripper_open_is_high: bool = True

    def __post_init__(self) -> None:
        if not isinstance(self.gripper_open_is_high, bool):
            raise ConfigError("gripper_open_is_high must be boolean")
        if self.gripper_threshold is not None:
            _number(self.gripper_threshold, "gripper_threshold")
        if not isinstance(self.name, str) or _NAME.fullmatch(self.name) is None:
            raise ConfigError(f"component name must match {_NAME.pattern!r}")
        if self.space not in _SPACES:
            raise ConfigError(f"component {self.name!r}.space must be joint or eef")
        if self.rotation_format not in _ROTATIONS:
            raise ConfigError(
                f"component {self.name!r}.rotation_format must be one of {sorted(_ROTATIONS)}"
            )
        for label, dims in (
            ("joint_dims", self.joint_dims),
            ("position_dims", self.position_dims),
            ("rotation_dims", self.rotation_dims),
        ):
            if dims is not None:
                _dims(dims, f"component {self.name}.{label}")
        self._validate_motion_dims()
        if self.space == "joint":
            if self.joint_wrap is None:
                object.__setattr__(self, "joint_wrap", "pi")
            if self.joint_wrap not in _JOINT_WRAPS:
                raise ConfigError(
                    f"component {self.name!r}.joint_wrap must be one of {sorted(_JOINT_WRAPS)}"
                )
        elif self.joint_wrap is not None:
            raise ConfigError(
                f"component {self.name!r}.joint_wrap applies only to joint components"
            )
        raw_dims = self.source_dims
        if self.gripper_dim is not None:
            _integer(self.gripper_dim, f"component {self.name}.gripper_dim")
            if self.gripper_dim in raw_dims:
                raise ConfigError(f"component {self.name!r}.gripper_dim overlaps motion dims")
            for anchor in (self.gripper_closed_value, self.gripper_open_value):
                _number(anchor, f"component {self.name}.gripper anchor")
            if self.gripper_closed_value == self.gripper_open_value:
                raise ConfigError(
                    f"component {self.name!r}.gripper_closed_value and gripper_open_value "
                    "must differ"
                )

    def _validate_motion_dims(self) -> None:
        if self.space == "joint":
            if self.joint_dims is None or not 1 <= len(self.joint_dims) <= 7:
                raise ConfigError(f"joint component {self.name!r} requires 1..7 joint_dims")
            if self.position_dims is not None or self.rotation_dims is not None:
                raise ConfigError(f"joint component {self.name!r} cannot define EEF dims")
            return
        if self.joint_dims is not None:
            raise ConfigError(f"EEF component {self.name!r} cannot define joint_dims")
        if self.position_dims is None or len(self.position_dims) != 3:
            raise ConfigError(f"EEF component {self.name!r} requires three position_dims")
        expected = 4 if self.rotation_format == "quaternion_xyzw" else 3
        if self.rotation_dims is None or len(self.rotation_dims) != expected:
            raise ConfigError(
                f"EEF component {self.name!r} requires {expected} rotation_dims for "
                f"{self.rotation_format}"
            )
        if set(self.position_dims).intersection(self.rotation_dims):
            raise ConfigError(f"EEF component {self.name!r} reuses a pose dimension")

    @property
    def source_dims(self) -> tuple[int, ...]:
        """All source motion columns in canonical component order."""
        if self.space == "joint":
            assert self.joint_dims is not None
            return self.joint_dims
        assert self.position_dims is not None and self.rotation_dims is not None
        return (*self.position_dims, *self.rotation_dims)

    @property
    def all_source_dims(self) -> tuple[int, ...]:
        """Motion plus optional binary-gripper source columns."""
        return (
            self.source_dims if self.gripper_dim is None else (*self.source_dims, self.gripper_dim)
        )

    @property
    def active_dims(self) -> tuple[int, ...]:
        """Shared-profile slots emitted by this component."""
        motion_width = len(self.joint_dims or ()) if self.space == "joint" else 6
        dims = tuple(range(motion_width))
        return dims if self.gripper_dim is None else (*dims, _GRIPPER_DIM[self.space])

    @classmethod
    def from_dict(cls, name: str, payload: object, *, path: str) -> ComponentSpec:
        values = _mapping(payload, path)
        allowed = {
            "space",
            "joint_dims",
            "joint_wrap",
            "position_dims",
            "rotation_dims",
            "rotation_format",
            "gripper_dim",
            "gripper_closed_value",
            "gripper_open_value",
            "gripper_threshold",
            "gripper_open_is_high",
        }
        _reject_unknown(values, allowed, path)
        gripper_dim = values.get("gripper_dim")
        if gripper_dim is not None:
            gripper_dim = _integer(gripper_dim, f"{path}.gripper_dim")
        return cls(
            name=name,
            space=str(_required(values, "space", path)),
            joint_dims=None
            if "joint_dims" not in values
            else _dims(values["joint_dims"], f"{path}.joint_dims"),
            joint_wrap=None if values.get("joint_wrap") is None else str(values["joint_wrap"]),
            position_dims=None
            if "position_dims" not in values
            else _dims(values["position_dims"], f"{path}.position_dims"),
            rotation_dims=None
            if "rotation_dims" not in values
            else _dims(values["rotation_dims"], f"{path}.rotation_dims"),
            rotation_format=str(values.get("rotation_format", "rotation_vector")),
            gripper_dim=gripper_dim,
            gripper_threshold=(values.get("gripper_threshold") if values.get("gripper_threshold") is not None
                               else (0.8 if values.get("gripper_open_is_high", True) else 0.2)
                               if "gripper_open_is_high" in values or "gripper_threshold" in values else None),
            gripper_open_is_high=values.get("gripper_open_is_high", True),
            gripper_closed_value=_number(
                values.get("gripper_closed_value", 0.0), f"{path}.gripper_closed_value"
            ),
            gripper_open_value=_number(
                values.get("gripper_open_value", 1.0), f"{path}.gripper_open_value"
            ),
        )

    def to_dict(self) -> dict[str, object]:
        payload: dict[str, object] = {"space": self.space}
        if self.gripper_dim is not None and self.gripper_threshold is not None:
            payload.update(gripper_threshold=self.gripper_threshold,
                           gripper_open_is_high=self.gripper_open_is_high)
        if self.space == "joint":
            payload["joint_dims"] = list(self.joint_dims or ())
            payload["joint_wrap"] = self.joint_wrap
        else:
            payload.update(
                {
                    "position_dims": list(self.position_dims or ()),
                    "rotation_dims": list(self.rotation_dims or ()),
                    "rotation_format": self.rotation_format,
                }
            )
        if self.gripper_dim is not None:
            payload.update(
                {
                    "gripper_dim": self.gripper_dim,
                    "gripper_closed_value": self.gripper_closed_value,
                    "gripper_open_value": self.gripper_open_value,
                }
            )
        return payload


@dataclass(frozen=True)
class PresetConfig:
    """A dataset- and embodiment-independent physical ``[B,T,D]`` contract."""

    name: str
    input_dim: int
    input: str
    components: tuple[ComponentSpec, ...]
    source_hz: float = 15.0
    target_hz: float = 15.0
    interpolation: str = "cubic"
    token_order: str = "time_major"

    def __post_init__(self) -> None:
        object.__setattr__(self, "name", canonical_preset_name(self.name))
        if not isinstance(self.name, str) or _NAME.fullmatch(self.name) is None:
            raise ConfigError(f"preset.name must match {_NAME.pattern!r}")
        _integer(self.input_dim, "preset.input_dim", positive=True)
        if self.input not in _INPUTS:
            raise ConfigError("preset.input must be absolute or delta")
        if not self.components:
            raise ConfigError("preset.components must not be empty")
        if not all(isinstance(component, ComponentSpec) for component in self.components):
            raise ConfigError("preset.components must contain ComponentSpec values")
        names = [component.name for component in self.components]
        if len(set(names)) != len(names):
            raise ConfigError("preset.components contains a duplicate name")
        self._validate_source_partition()
        for rate, path in ((self.source_hz, "source_hz"), (self.target_hz, "target_hz")):
            checked = _number(rate, f"preset.{path}")
            if checked <= 0.0:
                raise ConfigError(f"preset.{path} must be positive and finite")
        if self.interpolation not in _INTERPOLATIONS:
            raise ConfigError(f"preset.interpolation must be one of {sorted(_INTERPOLATIONS)}")
        if self.token_order not in {"time_major", "component_major"}:
            raise ConfigError("preset.token_order must be time_major or component_major")
        if self.input == "delta" and any(
            component.space == "eef" and component.rotation_format != "rotation_vector"
            for component in self.components
        ):
            raise ConfigError("EEF delta components require rotation_format=rotation_vector")

    def _validate_source_partition(self) -> None:
        validate_dimension_partition(
            ((component.name, component.all_source_dims) for component in self.components),
            self.input_dim,
            label="input dim",
            error_type=ConfigError,
        )

    @classmethod
    def from_dict(cls, payload: object, *, path: str = "preset") -> PresetConfig:
        values = _mapping(payload, path)
        allowed = {
            "name",
            "input_dim",
            "input",
            "source_hz",
            "target_hz",
            "interpolation",
            "components",
            "token_order",
        }
        _reject_unknown(values, allowed, path)
        raw_components = _mapping(_required(values, "components", path), f"{path}.components")
        components = tuple(
            ComponentSpec.from_dict(name, raw, path=f"{path}.components.{name}")
            for name, raw in raw_components.items()
        )
        return cls(
            name=str(_required(values, "name", path)),
            input_dim=_integer(
                _required(values, "input_dim", path), f"{path}.input_dim", positive=True
            ),
            input=str(_required(values, "input", path)),
            components=components,
            source_hz=_number(values.get("source_hz", 15.0), f"{path}.source_hz"),
            target_hz=_number(values.get("target_hz", 15.0), f"{path}.target_hz"),
            interpolation=str(values.get("interpolation", "cubic")),
            token_order=str(values.get("token_order", "time_major")),
        )

    @classmethod
    def from_yaml(cls, path: str | Path) -> PresetConfig:
        """Load one strict user preset without executing arbitrary code."""
        source = Path(path)
        try:
            payload = yaml.load(source.read_text(encoding="utf-8"), Loader=_UniqueKeyLoader)
        except yaml.YAMLError as error:
            raise ConfigError(f"cannot parse {source}: {error}") from error
        return cls.from_dict(payload, path=str(source))

    def save_yaml(self, path: str | Path) -> Path:
        """Save the complete reproducible preset and return its path."""
        target = Path(path)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(yaml.safe_dump(self.to_dict(), sort_keys=False), encoding="utf-8")
        return target

    def to_dict(self) -> dict[str, object]:
        payload = {
            "name": self.name,
            "input_dim": self.input_dim,
            "input": self.input,
            "source_hz": self.source_hz,
            "target_hz": self.target_hz,
            "interpolation": self.interpolation,
            "token_order": self.token_order,
            "components": {component.name: component.to_dict() for component in self.components},
        }
        return payload

    def summary(self) -> str:
        """Return a compact action-space label for discovery UIs."""
        spaces = "+".join(dict.fromkeys(component.space for component in self.components))
        return (
            f"{self.name} (D={self.input_dim}, {self.input}, {spaces}, {self.source_hz:g} Hz input)"
        )

    def describe(self) -> str:
        """Describe every source tensor column and its physical interpretation."""
        motion = "position" if self.input == "absolute" else "per-step delta"
        lines = [self.summary()]
        for component in self.components:
            if component.space == "joint":
                assert component.joint_dims is not None
                lines.append(
                    f"  dims {_format_dims(component.joint_dims)}: {component.name} "
                    f"generalized-joint {motion} [rad or registered canonical unit]"
                )
                if self.input == "absolute":
                    lines.append(
                        "    angles in (-pi, pi]; decode returns that interval"
                        if component.joint_wrap == "pi"
                        else "    continuous angles, may exceed one turn; decode never wraps"
                    )
            else:
                assert component.position_dims is not None
                assert component.rotation_dims is not None
                lines.append(
                    f"  dims {_format_dims(component.position_dims)}: {component.name} "
                    f"EEF translation {motion} (x, y, z in listed order) [m]"
                )
                rotation_motion = "orientation" if self.input == "absolute" else "increment"
                rotation_unit = (
                    "unit xyzw quaternion"
                    if component.rotation_format == "quaternion_xyzw"
                    else "rad"
                )
                lines.append(
                    f"  dims {_format_dims(component.rotation_dims)}: {component.name} EEF "
                    f"{rotation_motion} [{component.rotation_format}, {rotation_unit}]"
                )
                if self.input == "delta":
                    lines.append(
                        "    rotation_vector = axis * angle [rad], right-multiplicative "
                        "body-frame increment: R_next = R_previous @ Exp(rotvec)"
                    )
                elif component.rotation_format == "rotation_vector":
                    lines.append("    rotation_vector = axis * angle [rad], not Euler angles")
                elif component.rotation_format == "euler_xyz":
                    lines.append("    euler_xyz = extrinsic x, y, z Euler angles [rad]")
            if component.gripper_dim is not None:
                if component.gripper_threshold is not None:
                    comparison = ">=" if component.gripper_open_is_high else "<="
                    semantics = f"binary; open when raw {comparison} {component.gripper_threshold:g}"
                else:
                    semantics = (f"raw {component.gripper_closed_value:g} = closed, "
                                 f"raw {component.gripper_open_value:g} = open; "
                                 "levels are determined by the loaded profile")
                lines.append(f"  dim {component.gripper_dim}: {component.name} gripper command ({semantics})")
            else:
                lines.append(
                    f"  {component.name}: no gripper channel; all listed columns are motion"
                )
        if self.input == "absolute":
            lines.append(
                "  encode/decode require the same current_state (B,D), immediately before row 0"
            )
        else:
            lines.append(
                "  encode/decode: omit current_state; deltas are per-step increments, not velocities"
            )
        lines.append(
            f"  resampling: {self.source_hz:g} Hz default -> {self.target_hz:g} Hz codec "
            f"clock ({self.interpolation})"
        )
        return "\n".join(lines)

    def canonicalize(
        self,
        values: np.ndarray,
        *,
        source_hz: float | None = None,
        current_state: np.ndarray | None = None,
    ) -> dict[str, list[ActiveTrajectory]]:
        """Convert a batch into local paths for the two shared profiles."""
        batches = self._canonicalize_batch(values, source_hz=source_hz, current_state=current_state)
        return {
            component.name: [
                ActiveTrajectory(row, component.active_dims) for row in batches[component.name]
            ]
            for component in self.components
        }

    def _canonicalize_batch(
        self,
        values: np.ndarray,
        *,
        source_hz: float | None = None,
        current_state: np.ndarray | None = None,
    ) -> dict[str, np.ndarray]:
        array = np.asarray(values, dtype=np.float32)
        if array.ndim != 3 or len(array) < 1 or array.shape[1] < 1:
            raise ValueError("preset input must have non-empty shape [B, T, D]")
        if array.shape[2] != self.input_dim:
            raise ValueError(
                f"preset {self.name!r} expects input_dim={self.input_dim}, got {array.shape[2]}"
            )
        if not np.all(np.isfinite(array)):
            raise ValueError("preset input contains a non-finite value")
        state_array = optional_finite_array(current_state, shape=(len(array), self.input_dim))
        source_hz = self.source_hz if source_hz is None else source_hz
        return canonicalize_batch(self, array, source_hz=source_hz, current_state=state_array)

    def canonicalize_episode(
        self,
        values: np.ndarray,
        *,
        source_hz: float | None = None,
        current_state: np.ndarray | None = None,
    ) -> dict[str, ActiveTrajectory]:
        """Canonicalize one variable-length episode and optional ``(D,)`` anchor."""
        episode = np.asarray(values, dtype=np.float32)
        if episode.ndim != 2 or len(episode) < 1 or episode.shape[1] != self.input_dim:
            raise ValueError(
                f"preset {self.name!r} expects a non-empty [T, {self.input_dim}] episode"
            )
        state = optional_finite_array(current_state, shape=(self.input_dim,))
        canonical = self._canonicalize_batch(
            episode[None],
            source_hz=source_hz,
            current_state=None if state is None else state[None],
        )
        return {
            component.name: ActiveTrajectory(canonical[component.name][0], component.active_dims)
            for component in self.components
        }

    def profile_trajectories(
        self, values: np.ndarray, *, current_state: np.ndarray | None = None
    ) -> dict[str, list[ActiveTrajectory]]:
        """Canonicalize one episode and group its components by joint/EEF profile."""
        canonical = self.canonicalize_episode(values, current_state=current_state)
        output: dict[str, list[ActiveTrajectory]] = {}
        for component in self.components:
            output.setdefault(component.space, []).append(canonical[component.name])
        return output

    def _canonicalize_component(
        self,
        episode: np.ndarray,
        component: ComponentSpec,
        *,
        source_hz: float,
        current_state: np.ndarray | None,
    ) -> ActiveTrajectory:
        values = canonicalize_component_batch(
            self,
            np.asarray(episode, dtype=np.float32)[None],
            component,
            source_hz=source_hz,
            current_state=None
            if current_state is None
            else np.asarray(current_state, dtype=np.float32)[None],
        )
        return ActiveTrajectory(values[0], component.active_dims)


def format_action_spaces(
    presets: Mapping[str, PresetConfig],
    *,
    selected: str | None = None,
    action_dim: int | None = None,
    second_order_boundaries: Mapping[str, Mapping[str, int]] | None = None,
    codec_name: str = "ActionCodec2",
    chunk_horizon_decode: bool = False,
) -> str:
    """Explain registered tensor layouts without inferring semantics from shape."""
    if action_dim is not None:
        action_dim = _integer(action_dim, "action_dim", positive=True)
    if not presets:
        return "No action spaces are registered."
    candidates = sorted(
        (
            preset
            for preset in presets.values()
            if action_dim is None or preset.input_dim == action_dim
        ),
        key=lambda preset: (preset.input_dim, preset.name),
    )
    lines = [
        f"{codec_name} — action layout guide",
        f"Selected action_space: {selected or 'none (choose a layout before encoding)'}",
        "",
        "Input: NumPy arrays or Torch tensors, (T,D) or (B,T,D).",
        "  T = timesteps, D = columns per action, B = batch size; column indices start at 0.",
        "  Any positive T is accepted. A batch needs equal T; encode unequal-length episodes separately.",
        "  decode returns (B,T,D), including B=1 for a single episode; values are quantized, not lossless.",
        "",
        "How to choose:",
        "  joint = robot joint coordinates; EEF = end-effector (tool) position and orientation.",
        "  absolute = target coordinates; delta = per-step motion increments, not velocities.",
        "  Match column order, units, rotation convention, gripper commands and controller fps.",
        "  Shape alone cannot identify physical meaning: layouts sharing D are only candidates.",
        "  Consult your dataset/controller documentation if these meanings are unknown.",
        "  Supply physical units, not normalized [-1,1] features; convert degrees to radians.",
        "  Non-radian joint coordinates need an explicit mapping to the registered canonical units.",
        "  Pass the actual sampling rate as fps=...; defaults and codec rates appear below.",
        (
            "  Rows are independent chunks: decode(rows, horizon=T, fps=...) needs the chunk T."
            if chunk_horizon_decode
            else "  Resampling can change decoded T; token coverage determines the returned length."
        ),
        "",
        f"Registered configurations: {len(candidates)} shown / {len(presets)} total"
        + ("" if action_dim is None else f" (D={action_dim} candidates, not automatic selection)"),
    ]
    if not candidates:
        widths = ", ".join(map(str, sorted({preset.input_dim for preset in presets.values()})))
        lines.extend(
            [
                f"  No registered action space matches D={action_dim}. Available D: {widths}.",
                "  Confirm your column meanings and register a custom layout; do not pad/truncate to force a match.",
            ]
        )
    for preset in candidates:
        name = preset.name
        marker = " [selected]" if name == selected else ""
        description = preset.describe().splitlines()
        lines.extend(["", f"- {description[0]}{marker}"])
        lines.extend(f"  {line.strip()}" for line in description[1:])
        if second_order_boundaries is not None:
            boundaries = [
                f"{component!r}: (B,{width})"
                for component, width in second_order_boundaries.get(name, {}).items()
            ]
            lines.append(
                "  previous_first_order: "
                + (
                    "required in encode/decode, {" + ", ".join(boundaries) + "}; "
                    f"reconstructed increments at {preset.target_hz:g} Hz, B=1 for a single episode"
                    if boundaries
                    else "not required by this artifact for this layout"
                )
            )
    lines.extend(
        [
            "",
            "Next steps:",
            f"  Load:   {codec_name}.from_pretrained(path, action_space='NAME')",
            "  Switch: codec = codec.for_action_space('NAME')",
            "  Filter: codec.print_action_spaces(action_dim=D)",
            "  Config: codec.action_spaces['NAME'].to_dict()  # exact settings for a listed layout",
            "  Custom: codec.register_action_space(preset)  # explicit column definitions, then switch",
            "  Custom layouts can reorder columns; each joint component supports 1..7 coordinates.",
            "  EEF absolute rotations support rotation_vector, euler_xyz or quaternion_xyzw (unit x,y,z,w).",
            "  EEF delta rotations require rotation_vector with the convention shown above.",
            "  For absolute layouts, current_state is (B,D), even when actions are (T,D).",
        ]
    )
    return "\n".join(lines)


def resample_canonical_trajectory(
    trajectory: ActiveTrajectory,
    component: ComponentSpec,
    *,
    source_hz: float,
    target_hz: float,
    target_length: int | None = None,
    method: str = "linear",
) -> ActiveTrajectory:
    """Rate-convert one decoded canonical component without changing slots."""
    sampled = resample_canonical_batch(
        np.asarray(trajectory.values, dtype=np.float32)[None],
        component,
        source_hz=source_hz,
        target_hz=target_hz,
        target_length=target_length,
        method=method,
    )
    return ActiveTrajectory(sampled[0], trajectory.active_dims)


def externalize_canonical_trajectory(
    trajectory: ActiveTrajectory,
    component: ComponentSpec,
    *,
    current_state: np.ndarray | None,
) -> np.ndarray:
    """Return deltas, or absolute values when an external anchor is supplied."""
    return externalize_canonical_batch(
        np.asarray(trajectory.values, dtype=np.float32)[None],
        component,
        current_state=None
        if current_state is None
        else np.asarray(current_state, dtype=np.float32)[None],
    )[0]