fix(pybot×actpred): 微动作录制时间轴均匀化——首行分解为重复步,delay 列权重归零

pybot run():带 delay 的首行(无前序状态)不再单步执行,分解为
N=delay//10 个重复微步(插值退化恒等),录制数据全 10ms 均匀节拍,
delay 列恒 10ms——行级时长信息已隐式编码在步数中,消除首行 200ms
造成的时间采样不均(frame_len/history_len 等间隔假设被破坏)。

actpred default_column_weights:spk.delay 权重 0.2→0.0(步级恒值列
无信号,与保留位同级);spk.F 保持 0.2。

新录制数据集(952 样本,delay 恒 10)存于 data/pybot_record_v2(不入库)。
This commit is contained in:
qing committed 2026-08-14 21:56:01 +00:00
1 parent 1cf428f384
commit 99c1bc1739
5 files changed
+27 -18

No files matched your search

+4 -3
View File
@@ -85,14 +85,15 @@ def denormalize_action(norm_action, action_dim: int = None,
def default_column_weights(action_dim: int = 18) -> np.ndarray:
"""逐列损失权重默认表:舵机 1.0(安全关键)、spk 0.2(保留梯度)、
保留位 0.0(恒 0 纯噪声,参与只会稀释有效梯度)。
"""逐列损失权重默认表:舵机 1.0(安全关键)、spk.F 0.2(保留梯度)、
spk.delay 0.0(步级数据恒 10ms——微动作分解的隐式节拍,时长信息
已编码在步数中)、保留位 0.0(恒 0 纯噪声,参与只会稀释有效梯度)。
action_dim>18 的扩展列默认 1.0:超出当前 18 列语义布局的通道视为
新执行器,静默给 0 会让其完全不参与训练(P2-a1)。"""
w = np.zeros(action_dim, dtype=np.float32)
w[:min(10, action_dim)] = 1.0
if action_dim > 10:
w[10:min(12, action_dim)] = 0.2
w[10] = 0.2
if action_dim > 18:
w[18:] = 1.0
return w
+5 -4
View File
@@ -2004,13 +2004,14 @@ class TestFixedRangeNorm(unittest.TestCase):
from action_norm import default_column_weights
w18 = default_column_weights(18)
np.testing.assert_allclose(w18[:10], 1.0)
np.testing.assert_allclose(w18[10:12], 0.2)
np.testing.assert_allclose(w18[12:], 0.0)
self.assertAlmostEqual(float(w18[10]), 0.2, places=6) # spk.F 保留梯度
np.testing.assert_allclose(w18[11:], 0.0) # spk.delay 步级恒值 + 保留位
# 小 action_dim:全部落在舵机段 → 全 1.0
np.testing.assert_allclose(default_column_weights(4), 1.0)
# 12 列:舵机 1.0 + spk 0.2
# 12 列:舵机 1.0 + spk.F 0.2,delay 0
w12 = default_column_weights(12)
np.testing.assert_allclose(w12[10:], 0.2)
self.assertAlmostEqual(float(w12[10]), 0.2, places=6)
self.assertEqual(float(w12[11]), 0.0)
# 20 列:超出 18 列语义布局的扩展列默认 1.0(新执行器须参与训练)
w20 = default_column_weights(20)
np.testing.assert_allclose(w20[12:18], 0.0)
+9 -4
View File
@@ -148,11 +148,13 @@ class RobotStruct:
"""逐行执行 @Action 矩阵,带时延的行分解为 10ms 微步
微步规则:
- 首行(无前序状态):不分解,按原逻辑执行
- 后续行 + delay > 0:分解为 N=delay//10 个微步,舵机角度线性插值
- delay > 0:分解为 N=delay//10 个微步;后续行舵机角度线性插值,
首行(无前序状态)以行值重复 N 次(不插值)
- delay 必须是 10 的倍数
- 每步调 speaker.play(freq, 10) 提供声音 + 10ms 时序
- dry_run 下补 sleep(0.01) 维持节拍
- 全数据 10ms 均匀节拍:录制时每步一行、delay 列恒 10ms
(行级时长的信息已隐式编码在步数中)
recorder: 可选 EpisodeRecorder,每步录制 (frame, action)
"""
@@ -182,14 +184,17 @@ class RobotStruct:
vals = self._resolve_row(action, i, len(columns))
delay = int(vals[delay_idx]) if delay_idx is not None else 0
if delay > 0 and prev_vals is not None:
if delay > 0:
if delay % 10 != 0:
raise ValueError(
f"@Action row {i}: delay {delay} 不是 10 的倍数")
n = delay // 10
for k in range(n):
# 首行(无前序状态)不插值:prev=vals 使插值退化恒等,
# 产出 N 个重复步,保持全数据 10ms 均匀节拍
step_vals = self._build_step_vals(
prev_vals, vals, columns, k, n)
prev_vals if prev_vals is not None else vals,
vals, columns, k, n)
if recorder is not None:
recorder.record(
self._grab_frame(camera_var),
+3 -2
View File
@@ -68,8 +68,9 @@ def test_run_stand_action(mock_i2c):
# s9/s10: 90° → offval = int(1.5*4096/20) = 307
assert writes[32:36] == servo_writes(8, 307)
assert writes[36:40] == servo_writes(9, 307)
# 10 舵机 × 4 寄存器
assert len(writes) == 40
# 首行 delay=200 → 分解为 20 个重复微步:10 舵机 × 4 寄存器 × 20 步
assert len(writes) == 800
assert writes[40:44] == servo_writes(0, 204) # 第 2 步与第 1 步相同
robot.deactive()
+6 -5
View File
@@ -201,7 +201,7 @@ def test_action_vector_padding(tmp_path, mock_i2c):
robot.deactive()
actions = np.load(tmp_path / "episode_0000" / "actions.npy")
assert actions.shape == (19, 18) # 1(首行)+ 9×2(微步)= 19
assert actions.shape == (20, 18) # 10 行 × delay=20 → 各 2 微步 = 20
assert np.all(actions[:, 4:] == 0) # 列 4-17 全为零
@@ -216,8 +216,9 @@ def test_placeholder_resolution(tmp_path, mock_i2c):
actions = np.load(tmp_path / "episode_0000" / "actions.npy")
# s1.A: 全 45(首行 45,后续行 '-' 继承 45,插值 45→45 仍为 45)
assert np.all(actions[:, 0] == 45)
# s2.A: 首行 0,后续行插值
expected_s2 = [0, 5, 10, 15, 20, 25, 30, 15, 0, 5, 10, 15, 20, 25, 30, 15, 0, 5, 10]
# s2.A: 首行 0 重复 2 步(不插值),后续行插值
expected_s2 = [0, 0, 5, 10, 15, 20, 25, 30, 15, 0, 5, 10,
15, 20, 25, 30, 15, 0, 5, 10]
assert list(actions[:, 1]) == expected_s2
@@ -270,8 +271,8 @@ def test_micro_action_interpolation(tmp_path, mock_i2c):
robot.deactive()
actions = np.load(tmp_path / "episode_0000" / "actions.npy")
# 首行 s1=0(不分解),row2 s1 从 0→30 分 3 步
expected_s1 = [0, 10, 20, 30]
# 首行 s1=0 重复 3 步(不插值),row2 s1 从 0→30 分 3 步插值
expected_s1 = [0, 0, 0, 10, 20, 30]
assert list(actions[:, 0]) == expected_s1