feat(mtc): .mt 脚本编译器落地 + pybot 扬声器静音语义

mtc 新模块:命令式动作脚本(Python DSL)→ .mt 矩阵编译器,非阻塞时间轴 IR(重叠截断/rest 静止值)+ 10ms 栅格化,产物可由 pybot/bot4cj 直接执行;34 测试全绿(含 pybot parser e2e)。

pybot speaker.py:frequency<=0 改为静音节拍(仅 sleep 保时序,不再钳到 min_f 低鸣),对齐仓颉版 sleep 占位语义;mtc Beep 通道 rest=0,段前/段间/段后回落静音,两侧语义闭环。

修复 test_actpred_compatibility 过时断言(99c1bc1 首行分解为重复微步 19→20 帧时漏改,14→15)。

文档三处同步:根 README 回填 memdb/pybot/actpred 并新增 mtc(模块表+端口一览+章节),MODULES/ARCHITECTURE 同步 mtc,新建 mtc/README.md。
This commit is contained in:
qing committed 2026-08-15 13:54:47 +08:00
1 parent 205a60f039
commit 01126a5786
18 files changed
+1125 -7

No files matched your search

View File
Whitespace-only changes.
+25
View File
@@ -0,0 +1,25 @@
"""pytest 公共配置:mtc 根入 sys.path;pybot parser 独立加载(绕包 __init__)
Windows 开发机无 fcntl:import pybot 会经 robot.py → hd/i2c.py 触发
fcntl 导入失败。e2e 仅需 MatrixParser,而 parser.py 只依赖 logging,
用 spec_from_file_location 独立加载即可(与 pybot tests/conftest.py
的符号链接 fallback 同属 Windows 兼容层)。
"""
import importlib.util
import sys
from pathlib import Path
MTC_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(MTC_ROOT))
PYBOT_PARSER = MTC_ROOT.parent / "pybot" / "pybot" / "parser.py"
def load_pybot_parser():
"""独立加载 pybot/parser.py(不触发 pybot/__init__ 的硬件链)"""
spec = importlib.util.spec_from_file_location(
"pybot_parser_standalone", PYBOT_PARSER)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
+127
View File
@@ -0,0 +1,127 @@
"""DSL 收集测试:硬件声明 / 时间轴指令 / 校验规则"""
import pytest
from mtc.dsl import Collector, CompileError, compile_script, make_namespace
def run(source):
"""exec 脚本并返回 collector"""
c = Collector()
exec(compile(source, "<test>", "exec"), make_namespace(c))
return c
# ---------------- 硬件声明 ----------------
def test_hw_blocks_collected_in_order():
c = run('I2C("i2c-1", "/dev/i2c-1")\n'
'Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Servo("s2", 180, 0, 0, "pca9685-0", 1)\n')
assert [b[0] for b in c.blocks] == ["I2C", "Servo", "Servo"]
assert c.blocks[1][2] == ["s1", "90", "0", "0", "pca9685-0", "0"]
assert c.device_names() == {"i2c-1", "s1", "s2"}
def test_hw_block_arity_error():
with pytest.raises(CompileError, match="需要 6 个参数"):
run('Servo("s1", 90, 0)\n')
def test_register_block_extension():
c = run('register_block("ADS1115", ["name", "I2C", "addr"])\n'
'ADS1115("ads0", "i2c-1", "0x48")\n')
assert c.blocks[0][0] == "ADS1115"
assert c.blocks[0][1] == ["name", "I2C", "addr"]
def test_register_block_requires_name_first():
with pytest.raises(CompileError, match="首列必须是 name"):
run('register_block("Bad", ["addr"])\n')
# ---------------- 时间轴指令 ----------------
def test_action_delay_wait_cursor():
c = run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Servo("s2", 90, 0, 0, "pca9685-0", 1)\n'
'Action("s1", 10, 100)\n'
'Delay(50)\n'
'Action("s2", 20, 50)\n'
'Wait()\n')
assert c.cursor == 100.0 # Wait 推进到最晚完成时刻
assert c.channels["s1.angle"].segments[0].v1 == 10
assert c.channels["s2.angle"].segments[0].t0 == 50
def test_action_relative_accumulates():
c = run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Action("s1", 10, 100)\n'
'Wait()\n'
'Action("s1", 10, 100)\n')
segs = c.channels["s1.angle"].segments
assert (segs[1].v0, segs[1].v1) == (10, 20) # 相对叠加在前段目标值上
def test_setto_absolute():
c = run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Pose("s1", 45)\n'
'SetTo("s1", 90, 100)\n')
seg = c.channels["s1.angle"].segments[0]
assert (seg.v0, seg.v1) == (45, 90)
def test_beep_registers_frequency_channel():
c = run('Speaker("spk", 2, 48000, 0.6, 5000, 20)\n'
'Beep(1000, 50)\n')
assert "spk.frequency" in c.channels
assert c.channels["spk.frequency"].segments[0].v1 == 1000
# ---------------- 校验 ----------------
def test_action_undeclared_device_rejected():
with pytest.raises(CompileError, match="未声明"):
run('Action("s1", 10, 100)\n')
def test_dur_must_be_positive_multiple_of_10():
with pytest.raises(CompileError, match="10 的倍数"):
run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Action("s1", 10, 15)\n')
with pytest.raises(CompileError, match="10 的倍数"):
run('Delay(5)\n')
def test_delay_column_is_reserved():
with pytest.raises(CompileError, match="编译器管理"):
run('Speaker("spk", 2, 48000, 0.6, 5000, 20)\n'
'Action("spk", 10, 100, param="delay")\n')
def test_pose_after_action_rejected():
with pytest.raises(CompileError, match="Action 之前"):
run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'Action("s1", 10, 100)\n'
'Pose("s1", 45)\n')
def test_empty_script_rejected():
with pytest.raises(CompileError, match="没有任何动作"):
compile_script('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n')
def test_python_loop_and_math_work():
"""Python DSL 红利:for 循环与数学运算直接可用"""
c = run('Servo("s1", 90, 0, 0, "pca9685-0", 0)\n'
'for i in range(4):\n'
' Action("s1", 5 if i % 2 == 0 else -5, 20)\n'
' Wait()\n')
segs = c.channels["s1.angle"].segments
assert len(segs) == 4
assert [s.v1 for s in segs] == [5, 0, 5, 0]
def test_syntax_error_wrapped():
with pytest.raises(CompileError, match="SyntaxError"):
compile_script("def broken(:\n")
+64
View File
@@ -0,0 +1,64 @@
"""端到端:mtc 编译产物 → pybot MatrixParser 解析断言
验证闭环:编译产物必须被 pybot 零修改解析(列数一致、Q05 表头
'设备.参数' 校验、栅格行数与数值逐一比对)。
"""
from mtc.dsl import compile_script
from tests.conftest import load_pybot_parser
from tests.test_emit import EXAMPLE
def _parse(mt_text):
parser_mod = load_pybot_parser()
parser = parser_mod.MatrixParser(mt_text)
objects = []
while True:
obj = parser.parse()
if obj is None:
break
objects.append(obj)
return objects
def test_compiled_mt_parses_with_pybot():
objs = _parse(compile_script(EXAMPLE)[0])
names = [o.name for o in objs]
assert names == ["Servo", "Speaker", "Action"]
action = objs[names.index("Action")]
assert action.size == 11 # 表头 + 10 栅格行
assert action.col == 4 # s1/s2 + spk.frequency + spk.delay
for row in action.values: # 列数一致(DoesNotSync 不触发)
assert len(row) == action.col
def test_compiled_action_grid_values():
objs = _parse(compile_script(EXAMPLE)[0])
action = [o for o in objs if o.name == "Action"][0]
assert action.values[0] == ["s1.angle", "s2.angle",
"spk.frequency", "spk.delay"]
grid = action.values[1:]
assert [r[0] for r in grid] == [str(i) for i in range(1, 11)]
assert [r[1] for r in grid] == ["0"] * 5 + ["4", "8", "12", "16", "20"]
assert {r[2] for r in grid} == {"0"} # 无 Beep:frequency 恒 0
assert {r[3] for r in grid} == {"10"}
def test_compiled_mt_roundtrips_through_pybot():
"""编译产物经 pybot 解析→序列化→再解析等价(roundtrip 自检)"""
import io
parser_mod = load_pybot_parser()
objs = _parse(compile_script(EXAMPLE)[0])
out = io.StringIO()
parser_mod.MatrixSaver(out).save(objs)
objs2 = _parse(out.getvalue())
assert [(a.name, a.values) for a in objs] == \
[(b.name, b.values) for b in objs2]
def test_q05_header_contract_holds():
"""@Action 全部列名恰含一个 '.'(pybot robot._validate_action 契约)"""
objs = _parse(compile_script(EXAMPLE)[0])
action = [o for o in objs if o.name == "Action"][0]
for col in action.values[0]:
assert col.count(".") == 1
+97
View File
@@ -0,0 +1,97 @@
"""emit 测试:数值格式化 / .mt 文本结构 / 用户设计示例逐字比对"""
from mtc.dsl import compile_script
from mtc.emit import fmt_num
# 用户设计示例(修正后):非阻塞时间轴,s2 第 6 行起步
EXAMPLE = '''
Servo("s1", 180, 0, 0, "pca9685-0", 0)
Servo("s2", 180, 0, 0, "pca9685-0", 1)
Speaker("spk", 2, 48000, 0.6, 5000, 20)
Action("s1", 10, 100)
Delay(50)
Action("s2", 20, 50)
'''
EXPECTED_MT = """\
@Servo
name max min offset controller channel;
s1 180 0 0 pca9685-0 0;
s2 180 0 0 pca9685-0 1;
@
@Speaker
name channels sample_rate volumn max_f min_f;
spk 2 48000 0.6 5000 20;
@
@Action
s1.angle s2.angle spk.frequency spk.delay;
1 0 0 10;
2 0 0 10;
3 0 0 10;
4 0 0 10;
5 0 0 10;
6 4 0 10;
7 8 0 10;
8 12 0 10;
9 16 0 10;
10 20 0 10;
@
"""
def test_example_matches_user_design_exactly():
text, warnings = compile_script(EXAMPLE)
assert warnings == []
assert text == EXPECTED_MT
def test_fmt_num_integer_and_decimal():
assert fmt_num(10.0) == "10"
assert fmt_num(0) == "0"
assert fmt_num(-3.0) == "-3"
assert fmt_num(1.666666) == "1.667"
assert fmt_num(2.500) == "2.5"
def test_non_integer_step_values():
"""10°/6 步:栅格值保留 3 位小数(pybot parse_mt float 可解析)"""
text, _ = compile_script(
'Servo("s1", 180, 0, 0, "pca9685-0", 0)\n'
'Speaker("spk", 2, 48000, 0.6, 5000, 20)\n'
'Action("s1", 10, 60)\n')
rows = [l for l in text.splitlines() if l.endswith("10;")]
assert len(rows) == 6
assert rows[0].startswith("1.667 ")
assert rows[-1].startswith("10 ")
def test_no_speaker_no_delay_column_and_warns():
text, warnings = compile_script(
'Servo("s1", 180, 0, 0, "pca9685-0", 0)\n'
'Action("s1", 10, 20)\n')
assert any("无 Speaker" in w for w in warnings)
assert "s1.angle;" in text # 表头无 frequency/delay 列
assert "frequency" not in text
def test_beep_gap_falls_back_to_zero_silence():
text, _ = compile_script(
'Speaker("spk", 2, 48000, 0.6, 5000, 20)\n'
'Beep(1000, 30)\n'
'Delay(50)\n') # T = max(段尾30, 游标50) = 50 → 5 行
lines = text.splitlines()
assert "spk.frequency spk.delay;" in lines
rows = [l for l in lines if l.endswith(" 10;")]
# 段内 1000;段后回落 0 = 静音(对齐 pybot speaker freq<=0 语义)
assert [r.split()[0] for r in rows] == ["1000", "1000", "1000", "0", "0"]
def test_pure_delay_script_fills_freq_zero_column():
"""纯延时:自动补 spk.frequency 恒 0 列(SpeakerVar.play 两键契约)"""
text, _ = compile_script(
'Speaker("spk", 2, 48000, 0.6, 5000, 20)\n'
'Delay(50)\n')
assert "spk.frequency spk.delay;" in text
assert text.count("0 10;") == 5
+91
View File
@@ -0,0 +1,91 @@
"""时间轴 IR 测试:Segment 插值 / Channel 重叠截断 / 采样保持语义"""
import pytest
from mtc.timeline import Channel, Segment, total_ms
# ---------------- Segment ----------------
def test_segment_value_at_endpoints_and_midpoint():
s = Segment(t0=50, dur=100, v0=10, v1=30)
assert s.end == 150
assert s.value_at(50) == 10 # 段起点取 v0
assert s.value_at(100) == 20 # 中点线性
assert s.value_at(150) == 30 # 段尾取 v1
assert s.value_at(0) == 10 # 段前保持 v0
assert s.value_at(999) == 30 # 段后保持 v1
# ---------------- Channel ----------------
def test_channel_target_defaults_to_initial():
ch = Channel("s1.angle")
assert ch.target == 0.0 and ch.busy_until == 0.0
ch.set_initial(45)
assert ch.target == 45.0
def test_channel_set_initial_after_segment_rejected():
ch = Channel("s1.angle")
ch.add(0, 100, 0, 10)
with pytest.raises(ValueError):
ch.set_initial(45)
def test_channel_overlap_truncates_prev_and_warns():
"""重叠:前段截断至新段 t0,基准取前段目标值(由调用方传 v0)"""
ch = Channel("s1.angle")
ch.add(0, 100, 0, 10) # [0,100) 0→10
v0 = ch.target # 10(前段目标值)
ch.add(50, 100, v0, v0 + 10) # [50,150) 10→20
assert len(ch.segments) == 2
assert ch.segments[0].dur == 50 # 前段截断至 t0=50
assert ch.segments[0].end == 50
assert ch.overlap_warns and "截断" in ch.overlap_warns[0]
# 截断点连续:前段尾值 == 新段起点值
assert ch.value_at(50) == 10
assert ch.value_at(100) == 15
def test_channel_value_at_no_segment_uses_initial():
ch = Channel("s1.angle")
ch.set_initial(30)
assert ch.value_at(10) == 30
assert ch.value_at(9999) == 30
def test_channel_value_at_holds_last_target_after_all_segments():
ch = Channel("s1.angle")
ch.add(0, 100, 0, 10)
ch.add(200, 100, 10, 5)
assert ch.value_at(150) == 10 # 段间空隙保持前段目标值
assert ch.value_at(100000) == 5 # 全部完结后保持最后目标值
def test_channel_rest_value_falls_back_in_gaps():
"""frequency 型通道(rest=0):段前/段间空隙/段后统一回落静止值"""
ch = Channel("spk.frequency", rest=0.0)
ch.add(20, 30, 1000, 1000) # [20,50) 阶跃段
ch.add(70, 20, 2000, 2000) # [70,90)
assert ch.value_at(10) == 0 # 首段之前
assert ch.value_at(30) == 1000 # 段内
assert ch.value_at(50) == 1000 # 段尾端点仍属段内
assert ch.value_at(60) == 0 # 段间空隙回落 rest
assert ch.value_at(80) == 2000
assert ch.value_at(999) == 0 # 全部段后回落 rest
def test_channel_rest_does_not_affect_in_segment_value():
"""rest 仅作用段外:舵机通道设 rest 时段内插值不受影响"""
ch = Channel("s1.angle", rest=0.0)
ch.add(0, 100, 10, 20)
assert ch.value_at(50) == 15
def test_total_ms_takes_max_of_channels_and_cursor():
a, b = Channel("a"), Channel("b")
a.add(0, 100, 0, 1)
b.add(50, 100, 0, 1)
assert total_ms({"a": a, "b": b}, cursor=0) == 150
assert total_ms({"a": a, "b": b}, cursor=500) == 500