确定性、微秒级的反射安全层：给任何带动力的机器加一层不经过 AI、不经过任何高层决策的保护——
update() 管「停不停」，scale() 管「给多少」，实测每帧 1.3 µs。
带动力的机器迟早会遇到「已经不对了，但大脑还在想」的那半秒。这半秒里你需要的不是更聪明的判断， 而是一条固定的、快到不用思考的通路。
果蝇身上就有这么一条：视小叶里的 LC4 / LPLC2 神经元群体检测到有东西在迫近，汇聚到 DNp01 巨纤维， 阈值一到直接驱动逃逸肌肉——绕过中央复合体那套慢决策。整条通路固定、几十到几百个神经元、亚十毫秒。
真实工程里这层保护通常长成这样：散在控制循环各处的 if、跟具体机器绑死、写完就没人敢动、
改一个阈值要翻三个文件、出了事也说不清当时是哪条判定触发的。本库把它收成一个对象：
- 规则是数据：
Rule/TaperRule是 frozen dataclass，能 JSON 往返，能塞日志、塞仪表盘、塞经验库 - 判定有证据：每个
Verdict带reason/rule_id/value/threshold/t，事后能逐条复核 - 时间由你传：库内从不读时钟，同一份数据回放两遍结果逐帧完全一致
- 只停不够，还要渐弱：真实系统应该在接近极限时就减小输出，而不是撞线才急停
- 三道防误触发：连续 N 帧、传感器饱和忽略、上电静默期——这三样是真机上最常见的误停来源
本库不含任何特定机器的专有代码。它不知道你在做外骨骼、机械臂还是电动滑板。
# 纯标准库，零依赖
uv add "fly-reflex @ git+https://github.com/OwenZhao9/fly-reflex@v0.1.0"
# 想用脉冲后端（唯一会引入 numpy 的地方）
uv add "fly-reflex[spiking] @ git+https://github.com/OwenZhao9/fly-reflex@v0.1.0"或者 pip install "fly-reflex @ git+https://github.com/OwenZhao9/fly-reflex@v0.1.0"。
Python >= 3.11。
从源码开发：
git clone https://github.com/OwenZhao9/fly-reflex && cd fly-reflex
uv sync --all-extras
uv run pytest && uv run ruff check .from fly_reflex import Reflex, Rule, TaperRule, Action
reflex = Reflex.rules(
# 腰部加速度连续 3 帧超过 2.5 g -> 锁存急停；|值| > 3000 当作传感器饱和，忽略该帧
Rule(id="acc", signal="acc_mag", op=">", threshold=2.5, consecutive=3, ignore_above=3000.0),
# 电机过热 -> 软停，条件消失自动恢复
Rule(id="temp", signal="motor_c", op=">", threshold=70.0, action=Action.SOFT_STOP),
# 关节速度 150 °/s 开始渐弱，260 °/s 输出归零
tapers=[TaperRule(id="speed", signal="joint_dps", soft=150.0, hard=260.0)],
)
reflex.arm(t=0.0) # 武装，并开始算 warmup
while True: # 你的实时环，时间戳由你给
t, s = read_sensors() # s 是 {"ax":..., "ay":..., "az":..., "joint_dps":..., "motor_c":...}
v = reflex.update(t, s) # 停不停
k = reflex.scale(t, s) # 给多少（0.0 ~ 1.0）
if v.action is Action.HARD_STOP:
emergency_stop(v.reason) # 已锁存，只有 reflex.reset() 能解除
elif v.action is Action.SOFT_STOP:
send_torque(0.0)
else:
send_torque(desired_torque * k)开箱即用的预设：
reflex = Reflex.from_preset("wearable") # 穿在人身上
reflex = Reflex.from_preset("bench") # 设备放桌上自检跑起来看看：
uv run python examples/quickstart.py # 上面这段的可执行版
uv run python examples/compare_backends.py # 规则后端 vs 脉冲后端的对照数据
uv run python examples/bench.py # 本机 update() 实测耗时公开符号一共 6 个：Action、Verdict、Rule、TaperRule、Reflex、FlyReflexError。
单位一律写进字段名（acc_g / gyro_dps / tilt_deg / dt_s …），库本身不做任何单位换算——
你传进来什么单位，阈值就按什么单位写。
class Action(str, Enum):
OK = "ok" # 无事
SOFT_STOP = "soft" # 输出归零，条件消失可自动恢复
HARD_STOP = "hard" # 锁存，必须显式 reset() 才解除一帧的判定结果 + 证据。
| 字段 | 类型 | 含义 |
|---|---|---|
action |
Action |
结论 |
reason |
str |
人能读懂的原因，例：加速度 3.2 g > 2.5 g（撞击/摔倒） |
rule_id |
str | None |
是哪条规则给出的结论；OK 时为 None |
value |
float | None |
当时的信号值（单位随 signal） |
threshold |
float | None |
当时比的阈值 |
t |
float |
判定发生的时刻（秒），你传进来的那个 t |
to_dict() -> dict / Verdict.from_dict(d) -> Verdict：JSON 往返，json.dumps 直接能吃。
Rule(id: str,
signal: str, # sensors 的 key，或派生量名
op: Literal[">", "<", "abs>", "abs<"],
threshold: float, # 单位随 signal
action: Action = Action.HARD_STOP,
consecutive: int = 1, # 连续 N 帧满足才触发（防单帧毛刺）
ignore_above: float | None = None, # |值| 超过此值视为传感器饱和，忽略该帧
warmup_s: float = 0.0, # arm() 之后多久内不判定该规则（秒）
message: str = "{id}: {signal}={value:.2f} {op} {threshold}")- 比较一律是严格不等号：
op=">"、threshold=2.0、value=2.0→ 不触发。 ignore_above也是严格的：|value| > ignore_above才算饱和；正好相等按正常值判定。message可用的占位符：{id}{signal}{op}{value}{threshold}。 模板在构造期就试跑一遍，写错立刻ValueError，不会拖到运行期。- 同样有
to_dict()/from_dict()。
TaperRule(id: str,
signal: str,
soft: float, # 超过此值开始线性渐弱
hard: float, # 到达此值输出为 0（必须 > soft）
use_abs: bool = True,
ignore_above: float | None = None)三段式，边界含在「安全」那一侧：v <= soft → 1.0；soft < v < hard → 线性；v >= hard → 0.0。
use_abs=False 时是有符号的，可以写 soft=-50, hard=-10 这种单边渐弱。
同样有 to_dict() / from_dict()。
@classmethod
def rules(cls, *rules: Rule,
tapers: Sequence[TaperRule] = (),
derived: Mapping[str, Callable[[Mapping[str, float]], float]] | None = None) -> Reflex用若干条规则拼一个反射层。derived 里的函数必须是无状态纯函数（只看传进来的这一帧）。
@classmethod
def from_preset(cls, name: str) -> Reflex # "wearable" | "bench"@classmethod
def spiking(cls, *, n_neurons: int = 256, threshold: float = 1.0,
signals=("acc_mag","gyro_mag","tilt_deg","joint_dps"),
ranges=None, tau_m_s=0.02, tau_adapt_s=0.25, expansion_gain=2.0,
refractory_s=0.05, max_dt_s=0.05, ignore_above=3000.0, warmup_s=0.0,
taper_soft=0.35, taper_hard=0.85, action=Action.HARD_STOP,
tapers=(), derived=None, seed: int = 0) -> Reflex脉冲后端，接口与规则后端完全一致。需要 fly-reflex[spiking]。见 §6。
def arm(self, t: float) -> None武装并重置 warmup 计时。不解除锁存（只有 reset() 能解除），会清空连续帧计数。
没调用 arm() 就直接 update() 也能用：第一帧被当作 arm 时刻，并记一条 logging.warning——
宁可 warmup 从这里起算，也不让机器处在完全没保护的状态。
def update(self, t: float, sensors: Mapping[str, float]) -> Verdict判定这一帧。纯计算、无 IO、不阻塞，可以在实时环里调用。
同一帧多条规则命中时取最严重的（HARD_STOP > SOFT_STOP），同严重度取先声明的那条。
已锁存时直接返回当初那条判定（含当初的 t），不再评估规则。
def reset(self) -> None解除锁存，清空连续帧计数和脉冲网络状态。累计统计（stats()）不清零。
@property
def latched(self) -> Verdict | Nonedef scale(self, t: float, sensors: Mapping[str, float]) -> float
def scale_detail(self, t: float, sensors: Mapping[str, float]) -> dict[str, float]返回 0.0 ~ 1.0 的输出缩放系数 = 所有 TaperRule 的最小值；scale_detail 给出每条各自的系数（界面用）。
与 update() 相互独立：SOFT_STOP 不影响 scale()，scale() 也不推进 update() 的连续帧计数。
锁存时恒返回 0.0，没有任何 taper 时恒返回 1.0。这两个方法不改任何状态，调多少次都一样；
t 只为接口对称保留，当前不参与判定。
def stats(self) -> dict累计统计，JSON 可直接序列化。形如：
{
"acc": 1,
"gyro": 0, # 每条规则的触发帧数（契约要求的形状）
"frames": 2891,
"armed": True,
"armed_t": 0.0,
"latched": True,
"latched_rule": "acc",
"soft_stops": 0,
"hard_stops": 1,
"backend": "rules",
"triggers": {...},
"ignored": {...}, # ignored = 因饱和被忽略的帧数
"skipped": {...}, # skipped = 信号取不到、规则没法判定的帧数
"tapers": ["speed"],
}skipped 是发现「传感器 key 拼错了 / 忘了传」的地方——库不会为此报错，但会替你记账。
因此规则 id 不能叫 frames、latched、triggers 这些保留名，构造期会直接 ValueError。
本库的构造期异常基类，同时继承 ValueError，所以 except ValueError 和
except FlyReflexError 都接得住。运行期（update / scale）永远不会抛它。
acc_mag（由 ax,ay,az 求模）、gyro_mag（由 gx,gy,gz 求模）。取值顺序：
sensors 里的同名 key 优先 → 你传的 derived → 内置。也就是说你自己算好了就直接传值，
没传才会去算。缺轴按 0 计。
纯标准库，每帧 1.3 µs，触发点精确等于你写的阈值。
内置预设（阈值来自契约，不含任何特定机器的专有参数）：
| 预设 | acc_mag | gyro_mag | tilt_deg | joint_dps | 公共项 |
|---|---|---|---|---|---|
wearable |
> 2.5 g |
> 300 °/s |
abs> 45° |
abs> 450 °/s |
全部 HARD_STOP、连续 3 帧、ignore_above=3000 |
bench |
> 3.0 g |
> 400 °/s |
（不判定） | abs> 2500 °/s |
同上 |
预设需要你提供的 sensors key：ax/ay/az、gx/gy/gz（用来算 acc_mag / gyro_mag）、
tilt_deg、joint_dps。没传的 key 对应的规则会被静默跳过（并计入 stats()["skipped"]）——
上线前建议看一眼这个计数。
预设只含停机规则，不含渐弱 taper：渐弱阈值跟具体机器强相关，放进通用库就是瞎猜。
左右两侧分开判定（预设里的 joint_dps 是单路，想要每侧独立的饱和忽略就自己写）：
Reflex.rules(
Rule(
id="joint_l", signal="ldps", op="abs>", threshold=450.0, consecutive=3, ignore_above=3000.0
),
Rule(
id="joint_r", signal="rdps", op="abs>", threshold=450.0, consecutive=3, ignore_above=3000.0
),
)reflex = Reflex.spiking(n_neurons=256, threshold=1.0, seed=0) # 需要 fly-reflex[spiking]纯 numpy 的小型 LIF 网络，默认 225 个神经元（感觉层 192 + 汇聚层 32 + DNp01 1）， 输入是调用方实际拥有的信号（IMU、关节角速度），保留果蝇迫近→逃逸通路的结构特征： 群体编码 → 汇聚阈值 → 绕过高层直接触发。
结构受果蝇 LC4 / LPLC2 → DNp01 通路启发，不是连接组的复刻。 本库没有加载 MaleCNS / FlyWire 连接组，也没有用它们的任何权重——我们没有视觉输入，硬套就是造假。 import 时不下载任何数据集，运行期不打任何网络请求。
设计细节、增益标定方法和已知局限：docs/spiking-backend.md。 与规则后端的对照数据：docs/backend-comparison.md（见 §8）。
⚠️ 所有实例都不是线程安全的。 一个实例只在一个线程里用。实现里不加锁——加锁会拖慢热路径， 而热路径正是这个库存在的理由。要多线程就一个线程一个实例，或者自己在外面同步。
- ✅ 可以在实时环里调用：
update()/scale()/scale_detail()/latched——纯计算、无 IO、不阻塞、不读时钟。 ⚠️ stats()会建几个小 dict，别放在最内层循环里每帧调。- ❌ 有状态的派生量由调用方负责：
derived=里的函数必须是无状态纯函数（只看当前这一帧）。 需要时间窗的量（例如「最近 1 秒的平均功率」）请你自己算好，以普通 key 放进sensors传进来。 库内部不维护任何滑动窗口——因为窗口状态会让「回放两遍结果一致」这件事变得不可能保证。
- 构造期参数错误 →
ValueError（具体是FlyReflexError，它就是ValueError的子类）。 - 运行期永不抛异常到控制环：信号缺失、值不是实数、
derived函数自己炸了、message模板格式化失败——一律降级为「这条规则本帧跳过」，记一条logging.warning（同一原因只记一次）， 并在stats()里计数。 - 库代码不
print，只用logging（logger 名fly_reflex）。有测试守着。
- 不碰硬件：不 import
pyserial之类，不打开串口，不访问任何设备。有测试守着。 - 不打网络：不下载数据集、不上报、不做任何 HTTP。有测试守着。
- 不读时钟：时间一律由调用方传入。库里连
import time都没有，有测试守着。 - 不维护滑动窗口 / 不做任何时间积分类派生量（脉冲后端的膜电位除外，那是神经元模型本身）。
- 不做传感器融合、不做姿态解算：
tilt_deg要你自己给，库不从四元数或加计里解算姿态。 - 不做单位换算：传 g 就按 g 比，传 m/s² 就按 m/s² 比。
- 不加锁、不起线程、不开进程。
- 不管通信丢失 / 会话超时 / 数据流频率：这些需要时间窗或 IO 状态，是调用方的事。 真要在本库里表达，把「距上次收到数据的秒数」算好当普通 key 传进来即可。
- 不替你决定拿
Verdict做什么：库只给结论和证据，断电、刹车、归零由你实现。 - 不是功能安全认证件：没有做 IEC 61508 / ISO 13849 这类认证，别把它当唯一的安全措施。 真正的安全系统需要硬件层面的冗余。
uv run pytest：121 个测试（不装 numpy 时 96 通过 / 3 跳过），ruff check + ruff format --check 干净。
CI 在 Python 3.11 / 3.12 / 3.13 上跑，并且额外跑一遍不装 numpy 的纯标准库环境。
覆盖点：每条 op 的边界（严格不等号）· consecutive 计数与中断重置 · ignore_above 饱和忽略
（含「饱和帧打断连续计数」）· warmup_s（含 arm() 重置与跨热身边界不累计）· 锁存与 reset() ·
arm() 不解锁存 · 同严重度/跨严重度的优先级 · 派生量优先级与用户覆盖 · 缺 key / 非实数 /
derived 抛异常都不打断控制环 · stats() 形状与 JSON 可序列化 · 构造期校验（重复 id、保留名、
类型、NaN/Inf、坏模板）· scale() 三段与多条取最小 · 有符号 taper · 锁存时 scale() 归零 ·
scale() 不改状态 · 确定性（两个实例 / 同实例 reset 后回放，逐帧一致）· 预设阈值逐条比对契约 ·
触发时刻精确等于「穿过阈值后第 3 帧」· 脉冲后端全套 + 与规则后端对照 · benchmark ·
工程纪律（库内无时钟、无 print、无锁、无硬件/网络 import、import fly_reflex 不拉 numpy）。
Apple M4 / macOS 26.2 / CPython 3.11.15，单线程：
| 场景 | mean | p50 | p99 |
|---|---|---|---|
update() · wearable 预设（4 条规则） |
1.25 µs | 1.25 µs | 1.67 µs |
scale() · 2 条 taper |
0.32 µs | 0.33 µs | 0.38 µs |
update() · 脉冲后端（225 个神经元） |
11.5 µs | 11.3 µs | 19 µs |
契约要求 update() 在 M 系列上 ≤ 50 µs：规则后端 1.25 µs，约为预算的 1/40；
脉冲后端 11.5 µs 也在预算内。240 Hz 的控制环里，规则后端占用 0.03% 的周期。
p99 偶尔会被 GC / 调度打到 40~100 µs，那是解释器的抖动，不是本库的算法行为。
tests/test_benchmark.py 里的断言阈值放宽到 250 µs，因为 CI runner 比 M 系列慢得多——
它只用来兜住「某次改动把热路径写崩了」，不是性能声明。
输入是合成回放（3360 帧 @ 240 Hz：站立 → 走路 → 1 帧 700 °/s 毛刺 → 5 帧 ±3276.7 编码器饱和 → 60 ms 内加速度冲到 4.5 g 的撞击）：
| 场景 | 后端 | 触发 t (s) | 相对事件延迟 | 事件前误触发 | 误触发率 |
|---|---|---|---|---|---|
| A. 撞击/摔倒（事件 t=12.0） | rules | 12.0417 | +41.7 ms | 0 | 0.0000% |
| A. 撞击/摔倒 | spiking | 12.0333 | +33.3 ms | 0 | 0.0000% |
| B. 慢速倾倒（穿阈 t=2.4292） | rules | 2.4375 | +8.3 ms | 0 | — |
| B. 慢速倾倒 | spiking | 2.1375 | −291.7 ms | 1 | — |
怎么读：
- 快事件脉冲后端早约 8 ms——迫近通道对「快速增长」敏感，不必等绝对值越线。
- 慢事件脉冲后端在阈值前 292 ms 就发放（倾角 38° 时）。算不算误触发取决于你的定义， 但事实是：脉冲后端没有可审计的确定阈值。要「触发点精确等于你写的阈值」就用规则后端。
- 两者在含毛刺和编码器饱和的静息/走路段都是 0 次误触发。
完整表格与多 seed 结果（seed = 0/1/7/42/123 下延迟 29~33 ms、误触发全 0）：docs/backend-comparison.md。
数据来源声明：以上全部是合成输入（examples/synthetic.py，可播种、可复现）。
本仓库没有真机录制数据，README 里也不会假装有。
这条通路和它的工程化，站在下面这些工作的肩膀上：
论文
- arXiv:2602.17997 —— 全脑连接组图模型驱动果蝇全身运动控制。 本库借的是「一条固定小电路直接驱动行为」这个形状，不是它的连接组数据。
- arXiv:2508.16792 —— 果蝇连接组在 Loihi 2 神经形态芯片上的仿真。 本库的脉冲后端只是 numpy，没有做神经形态硬件部署。
实现参考
- JHC56/fly-swing —— LC4 / LPLC2 → DNp01 的 LIF 实现
- 5p00kyy/neuroterrarium —— 51 神经元的巨纤维微电路
- cobanov/flyjump —— 小电路 + 静默对照的方法学标杆； 本库 §8 的「同一段输入下两个后端对照」就是照着这个方法学做的
- cobanov/awesome-fly —— 果蝇连接组生态的索引
再说一遍红线：本库的脉冲后端结构受 LC4 / LPLC2 → DNp01 通路启发，不是连接组的复刻； 没有加载 MaleCNS / FlyWire 连接组；import 时不下载任何数据集。
MIT，见 LICENSE。
契约没说死的地方，这里列出我当时怎么选、为什么这么选。有异议的请提 issue，改动只会走
v0.1.x 补丁号，公开 API 的形状不变。
- 饱和帧（
ignore_above）会打断consecutive计数。 「忽略该帧」既可以理解为「这帧不算数，计数保持」，也可以理解为「这帧不可信，连续中断」。 选了后者——三道机制的目的都是防误触发，而且这与下游现有实现的语义一致。 代价：真实危险事件里如果夹着饱和尖峰，可能凑不满连续 N 帧。 - 没
arm()就update()→ 自动把第一帧当 arm 时刻，并记一条 warning。 另一种做法是「没武装就永远不触发」，但那会让忘记arm()的机器完全没有保护，比误触发危险得多。 - 锁存期间
update()直接返回当初那条Verdict（含当初的t），不重新评估规则、 不推进连续帧计数。逐帧记录时会看到重复的t，那是特意的——它标识的是哪次事件锁的。 arm()不解除锁存，只有reset()能解除。- taper 没有 warmup 概念（
TaperRule里没有这个字段）。需要「热身期不出力」请调用方自己按t判断。 - 信号取不到时：规则本帧跳过（计入
stats()["skipped"]），taper 系数取 1.0（不压输出）。 两者都不会让机器停下来，也都不抛异常——但会在统计里留痕。 - 饱和帧的 taper 系数取 1.0（这条 taper 本帧不出力），与「忽略该帧」一致。
scale()/scale_detail()的t参数当前不参与任何判定，保留它只为接口对称， 以及给未来可能的时变 taper 留位置。stats()把每条规则的触发次数放在顶层（契约里的{rule_id: 触发次数, frames: n, ...}就是这个形状），因此规则 id 不能用保留名，构造期直接报错。触发次数统计的是命中帧数：HARD_STOP因为锁存只会记 1 次，SOFT_STOP会每帧记一次。- 预设里的
joint_dps是单路信号。左右腿分开判定（各自独立的饱和忽略与连续计数）需要自己写 两条规则，§6 给了例子。 tilt_deg不是内置派生量：契约只规定了acc_mag和gyro_mag两个内置派生量， 所以tilt_deg需要调用方自己给（例如max(abs(pitch), abs(roll))），或者通过derived=传一个纯函数进来。Rule/TaperRule也实现了to_dict()/from_dict()：契约第 1 节只在Verdict上标了， 但第 0.2 节要求所有公开数据结构都能 JSON 往返，按后者执行。FlyReflexError同时继承ValueError：契约 0.4 既要求构造期抛ValueError， 又要求每个库有自己的<Lib>Error基类，让它同时是两者是唯一不用新增第二个符号的办法。- 脉冲后端的
scale()读出是无状态的（只看当前帧的群体驱动），这样 「scale()与update()相互独立」这条对两个后端都成立。