File size: 4,977 Bytes
1bfc3f0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from .preprocessing import PreprocessingSpec
MANIFEST_NAME = "manifest.json"
SUPPORTED_FORMAT = 1
_USER_TO_STORAGE = {"2bin": "binary", "11bin": "eleven", "21bin": "twenty"}
_GRIPPER_LIMIT = {"2bin": 0.5, "11bin": 0.05, "21bin": 0.025}
_GRIPPER_BINS = {"2bin": 2, "11bin": 11, "21bin": 21}
_GRIPPER_COMPONENT = "gripper_open_command"
def normalize_gripper_mode(token: str) -> str:
"""Map user-facing and legacy gripper-mode tokens onto ``2bin``/``11bin``/``21bin``."""
token = str(token).strip().lower()
if token in _USER_TO_STORAGE:
return token
legacy = {v: k for k, v in _USER_TO_STORAGE.items()}
if token in legacy:
return legacy[token]
raise ValueError(
f"unknown gripper_mode {token!r}; expected one of "
f"{sorted(_USER_TO_STORAGE)} (legacy tokens accepted)"
)
def read_manifest(package: str | Path) -> dict[str, Any]:
path = Path(package)
manifest_path = path / MANIFEST_NAME
if not manifest_path.is_file():
raise FileNotFoundError(
f"{path} is not a release package: missing {MANIFEST_NAME}; "
"expected a release directory containing manifest.json"
)
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
fmt = manifest.get("format")
if type(fmt) is not int or not 1 <= fmt <= SUPPORTED_FORMAT:
raise ValueError(f"unsupported manifest format {fmt!r} in {manifest_path}")
return manifest
def release_tokenizer(manifest: dict[str, Any]) -> str:
"""Tokenizer family of a release; manifests before ActionCodec2Sparse omit it."""
return manifest.get("tokenizer", "actioncodec2")
def selected_variants(
package: str | Path, *, gripper_mode: str, order: int, tokenizer: str = "actioncodec2"
):
"""Select one vocabulary per available physical profile from a release."""
if type(order) is not int or order not in (1, 2):
raise ValueError("primitive_order must be 1 or 2")
mode = normalize_gripper_mode(gripper_mode)
manifest = read_manifest(package)
found = release_tokenizer(manifest)
if found != tokenizer:
frontend = {"actioncodec2": "ActionCodec2", "actioncodec2_sparse": "ActionCodec2Sparse"}
raise ValueError(
f"{package} is an {found!r} release; load it with {frontend.get(found, found)}"
)
selected = {}
for variant in manifest["variants"]:
if variant["order"] != order or variant["gripper_mode"] != mode:
continue
profile = variant["profile"]
if profile not in {"joint", "eef"} or profile in selected:
raise ValueError(f"invalid or duplicate release profile {profile!r}")
selected[profile] = variant
if not selected:
raise ValueError(f"{package} has no release variants for {mode}/order{order}")
return manifest, mode, selected
def check_gripper_consistency(
entry: Path,
processor,
preprocessing: PreprocessingSpec,
mode: str,
) -> None:
expected_token = _USER_TO_STORAGE[mode]
if preprocessing.gripper_mode != expected_token:
raise ValueError(
f"{entry}: preprocessing.json gripper_mode "
f"{preprocessing.gripper_mode!r} does not match requested {mode!r}"
)
component = processor.config.physical.components.get(_GRIPPER_COMPONENT)
if component is None:
return
limit = component.distortion_limit
limit = float(limit[0]) if isinstance(limit, (list, tuple)) else float(limit)
if abs(limit - _GRIPPER_LIMIT[mode]) > 1e-9:
raise ValueError(
f"{entry}: gripper distortion_limit {limit} does not match "
f"{mode} ({_GRIPPER_LIMIT[mode]})"
)
if processor.bins[-1] != _GRIPPER_BINS[mode]:
raise ValueError(
f"{entry}: last primitive bin count {processor.bins[-1]} "
f"does not match {mode} ({_GRIPPER_BINS[mode]})"
)
def check_release_gates(entry: Path) -> None:
validation_path = entry / "validation.json"
derived_path = entry / "derived_report.json"
if validation_path.is_file():
report = json.loads(validation_path.read_text(encoding="utf-8"))
additive = [x for x in report["additive_within_limit"] if x is not None]
ok = (
report["primitive_zero_exact"]
and report["set_bpe_cell_roundtrip_exact"]
and all(additive)
and report["passed"]
)
if not ok:
raise ValueError(f"{entry}: recorded validation gates failed")
elif derived_path.is_file():
report = json.loads(derived_path.read_text(encoding="utf-8"))
if report["additive_within_limit_segments"] != report["segments"]:
raise ValueError(f"{entry}: derived probe gates failed")
else:
raise FileNotFoundError(f"{entry}: no validation.json or derived_report.json")
|