# -*- coding: utf-8 -*-
"""
故障注入控制器 Fault Injector Controller
========================================
平台 : Raspberry Pi Pico + MicroPython
接口 : UART0 (GP0 = TX, GP1 = RX) 9600 8N1；同时兼容 USB-CDC(Thonny / REPL 输入)
被测对象: DUT 的三个引脚 D0 / D1 / D2
故障定义
--------
D0Open / D1Open / D2Open -> 对应引脚断路（DUT 引脚与其下游负载断开）
D1D0 -> D1 与 D0 短路
D1D2 -> D1 与 D2 短路
同一条指令再次下发即解除该故障，恢复原样。
（D0 与 D2 之间不存在任何硬件通路，软件亦强制互斥，永不产生 D0-D2 短路）
串口应答协议
------------
上位机发一条指令 -> Pico 执行后固定回一条应答行：
执行成功 -> "SetSuccess"
指令非法/被禁止 -> "SetFailed"
应答行固定在指令被完整解析后立刻发出，行尾 CRLF，不带任何前缀。
★ 应答行会【同时】发往两个通道，二者内容完全一致：
- USB-CDC : Pico 原生 USB 口（设备管理器里那个 "USB 串行设备"）
- 硬件 UART: UART0, GP0=TX / GP1=RX
因此上位机接哪个口都能用，不必关心指令从哪进来。
（只想用其中一个时，改 REPLY_TO_USB / REPLY_TO_UART 开关）
给人看的日志（上电横幅、Help、Test 进度等）只打到 USB-CDC，
不会混进协议行；上位机按行匹配 "SetSuccess"/"SetFailed" 即可。
硬件信号链（详见 README.md）
---------------------------
Pico GP18 -> 74HCT595 DS (pin14)
Pico GP19 -> 74HCT595 SHCP (pin11)
Pico GP20 -> 74HCT595 STCP (pin12)
74HCT595 MR(pin10)=+5V, OE(pin13)=GND, Q7S(pin9)=NC, VCC(pin16)=+5V
74HCT595 Q0..Q7 -> ULN2803 1B..8B
ULN2803 GND(pin9)=GND, COM(pin12)=+12V, 1C..5C 各经上拉电阻接 +12V
1C(pin20) -> CD4053B C(pin9) D2 断路通道选择
2C(pin19) -> CD4053B B(pin10) D1 断路通道选择
3C(pin18) -> CD4053B A(pin11) D0 断路通道选择
4C(pin17) -> TPW4054 S0(pin7) 短路通道选择低位
5C(pin16) -> TPW4054 S1(pin6) 短路通道选择高位
6C/7C/8C -> 悬空
★★ 关键电平关系（ULN2803 是"反相 + 集电极开路"）
Qn = 1 -> ULN2803 输出拉低(~0.8V) -> 控制端 = 逻辑 0
Qn = 0 -> ULN2803 输出高阻,被上拉到 +12V -> 控制端 = 逻辑 1
即：控制端电平 = NOT(Qn)，程序里已统一用 Qn = 1 - 目标电平 处理。
之所以要 12V：CD4053B 在 VDD=12V 时 VIH(min) 约 8.4V(=0.7*VDD)，
Pico 的 3.3V 或 595 的 5V 都推不动，必须升到 12V；
而被拉高的 12V 电平同时满足 TPW4054(VCC=12V, VIH=2.4V) 的要求。
"""
from machine import Pin, UART
import sys
import time
# ============================================================================
# 1. 用户配置区
# ============================================================================
# ---- CD4053B 的接线方式 -------------------------------------------------
# True : AY/BY/CY (pin13/1/3) 悬空 —— 控制端 = 1 时该引脚断路
# 【与当前现有接线一致】上电瞬间(595 输出未定义)可能落在"全部断路"
# False : AX/BX/CX (pin12/2/5) 悬空 —— 控制端 = 1 时该引脚正常
# 【推荐的改法】上电即为"正常"，失效安全
CD4053_OPEN_WHEN_CTRL_HIGH = True
# ---- 引脚定义 -----------------------------------------------------------
PIN_DS = 18 # 74HCT595 DS (pin14) 串行数据
PIN_SHCP = 19 # 74HCT595 SHCP (pin11) 移位时钟
PIN_STCP = 20 # 74HCT595 STCP (pin12) 锁存时钟
# ---- 串口定义 -----------------------------------------------------------
UART_ID = 0
UART_TX = 0 # Pico GP0 -> 上位机 RX
UART_RX = 1 # Pico GP1 <- 上位机 TX
UART_BAUD = 9600
# ---- 串口应答字符串（上位机按行匹配，如需改动只改这里） ------------------
REPLY_SUCCESS = "SetSuccess" # 指令执行成功
REPLY_FAILED = "SetFailed" # 指令非法 / 被禁止 / 执行失败
REPLY_EOL = "\r\n"
# ---- 调试开关 -----------------------------------------------------------
UART_ECHO_STATUS = False # True: 应答行之后额外把状态文本也发到串口(两个通道都会发)
UART_BOOT_BANNER = False # True: 上电时向两个通道各发一行就绪信息 "FaultInjector ready"
# ---- 应答通道开关（重要）------------------------------------------------
# 指令可能从两个通道进来：硬件串口 UART0(GP0/GP1) 或 USB-CDC(原生 USB 口)。
# 应答行默认【两个通道都发】，这样无论上位机连的是哪个口，都能拿到干净的
# SetSuccess / SetFailed，不必关心指令是从哪进来的。
# 只想走其中一个通道时，把另一个改成 False。
REPLY_TO_USB = True # 应答行发到 USB-CDC（原生 USB 口 / Thonny）
REPLY_TO_UART = True # 应答行发到硬件串口 UART0（GP0=TX, GP1=RX）
# 调试用：额外在 USB 控制台回显 "<< SetSuccess" 形式。仅给人看，默认关。
DEBUG_ECHO_REPLY = False
# ---- TPW4054 通道分配 ---------------------------------------------------
# S1=0,S0=0 -> A0(pin8) : 悬空 = 正常(不短路)
# S1=0,S0=1 -> A1(pin3) : 接 D0 = D1 短路到 D0
# S1=1,S0=0 -> A2(pin1) : 接 D2 = D1 短路到 D2
# S1=1,S0=1 -> A3(pin2) : 悬空 = 备用(同样不短路)
SHORT_NONE = 0 # 无短路
SHORT_D1D0 = 1 # D1-D0
SHORT_D1D2 = 2 # D1-D2
# ============================================================================
# 2. 74HCT595 移位输出
# ============================================================================
class Shift595:
"""74HCT595 驱动。
74HCT595 的移位方向：DS 先进 stage0(即 Q0)，后续位把前面的位往 Q7 推。
因此 8 个时钟后，**最先送出的位落在 Q7**，最后送出的位留在 Q0。
所以按 MSB->LSB 顺序发送时：bit7->Q7, ..., bit0->Q0。
"""
def __init__(self, ds, shcp, stcp):
self.ds = Pin(ds, Pin.OUT, value=0)
self.shcp = Pin(shcp, Pin.OUT, value=0)
self.stcp = Pin(stcp, Pin.OUT, value=0)
def write(self, value):
value &= 0xFF
for i in range(7, -1, -1): # MSB first
self.ds.value((value >> i) & 1)
self.shcp.value(1)
self.shcp.value(0)
# 8 位移完后统一锁存到输出寄存器，避免输出过程中出现中间态
self.stcp.value(1)
self.stcp.value(0)
# ============================================================================
# 3. 故障注入控制器
# ============================================================================
class FaultInjector:
def __init__(self):
self.sr = Shift595(PIN_DS, PIN_SHCP, PIN_STCP)
# 三个引脚的断路状态
self.open = [False, False, False] # 索引 0=D0, 1=D1, 2=D2
# 短路状态（互斥，D0-D2 永不同时接通）
self.short = SHORT_NONE
self.apply(verbose=False)
# ------------------------------------------------------------------
# 3.1 由故障状态推导 74HCT595 输出字节
# ------------------------------------------------------------------
def _ctrl_level(self, is_open):
"""期望加到 CD4053B 控制端的逻辑电平(1 = 12V, 0 = 0V)。"""
if CD4053_OPEN_WHEN_CTRL_HIGH:
return 1 if is_open else 0
return 0 if is_open else 1
def build_byte(self):
b = 0
# ---- CD4053B：三个独立的 2:1 模拟开关，各自控制一个引脚的断路 ----
lvl_A = self._ctrl_level(self.open[0]) # D0 -> CD4053B A (pin11)
lvl_B = self._ctrl_level(self.open[1]) # D1 -> CD4053B B (pin10)
lvl_C = self._ctrl_level(self.open[2]) # D2 -> CD4053B C (pin9)
# ULN2803 反相：控制端电平 = NOT(Qn) => Qn = 1 - 电平
q0 = 1 - lvl_C # Q0 -> 1C -> CD4053B C (D2)
q1 = 1 - lvl_B # Q1 -> 2C -> CD4053B B (D1)
q2 = 1 - lvl_A # Q2 -> 3C -> CD4053B A (D0)
# ---- TPW4054：4 选 1，公共端 A 接 D1，其余通道接 D0 / D2 ----
if self.short == SHORT_D1D0: # A1: D1 <-> D0
s1, s0 = 0, 1
elif self.short == SHORT_D1D2: # A2: D1 <-> D2
s1, s0 = 1, 0
else: # A0: 悬空 = 不短路
s1, s0 = 0, 0
q3 = 1 - s0 # Q3 -> 4C -> TPW4054 S0 (pin7)
q4 = 1 - s1 # Q4 -> 5C -> TPW4054 S1 (pin6)
b |= (q0 & 1) << 0
b |= (q1 & 1) << 1
b |= (q2 & 1) << 2
b |= (q3 & 1) << 3
b |= (q4 & 1) << 4
# Q5/Q6/Q7 -> 6C/7C/8C 未使用，保持 0（ULN 输出高阻，不耗电）
return b
# ------------------------------------------------------------------
# 3.2 下发到硬件
# ------------------------------------------------------------------
def apply(self, verbose=True):
# 安全约束：D0 与 D2 绝不允许被短接（硬件上也无此通路，软件再兜一层）
assert self.short in (SHORT_NONE, SHORT_D1D0, SHORT_D1D2)
b = self.build_byte()
self.sr.write(b)
if verbose:
log(self.status_text())
# ------------------------------------------------------------------
# 3.3 状态文本
# ------------------------------------------------------------------
def status_text(self):
parts = []
for i, name in enumerate(("D0", "D1", "D2")):
parts.append("%sOpen=%s" % (name, "ON" if self.open[i] else "OFF"))
if self.short == SHORT_D1D0:
s = "D1D0=ON"
elif self.short == SHORT_D1D2:
s = "D1D2=ON"
else:
s = "SHORT=OFF"
parts.append(s)
parts.append("byte=0x%02X" % self.build_byte())
return " ".join(parts)
# ------------------------------------------------------------------
# 3.4 指令处理（同指令再发一次 = 解除）
# 返回 (ok, payload)
# ok : True 成功 / False 非法
# payload : 需要一并回给上位机的附加文本，None 表示只有应答行
# ------------------------------------------------------------------
def cmd(self, name):
ok, payload = True, None
if name in ("D0OPEN", "D0"):
self.open[0] = not self.open[0]
elif name in ("D1OPEN", "D1"):
self.open[1] = not self.open[1]
elif name in ("D2OPEN", "D2"):
self.open[2] = not self.open[2]
elif name in ("D1D0", "D1-D0", "SHORTD1D0"):
# 与 D1D2 互斥：TPW4054 是 break-before-make，切换瞬间不会把
# D0-D1-D2 三个网络同时接通，因此不会衍生出 D0-D2 短路
self.short = SHORT_NONE if self.short == SHORT_D1D0 else SHORT_D1D0
elif name in ("D1D2", "D1-D2", "SHORTD1D2"):
self.short = SHORT_NONE if self.short == SHORT_D1D2 else SHORT_D1D2
elif name in ("D2D0", "D0D2"): # 明令禁止：D0 与 D2 不允许短路
log("ERR 禁止 D0-D2 短路，指令已忽略")
return False, None
elif name in ("CLEAR", "CLR", "RESET", "NORMAL", "CLEARALL"):
self.open = [False, False, False]
self.short = SHORT_NONE
elif name in ("STATUS", "STATE", "?"):
payload = self.status_text()
log(payload)
return True, payload
elif name in ("TEST", "DEMO"):
self.run_test()
return True, None
elif name in ("HELP", "H"):
return True, HELP_LINES
elif name == "":
return False, None
else:
log("ERR 未知指令: %s (HELP 查看指令表)" % name)
return False, None
self.apply()
return ok, payload
# ------------------------------------------------------------------
# 3.5 自动巡回测试（台架验证用）
# ------------------------------------------------------------------
def run_test(self):
steps = [
("D0Open", [True, False, False], SHORT_NONE),
("D1Open", [False, True, False], SHORT_NONE),
("D2Open", [False, False, True], SHORT_NONE),
("D1D0", [False, False, False], SHORT_D1D0),
("D1D2", [False, False, False], SHORT_D1D2),
("Normal", [False, False, False], SHORT_NONE),
]
log("TEST 开始自动巡回 ...")
for name, opens, short in steps:
self.open = list(opens)
self.short = short
self.apply(verbose=False)
log("TEST -> %-7s %s" % (name, self.status_text()))
time.sleep_ms(1500)
log("TEST 完成，已回到正常状态")
# ============================================================================
# 4. 输出通道
# log() -> 只打到 USB-CDC，供 Thonny 调试，不占串口
# reply() -> 打到串口，供上位机按行解析
# ============================================================================
_uart = None
HELP_LINES = [
"指令表 (大小写不敏感, 回车结束):",
" D0Open / D1Open / D2Open 对应引脚断路, 再发一次解除",
" D1D0 D1 短路 D0, 再发一次解除",
" D1D2 D1 短路 D2, 再发一次解除",
" Clear 全部恢复正常",
" Status 查询当前状态",
" Test 自动巡回演示各故障",
" Help 本帮助",
"每条有效指令执行成功后回复: " + REPLY_SUCCESS,
]
def log(msg):
"""仅 USB-CDC 控制台输出（给人看的调试信息，不是协议行）。
行尾同样用 CRLF，保证两个通道上"每行都以 CRLF 结束"这一条不变量成立，
上位机用 readline() 按行读时不会踩到混用的行尾。
"""
try:
sys.stdout.write("%s\r\n" % (msg,))
except Exception:
pass
def reply(msg):
"""协议应答行 -> 同时发往 USB-CDC 与硬件 UART。
行尾固定 CRLF，且不加任何前缀 —— 上位机按行取到的就是一整条
"SetSuccess" / "SetFailed"。
"""
line = msg + REPLY_EOL
if REPLY_TO_USB:
try:
sys.stdout.write(line)
except Exception:
pass
if REPLY_TO_UART and _uart is not None:
try:
_uart.write(line.encode())
except Exception:
pass
if DEBUG_ECHO_REPLY:
log("<< " + msg)
def reply_lines(payload):
"""应答行 + 附加信息（可能是字符串或字符串列表）。"""
if payload is None:
return
if isinstance(payload, (list, tuple)):
for line in payload:
reply(line)
else:
reply(payload)
def dispatch(fi, name):
"""解析出的一条指令 -> 执行 -> 回显应答行。返回 True/False。"""
ok, payload = fi.cmd(name)
reply(REPLY_SUCCESS if ok else REPLY_FAILED)
if ok and payload is None and UART_ECHO_STATUS:
payload = fi.status_text()
reply_lines(payload)
return ok
# ============================================================================
# 5. 主程序
# ============================================================================
def main():
global _uart
# 5.1 上电第一件事：立刻把 74HCT595 输出刷成"正常"状态
fi = FaultInjector()
# 5.2 打开串口
try:
_uart = UART(UART_ID, baudrate=UART_BAUD, tx=Pin(UART_TX), rx=Pin(UART_RX))
except Exception as e:
_uart = None
log("UART 初始化失败: %s" % e)
log("")
log("=== Fault Injector for D0/D1/D2 ready (9600 8N1) ===")
log("初始状态: " + fi.status_text())
for line in HELP_LINES:
log(line)
if UART_BOOT_BANNER and _uart is not None:
reply("FaultInjector ready")
# 5.3 USB-CDC 输入（可选，方便在 Thonny 里直接敲指令）
poller = None
try:
import select
poller = select.poll()
poller.register(sys.stdin, select.POLLIN)
except Exception:
poller = None
rxbuf = b""
# 5.4 主循环：非阻塞收指令
while True:
chunk = b""
if _uart is not None and _uart.any():
chunk += _uart.read() or b""
if poller is not None:
try:
if poller.poll(0):
c = sys.stdin.read(1) or ""
if isinstance(c, str):
c = c.encode()
chunk += c
except Exception:
pass
if chunk:
rxbuf += chunk
while True:
idx = -1
for sep in (b"\n", b"\r"):
k = rxbuf.find(sep)
if k != -1 and (idx == -1 or k < idx):
idx = k
if idx == -1:
break
line = rxbuf[:idx]
rxbuf = rxbuf[idx + 1:]
name = line.decode("utf-8", "ignore").strip()
name = name.replace(" ", "").replace("\t", "").upper()
if not name:
continue
dispatch(fi, name)
if len(rxbuf) > 64: # 防止一直不给换行导致缓冲膨胀
rxbuf = rxbuf[-64:]
time.sleep_ms(5)
if __name__ == "__main__":
main()
FPS: 0
Power: 0.00W
Power: 0.00W