Spaces:
Running
Running
File size: 6,673 Bytes
edae745 af470e1 edae745 af470e1 edae745 af470e1 edae745 af470e1 | 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 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | import json
import tempfile
import unittest
from pathlib import Path
import numpy as np
from app.routes.video import (
_apply_tactile_scale,
_compute_tactile_display_range,
_decode_tactile_matrix,
_tactile_scale_factor,
)
class TactileDecodeTests(unittest.TestCase):
def test_interleaved_row_diff_is_decoded_before_hwc_reshape(self):
matrix = np.zeros((250, 250, 3), dtype=np.int16)
matrix[12, 34] = [7, -2, 19]
matrix[12, 35] = [8, -1, 23]
matrix[200, 249] = [-11, 3, 31]
rows = matrix.reshape(250, 750)
diff = np.empty_like(rows)
diff[:, 0] = rows[:, 0]
diff[:, 1:] = rows[:, 1:] - rows[:, :-1]
decoded = _decode_tactile_matrix(diff.reshape(-1).tolist())
self.assertEqual(decoded.shape, (3, 250, 250))
np.testing.assert_array_equal(decoded.transpose(1, 2, 0), matrix)
class TactileScaleTests(unittest.TestCase):
"""量化整数 → mN 的比例因子必须取自元数据,不能硬编码。
新采集端按 ``scale: 100`` 存储(units 仍是 mN);旧数据无该字段且
本身就是 mN。硬编码 100 会把旧数据错误缩小 100 倍。
"""
def _dataset(self, root: Path, scale) -> Path:
(root / "meta").mkdir(parents=True)
(root / "data" / "chunk-000").mkdir(parents=True)
feature = {"dtype": "int16", "shape": [250, 250, 3],
"units": ["mN", "mN", "mN"]}
if scale is not None:
feature["scale"] = scale
(root / "meta" / "info.json").write_text(json.dumps({
"features": {"observation.gripper_left_force_matrix": feature},
}), encoding="utf-8")
return root / "data" / "chunk-000" / "episode_000000.parquet"
def test_reads_declared_scale(self):
with tempfile.TemporaryDirectory() as temp:
path = self._dataset(Path(temp), 100)
self.assertEqual(_tactile_scale_factor(path), 100.0)
def test_defaults_to_one_when_declared_or_missing(self):
with tempfile.TemporaryDirectory() as temp:
path = self._dataset(Path(temp), None)
self.assertEqual(_tactile_scale_factor(path), 1.0)
with tempfile.TemporaryDirectory() as temp:
path = self._dataset(Path(temp), 0)
self.assertEqual(_tactile_scale_factor(path), 1.0)
def test_missing_metadata_is_not_an_error(self):
with tempfile.TemporaryDirectory() as temp:
path = Path(temp) / "data" / "chunk-000" / "episode_000000.parquet"
path.parent.mkdir(parents=True)
self.assertEqual(_tactile_scale_factor(path), 1.0)
def test_old_data_is_left_untouched(self):
"""scale=1 时不得改变数值 —— 旧数据必须保持原样。"""
matrix = np.full((3, 250, 250), 5.0, dtype=np.float32)
scaled = _apply_tactile_scale(matrix, 1.0)
np.testing.assert_array_equal(scaled, matrix)
def test_new_data_is_divided(self):
matrix = np.full((3, 250, 250), 250.0, dtype=np.float32)
scaled = _apply_tactile_scale(matrix, 100.0)
np.testing.assert_allclose(scaled, 2.5, atol=1e-6)
def test_none_matrix_is_passed_through(self):
self.assertIsNone(_apply_tactile_scale(None, 100.0))
def _write_episode(path: Path, matrices) -> Path:
"""Write an episode whose row-diff column decodes back to ``matrices``."""
import pyarrow as pa
import pyarrow.parquet as pq
flattened = []
for matrix in matrices:
rows = np.asarray(matrix, dtype=np.int16).reshape(250, 750)
diff = np.empty_like(rows)
diff[:, 0] = rows[:, 0]
diff[:, 1:] = rows[:, 1:] - rows[:, :-1]
flattened.append(diff.reshape(-1))
table = pa.table({
"frame_index": pa.array(range(len(matrices)), type=pa.int64()),
"observation.gripper_left_force_matrix":
pa.array(flattened, type=pa.list_(pa.int16())),
})
path.parent.mkdir(parents=True, exist_ok=True)
pq.write_table(table, str(path))
return path
def _episode_frames(quiet=9, contact=80):
"""A quiet episode (noise only) with one frame carrying real contact."""
rng = np.random.default_rng(20260911)
matrices = []
for _ in range(quiet):
frame = np.zeros((250, 250, 3), dtype=np.int16)
frame[:, :, 2] = rng.integers(0, 3, (250, 250))
matrices.append(frame)
loaded = np.zeros((250, 250, 3), dtype=np.int16)
loaded[100:130, 100:130, 2] = contact
matrices.append(loaded)
return matrices
class TactileDisplayRangeTests(unittest.TestCase):
"""色阶必须取自整集,不能逐帧自比。
没有接触的帧里唯一超过噪声门限的就是噪声本身;逐帧归一化会把这份噪声
当成满量程,每个噪点都被拉成"接触点"。整集取色阶后安静帧保持背景色。
"""
def _episode(self, temp: str):
path = _write_episode(
Path(temp) / "data" / "chunk-000" / "episode_000000.parquet",
_episode_frames())
return path
def test_range_is_driven_by_the_contact_not_the_noise(self):
with tempfile.TemporaryDirectory() as temp:
path = self._episode(temp)
lo, hi = _compute_tactile_display_range(
path, "observation.gripper_left_force_matrix")
self.assertGreaterEqual(hi, 80.0)
self.assertLess(lo, hi)
def test_quiet_frame_stays_background(self):
"""整集色阶下,无接触的帧几乎全黑;旧逐帧逻辑在该帧给出 174 级亮度。"""
from app.routes.video import _tactile_range_payload
with tempfile.TemporaryDirectory() as temp:
path = self._episode(temp)
payload = _tactile_range_payload(path, 0, 10, "left")
self.assertIsNotNone(payload)
pixels = 250 * 250
blocks = np.frombuffer(payload[:10 * pixels], dtype=np.uint8)
quiet = blocks[:9 * pixels].reshape(9, pixels)
loud = blocks[9 * pixels:].reshape(pixels)
self.assertLess(float((quiet >= 128).mean()), 0.01)
self.assertGreater(int(loud.max()), 150)
def test_flat_episode_is_drawable(self):
with tempfile.TemporaryDirectory() as temp:
blank = np.zeros((250, 250, 3), dtype=np.int16)
path = _write_episode(
Path(temp) / "data" / "chunk-000" / "episode_000000.parquet",
[blank, blank])
lo, hi = _compute_tactile_display_range(
path, "observation.gripper_left_force_matrix")
self.assertLess(lo, hi)
if __name__ == "__main__":
unittest.main()
|