Metadata-Version: 2.5
Name: serialforge
Version: 1.9.8
Summary: Qt Core serial communication toolkit for Windows
License: MIT
License-File: LICENSE
Classifier: Operating System :: Microsoft :: Windows :: Windows 10
Classifier: Operating System :: Microsoft :: Windows :: Windows 11
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Terminals :: Serial
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: loguru>=0.7
Requires-Dist: pyserial>=3.5
Provides-Extra: pyqt5
Requires-Dist: pyqt5>=5.15.11; extra == 'pyqt5'
Provides-Extra: pyqt6
Requires-Dist: pyqt6>=6.11; extra == 'pyqt6'
Provides-Extra: pyside6
Requires-Dist: pyside6-essentials>=6.11.2; extra == 'pyside6'
Description-Content-Type: text/markdown

# SerialForge

> 基线 commit: `fd434c7`（2026-11-02）

> 写给那些只想在 Qt 界面上点个按钮读写串口，却被多线程、分包断句、乱码和界面卡死折磨到怀疑人生的开发者。

如果你正在和 PySerial 纠缠，每天都在手动解析字节流、处理各种莫名其妙的缓冲区溢出、或者是为了让 GUI 线程不卡死而掉头发，那么恭喜你，你找对地方了。

开发上位机串口程序，大概是每个写 GUI 的人迟早都要趟一遍的浑水。

如果用原生方案写，你的代码通常会长成这样：开一个 `QThread`，在死循环里用 `pyserial.read()` 啃字节；因为串口本质上是流水，一口气读出来的 50 个字节可能前半截是上一条回复的尾巴，中间混着设备被踢掉线时的乱码，后半截才是刚刚发出的回包；你得手写状态机拼接 `\r\n`，再把文本拆开转成浮点数；如果设备还喜欢复读你的命令（回显），或者冷不丁主动报个警，你还得区分哪个是回复、哪个是上报；最后还要通过自定义的 Qt Signal 辛辛苦苦把数据丢回主线程更新界面。

一旦有人拔掉 USB 线，或者遇上一个打不开的虚拟蓝牙串口，整个程序直接在 `open()` 里面假死，Windows 标题栏挂上灰色的“未响应”。

**SerialForge 就是为了消灭这几百行脏代码而生的。**

你只管声明业务逻辑：“这种设备发什么文本、按什么正则匹配、帮我转成什么类型”。至于底层的后台 I/O 线程、断句切帧、排队限流、超时兜底、断线重连，以及把解析好的数据直接送到 Qt 主线程闭包，全部由库包办。

为什么你需要 SerialForge？

* 告别“手动拼接字节”的原始人时代：你还在手动计算校验和？还在担心包头包尾？把那些苦力活交给我们。我们用 Protocol
    模型直接把串口变成对象，把你的思维从“Raw Bytes”中解放出来。
* 真正的异步，不卡 GUI：写 Qt
    应用时还要为串口阻塞而发愁？我们的异步架构原生支持，让你的界面即使在发送庞大数据包时，依然丝滑顺畅。
* 强类型守护：你定义什么数据结构，我们就吐出什么数据结构。IDE 的自动补全不再是摆设，这是真正的类型安全。
* 集中管理业务协议：型号保存串口配置和命令声明，解析结果通过回调或声明信号交付，界面不需要自己处理串口线程和字节分帧。

```powershell
uv add serialforge              # 项目里已经有 Qt（PySide6 / PyQt6 / PyQt5），装本体就够
uv add "serialforge[pyside6]"   # 环境里一个 Qt 都没有？让它顺手带一份，也可换 pyqt6 / pyqt5
```

---

## 10 秒钟看懂它是怎么工作的

假设你的设备协议很简单：发 `PING`，回 `PONG`。

```python
from serialforge import SerialForge, Spec

forge = SerialForge(application="core")   # 纯脚本托管应用；GUI 里先建自己的 QApplication，再调 SerialForge()

PING = Spec("PING", r"PONG")              # 发什么、回什么算对
model = forge.add_device("controller", baudrates=[115200], probe=PING)
model.add(PING)                           # 这条声明算进这个型号的命令表

devices = forge.scan_sync()               # 谁回 PONG，谁就是 controller
if devices:
    forge.connect_sync(devices)           # 非空列表自动取第一台
    print(forge.send_sync(PING).data)     # ['PONG']
forge.disconnect()
```

一台机器上往往不止一种设备：泵、传感器、控制器，探针和命令各不相同。所以 SerialForge 的入口只有一句话——**先把型号报备清楚，再让它自己去认门**。`add_device` 返回的就是那个型号对象，命令加在它身上。

型号在第一次被扫描或连接用到之后就会上锁（`sealed=True`），之后再 `add` 命令会直接报 `ConfigError`。想改？重建一个型号再注册，别偷偷改一个正在用的型号，那不是勇敢，那是给半夜的自己挖坑。

---

## 多种型号，各认各的门

两个型号都爱说 `READ?`，但一个回流量、一个回温度。以前这种冲突根本放不进同一个门面，现在它们各住各家：

```python
from serialforge import SerialForge

forge = SerialForge(application="core")

pump = forge.add_device("pump", baudrates=[9600], probe="PUMP?", probe_pattern="PUMP")
pump_read = pump.add("READ?", r"FLOW:(?P<value>\d+)", int)

sensor = forge.add_device("sensor", baudrates=[115200], probe="SENSOR?", probe_pattern="SENSOR")
sensor_read = sensor.add("READ?", r"TEMP:(?P<value>\d+)", int)   # 同名，互不打架
```

扫描时，端口之间并行跑，同一个端口里型号按顺序一个个试，端口只 open 一次，切型号时重设串口参数。每个端口最多交出一条结果，`DeviceInfo.model_name` 会告诉你它是谁。

想省时间就直接指定型号：`forge.scan_sync(model="pump")`；连接也一样，`forge.connect_sync(model="pump")`。发送只认当前激活型号的声明，拿旧型号的 `Spec` 去发会被 `CommandError` 拦下来。

---

## 几个让你省下大把时间的设计

### 1. 正则匹配 + 自动转类型，告别手撕字符串

很多协议长成 `TEMP: 24.8` 这样。以前你得拿正则匹配，取出 group(1)，再用 `float()` 强转，还要套 try-except 防炸。

在 SerialForge 里，支持直接绑定数值或 Python `@dataclass`：

```python
from dataclasses import dataclass

from serialforge import SerialForge

@dataclass
class SensorData:
    channel: int
    voltage: float

forge = SerialForge(application="core")
model = forge.add_device("adc", baudrates=[115200], probe="READ_ADC", probe_pattern=r"CH\d+:[\d.]+V")

# 想要浮点数：正则捕获的那组数字自动转 float
GET_TEMP = model.add("GET_TEMP", r"TEMP:\s*(?P<val>[\d\.]+)", result_type=float)

# 想要强类型结构体：命名捕获组自动映射到 dataclass 字段
READ_ADC = model.add(
    "READ_ADC",
    r"CH(?P<channel>\d+):(?P<voltage>[\d\.]+)V",
    result_type=SensorData,
)
```

### 2. 推荐的异步写法：闭包就地改界面，不卡主线程

在桌面上位机里，最怕的就是点击一个按钮后整个界面直接“未响应”。

传统的异步写法往往极其繁杂：在发命令的地方把请求扔出，在别处定义各种自定义 Qt Signal，或者在总体的接收槽函数里用一大长串 `if-elif` 逐个判断属于哪条指令的回包，还得在类里维护凌乱的字典或全局变量来记住“刚刚点了哪个控件”。一旦业务复杂，改个逻辑要在好几个分散的方法甚至文件之间来回跳跃排查。

**SerialForge 推荐使用“异步发送 + 主线程闭包”的模式**：

```python
from serialforge import Message, SerialForge


class MainWindow:
    def __init__(self) -> None:
        self.forge = SerialForge()
        self.model = self.forge.add_device(
            "controller", baudrates=[115200], probe="GET_TEMP", probe_pattern=r"TEMP:\s*[\d\.]+"
        )
        self.GET_TEMP = self.model.add("GET_TEMP", r"TEMP:\s*(?P<val>[\d\.]+)", result_type=float)

    def on_refresh_clicked(self) -> None:
        """点击刷新按钮：请求发起与界面更新完整收拢在这一个函数内"""
        def on_done(message: Message[float]) -> None:
            # 回调统一排队回 Qt 主线程，闭包里直接操作控件，安全且直观
            self.temp_label.setText(f"{message.data[0]:.1f} ℃")
            self.refresh_btn.setEnabled(True)

        self.refresh_btn.setEnabled(False)
        self.forge.send_async(self.GET_TEMP).add_done_callback(on_done)
```

这种写法的优势非常直观：

- **绝不卡住主线程**：串口 I/O、切帧定界、超时检测都在后台线程，界面操作丝滑。
- **直接在回调里写界面代码**：闭包由 Qt 事件循环调度回主线程，拿到数据直接 `setText`、`setEnabled`，不用自己接跨线程信号。
- **逻辑高内聚**：从“禁用按钮”到“回包后更新界面并恢复按钮”全在一个函数里，闭包还能顺手捕获控件和局部变量，不用在类里堆中间状态。

扫描和连接也有对应的 `scan_async()` / `connect_async()`，它们的成功闭包同样在主线程执行，拿到 `list[DeviceInfo]` 和 `DeviceInfo` 就能直接在界面里挑设备、连设备。

### 3. 急事可以插队（URGENT 优先级）

队列里排了 10 条轮询传感器数据的指令，操作员突然按了红色的“紧急停机”按钮。按传统 FIFO 慢慢排，等轮到停机命令时黄花菜都凉了。

给紧急命令加个高优先级，它立刻插到队列最前头：

```python
from serialforge import ResponseMode, SerialForge
from serialforge.advanced import CommandPriority

forge = SerialForge(application="core")
model = forge.add_device("pump", baudrates=[9600], probe="PUMP?", probe_pattern="PUMP")
STOP = model.add("STOP", mode=ResponseMode.NO_REPLY)

def emergency_stop() -> None:
    # 停机指令加塞到队列最前头，且不等回复
    forge.send_async(STOP, priority=CommandPriority.URGENT)
```

### 4. 知道 USB 身份时，扫描快到飞起

很多人扫描串口很慢，是因为电脑上插了各种蓝牙转串口、虚拟调制解调器，或者驱动稀烂的设备，打开一次就要等一两秒。

如果你知道自己设备的 USB 芯片型号（VID/PID），注册型号时一并告诉它，扫描阶段就直接跳过无关端口：

```python
from serialforge import SerialForge

forge = SerialForge(application="core")
model = forge.add_device(
    "controller",
    baudrates=[115200],
    probe="PING",
    probe_pattern="PONG",
    vid_pid=[(0x1A86, 0x7523)],   # 比如只敲 CH340 的门
)
```

多个型号各写各的 `vid_pid`，扫描时取并集过滤；某个型号没写过滤，就保留全部端口，别指望它帮你猜。扫描时间从 5 秒跌到几十毫秒，通常就是这个开关的功劳。

### 5. 失败不调成功回调

许多库喜欢不管成功失败都调同一个回调，让用户在回调里写一堆 `if result.is_error(): return`。

`add_done_callback` 只在命令成功、且主线程交付时仍未到本次发送期限时执行。`NO_REPLY` 的成功仅表示写入被驱动接受。如果超时、设备掉线或者通信出错，成功回调不会执行；异常从 `call.result()` 重抛，无所属调用的后台错误从 `check_errors()` 获取。已成功的结果即使因主线程忙碌而错过闭包期限，仍可从调用对象读取。

---

## 示例代码目录

仓库 `examples/` 目录下带了一台内建的虚拟串口设备，不需要任何硬件，敲命令就能跑：

```powershell
# 简单入门
uv run python examples/simple/first_command.py      # 同步阻塞最小脚本
uv run python examples/simple/async_callback.py     # 异步回调最小脚本

# 进阶特性
uv run python examples/advanced/multi_models.py     # 两种型号同场竞技，各自扫各自的
uv run python examples/advanced/async_callbacks.py  # 生产推荐写法：异步发现 + 闭包处理 + STREAM
uv run python examples/advanced/auto_connect.py     # 扫描模式与三种连接姿势
uv run python examples/advanced/quick_start.py      # 单发单收、多收、流式、上报、无回复五种模式
uv run python examples/advanced/interactive.py      # 带命令行界面的交互控制台
```

---

## 纯洁与边界

- **只依赖 QtCore**：绝不导入 `QtWidgets` 或 `QtGui`。你用它写无界面的后台服务、控制台小工具，还是搭配庞大的桌面 UI，全凭你心意。
- **导入无副作用**：`import serialforge` 不会擅自起线程、不会开串口、不会动文件系统，更不会弹窗。
- **支持三套绑定**：PySide6、PyQt6、PyQt5 都可以。如果同时装了多个，设置环境变量 `$env:QT_API="pyside6"` 指定即可。
- **环境要求**：Windows 10/11 x64，Python 3.10–3.14（64 位）。
- **软件测试不等于硬件验收**：仓库里的假后端能覆盖逻辑，但真机矩阵仍在排队，别拿 CI 绿了当插上就通。

---

## 从旧写法迁移

门面身上的 `configure` / `add` / `remove` / `set_init_sequence` 已经删了，职责整体搬去了型号。对号入座即可：

| 原入口（已移除） | 新入口 |
| --- | --- |
| `forge.add/remove` | `model.add/remove` |
| `forge.set_init_sequence` | `model.set_init_sequence` |
| `forge.configure`、构造 `profile` | `forge.add_device` 或 `DeviceModel` 构造字段 |
| 扫描时传波特率、探针或 profile | 注册型号时保存，扫描只传 `model` / `mode` / `port` |

`SerialHandler`、`DeviceFinder(profile)`、`DeviceProfile` 这些过渡期名字还留着，行为和以前一致，只是不接入型号机制。

---

## 完整文档

更详细的指南在在线文档站点：<https://271374667.github.io/serialforge/>

- [新手入门 · 快速开始](docs/beginner/getting-started.md)：手把手搭建第一个工程
- [进阶指南 · 扫描与连接](docs/advanced/scan-connect.md)：多型号配对、预算与三种连接姿势
- [进阶指南 · 声明与响应模式](docs/advanced/declarations.md)：单发多收、流式持续采集与主动上报
- [进阶指南 · 声明信号、流式与上报](docs/advanced/stream-events.md)：从 Spec 直接订阅解析后的 Message，区分持续信号与单次成功闭包
- [进阶指南 · 定界与编码](docs/advanced/framing.md)：遇到不说“句号”的 Modbus 或二进制设备怎么办
- [附录 · 串口与上位机入门](docs/appendix/serial-primer.md)：刚接手硬件项目的纯软件开发者先看这个
- [附录 · 术语表](docs/appendix/glossary.md)：波特率、CRLF、探针、8N1 的人话速查

## 直接从 Spec 订阅命令回复和主动上报

使用 `spec.received.connect(slot)` 直接订阅某条声明，槽函数收到完整的 `Message`，`data` 仍为列表。同一个 Spec 的全部槽都会收到通知；不同设备/业务使用独立的 Spec 对象。

```python
from serialforge import Message, SerialForge

forge = SerialForge(application="core")
model = forge.add_device(
    "controller", baudrates=[115200], probe="PING", probe_pattern="PONG",
)
alarm = model.add(pattern=r"ALARM:(?P<code>\d+)", result_type=int)

def on_alarm(message: Message[int]) -> None:
    print("报警码:", message.data[0])

alarm.received.connect(on_alarm)
# 取消这个槽的订阅：alarm.received.disconnect(on_alarm)
```

声明要在首次扫描/连接前加入型号；信号可以在型号密封后连接或断开。Spec 可以提前声明；首次访问信号需要已创建 Qt 应用并位于主线程。以上代码完成注册与订阅，实际接收还需要连接设备并运行 Qt 事件循环；GUI 使用宿主应用的事件循环，脚本可运行 `QCoreApplication.instance().exec()`。

- 单条回复成功后通知一次，多条回复按声明完成规则成功结束后通知一次汇总；主动上报和流数据逐帧通知。
- 不同设备即使指令文本相同，也建议分别创建 Spec，互不影响；主动复用同一个对象时，其所有槽都会收到消息。`message.spec` 保持原对象身份。
- 扫描探测、失败、超时、取消、`NO_REPLY` 不触发这个信号；当前连接上的初始化、心跳成功回复会触发。
- 声明信号没有额外的主线程交付期限；已成功消息可稍后交付，但旧会话尚未发射的通知会在断开/重连时作废。订阅保留在 Spec 上，不再需要时显式断开槽。
- 单次 `add_done_callback` 仍受本次发送期限约束；同时注册成功闭包和声明信号时，两条途径独立通知，命令终态共享同一 Message。
- `forge.received` 用于非空原始字节日志；`spec.received` 用于解析后的业务消息。两者可以同时使用，详见[完整规则](docs/advanced/stream-events.md)。
