系统部署架构

三层逻辑架构（Brain / Spine / Body）需要落地到具体的硬件平台和软件运行时上。本页描述 EICPS 在 Prj167（500kV 架空输电线路运检）中的部署方案：什么计算跑在哪里、各层之间用什么通道交互、频率鸿沟如何处理。

Brain 层：语义规划（~1 Hz）

硬件：边缘 GPU 服务器，搭载无人机平台，运行非实时 Linux + Docker。

职责：接收操作员工单，完成任务语义理解；调用 HTN 规划引擎分解任务，生成 ETL 任务图；制定本任务段的 STL 规约集合 $\{\varphi_i\}$，随指令下发给 Spine。Brain 只输出”去哪里”和”满足什么约束”，不接触电机。

关键约束：Brain 运行在非实时 OS，允许偶发延迟。Spine 不等待 Brain——Brain 无响应时，由 ZOH 零阶保持兜底，Spine 继续执行最后一次收到的指令。

Prj167 双引擎 Brain

防振锤更换任务中，Brain 层内部由两个引擎串行工作：

LLM 前处理引擎（~2–5s）：接收自然语言工单，结合领域知识库理解意图，输出 ETL 草稿任务图。处理语义歧义、补全上下文，供操作员审核确认后交给规划引擎。

HTN DFS 规划引擎（≤0.2ms）：以确定性深度优先搜索展开 HTN 任务网络，生成包含约 26 个原子算子的有序任务序列。规划前执行三谱联合剪枝（GeoPruner）：

$\text{算子合法} \iff \Sigma_G \;\wedge\; \Sigma_T \;\wedge\; \Sigma_S$

过滤不满足几何可达性（$\Sigma_G$）、路径连通性（$\Sigma_T$）、电气安全距离（$\Sigma_S$）的算子分支，保证输出的任务序列先验满足全部 STL 前提。

Spine 层：实时控制（~1 kHz）

硬件：ESP 工控机（ARM 或 x86 实时核），运行 RT-Linux，C++ 实时控制循环，1ms 固定周期。

职责：接收 Brain 下发的 STL 规约包和 ETL 任务图，独立运行实时控制与安全监控——EKF 状态估计、CBF-QP 约束修正、STL-RHC 鲁棒度监控、模态调度。Spine 自主运行，不依赖 Brain 干预。

关键约束：1ms 控制周期的时延抖动（jitter）须小于 0.1ms，必须运行在实时 OS 上。任何软件层超时直接触发硬件安全停机。

混合系统调度：H = (C, F, D, G)

Spine 每个控制周期（1ms）执行三步检查：

Zeno 检测防止无限快速切换：相邻两次 Jump 的时间间隔必须 $\geq \tau_{min} = 10\text{ms}$。

收到 ETL 任务包后，Spine 独立运行 STL-RHC 安全监控，三级响应：

ρ > ρ_warn（0.5m） → 继续执行，定时上报
0 < ρ ≤ ρ_warn   → 预警：上报 Brain，降速
ρ ≤ 0             → 强制 FAILSAFE，无需等 Brain 确认

Body 层：物理执行（~10 kHz）

硬件：关节电机驱动单元 + IMU + 六轴力觉传感器，电机固件以 PWM 10kHz 运行，通过 EtherCAT 总线（≤1ms 时延）与 Spine 通信。

职责：执行 Spine 下发的参考力矩 $\tau_{ref}$，采集编码器、IMU、力觉传感器的原始数据并高频回流 Spine。

关键约束：Body 层不包含任何决策逻辑，是纯粹的执行与感知单元。传感器原始数据不经任何软件过滤直接回流，确保 Spine 的 EKF 看到真实物理状态。safety_flag = 0xFF 时，电机驱动单元立即切断力矩输出，不等待上层确认——这是物理层的最后防线。

跨层信息流

三个频率段之间各相差约 10 倍，频率差来自物理约束而非设计选择。系统工程的任务是让三层在不互相阻塞的前提下协同运行。

下行（语义 → 物理）：Brain 生成意图，经接口 A 下发给 Spine。每次下发包含 STL 规约包（本任务段的时序安全约束）和 ETL 任务图（结构化任务序列）。Spine 收到后立即生效，旧指令覆盖。Brain 无响应时，ZOH 维持最后一次指令，Spine 不停机不等待。

上行（物理 → 语义）：Spine 将运行状态压缩后事件触发上报 Brain。触发条件：STL 鲁棒度 $\rho$ 跌破预警阈值、发生模态切换、或距上次上报超过 500ms（心跳）。Spine 不将 1kHz 传感器流原样上报，仅推送结构化状态摘要。

回流（物理 → 数字）：Body 层传感器原始数据经接口 B 以 1kHz 回流 Spine，不经任何软件过滤，确保 EKF 状态估计和 CBF 安全监控看到真实物理状态。

算法实现

import numpy as np
from scipy.integrate import solve_ivp
from dataclasses import dataclass
from enum import Enum
from typing import Callable

class Mode(Enum):
    IDLE     = 0
    CRUISE   = 1
    AVOID    = 2
    GRASP    = 3
    FAILSAFE = 4

@dataclass
class HybridState:
    """Spine 层状态点：连续物理状态 + 离散逻辑模态"""
    x:    np.ndarray
    mode: Mode  = Mode.CRUISE
    t:    float = 0.0

class EICPSScheduler:
    """Spine 层混合系统调度器：H = (C, F, D, G)"""

    def __init__(self,
                 flow_map:   Callable,   # F(x, u, mode) → ẋ
                 jump_map:   Callable,   # G(state) → new_state
                 flow_guard: Callable,   # C(state) → bool
                 jump_guard: Callable,   # D(state) → bool
                 controller: Callable,   # π(state) → u_ref
                 tau_min:    float = 0.01):
        self.F, self.G, self.C, self.D = flow_map, jump_map, flow_guard, jump_guard
        self.pi = controller
        self.tau_min = tau_min
        self._last_jump_t = -np.inf

    def step(self, state: HybridState, dt: float) -> HybridState:
        t = state.t

        # ① 跳集检测（含 Zeno 保护）
        if self.D(state) and (t - self._last_jump_t) >= self.tau_min:
            self._last_jump_t = t
            return self.G(state)

        # ② 流集合法性检查
        if not self.C(state):
            return HybridState(x=state.x, mode=Mode.FAILSAFE, t=t)

        # ③ Flow 演化：CBF 修正后的控制量
        u = self.pi(state)

        def ode(_, x):
            return self.F(x, u, state.mode)

        sol = solve_ivp(ode, [0, dt], state.x, method='RK45', max_step=dt/5)
        return HybridState(x=sol.y[:, -1], mode=state.mode, t=t + dt)


class ZOHScheduler:
    """跨频调度：ZOH 零阶保持 + 事件驱动上报"""

    def __init__(self, report_interval: float = 0.5, rho_warn: float = 0.5):
        self._cmd:          np.ndarray | None = None
        self._last_report:  float = -np.inf
        self.rho_warn       = rho_warn

    def brain_update(self, waypoint: np.ndarray) -> None:
        """Brain 下发新指令（~1Hz）"""
        self._cmd = waypoint

    def spine_get_cmd(self) -> np.ndarray:
        """Spine 读取当前指令（ZOH，1kHz 调用）"""
        return self._cmd if self._cmd is not None else np.zeros(3)

    def should_report(self, t: float, rho: float) -> bool:
        """判断是否触发上报"""
        if rho <= self.rho_warn:
            return True                    # 安全裕量告警
        if t - self._last_report >= 0.5:
            self._last_report = t
            return True                    # 500ms 心跳
        return False

→ 接口协议 A/B（报文格式、STL 封装、Bumpless Transfer）

→ 跨频调度（ZOH、最小驻留时间、Zeno 检测的理论基础）

→ 状态估计（EKF 在 Spine 层的实现细节）