2026 年，工业 AI 的主线正在从“做出一个演示”转向“接入真实生产并持续运行”。工业企业应用大模型和智能体的比例从 2024 年的 9.6% 上升到 2025 年的 47.5%；上海提出，到“十五五”末，规上企业智能体应用普及率超过 80%。但模型能力并不是唯一瓶颈：设备数据散落在多套系统中，IT 与 OT 使用不同协议、标识和语义，AI 在调用数据之前，必须先解决连接、标准化和治理问题。
这也是从一份 PR 参与 open supOS 相关赛题的现实意义：贡献者不一定要训练新模型，可以从一个设备适配器、一份数据模型、一个异常处理机制或一组可复现测试入手，把工业数据变成智能体能够稳定消费的上下文。
真正难的不是“读到数据”，而是解释数据
工厂里的同一个测点，可能同时出现在 PLC、SCADA、MES、时序数据库和 Excel 台账中。它们记录的可能都是温度，但字段名、单位、时间精度和设备编号并不一致：
- PLC 使用
T_101，MES 使用reactor_1_temperature。 - 一个系统返回摄氏度，另一个系统保存华氏度。
- 设备侧使用本地时间，平台侧要求 UTC。
- 通信中断时，旧值可能仍然被重复上报。
0可能是真实读数，也可能代表传感器故障。
因此，一个有价值的工业开源 PR，不能只证明“接口能通”。它至少要明确四件事：数据来自哪里、字段代表什么、质量是否可信、失败后如何恢复。
可以把接入链路拆成三个边界清晰的部分：
- 采集适配层：连接 Modbus、OPC UA、MQTT、数据库或文件系统。
- 语义归一层：统一设备标识、时间戳、单位、数据类型和质量码。
- 平台写入层：通过目标平台实际支持的接口提交数据，并实现重试、限流与幂等。
这种拆分也方便评审 PR。协议处理、业务映射和平台接口不会混在一个函数里，测试失败时更容易定位责任边界。
先定义最小数据契约
在编写连接器之前，可以先提交一份可讨论的数据契约。下面的 YAML 是一个可改造的最小示例，并不代表 open supOS 的固定接口；实际接入时，需要根据赛题仓库提供的模型和 API 调整 target 部分。
version: 1
source:
type: mqtt
topic: factory/line-1/reactor-1/telemetry
device:
source_id: PLC-A-001
canonical_id: line-1.reactor-1
fields:
T_101:
target: temperature
source_unit: F
target_unit: Cel
type: float
P_101:
target: pressure
source_unit: kPa
target_unit: kPa
type: float
quality:
accepted: [good]
stale_after_seconds: 30
timestamp:
source_field: ts
input_timezone: Asia/Shanghai
output_timezone: UTC
这份契约把容易隐藏在代码里的决定显式化了。评审者可以直接检查设备标识是否稳定、单位转换是否正确、陈旧数据如何判断，而不必先阅读整个连接器实现。
一个可运行的归一化适配器
下面的 Python 示例模拟从设备侧收到一条消息，再输出平台侧可消费的标准事件。它只依赖 Python 标准库，可以直接运行。示例假设输入时间带有明确时区；真实项目还应按照数据契约处理质量码、断线重连和批量写入。
#!/usr/bin/env python3
import json
from datetime import datetime, timezone
DEVICE_MAP = {
"PLC-A-001": "line-1.reactor-1",
}
def fahrenheit_to_celsius(value: float) -> float:
return round((value - 32.0) * 5.0 / 9.0, 3)
def normalize(message: dict) -> dict:
if message.get("quality") != "good":
raise ValueError("reject telemetry with non-good quality")
source_id = message["device_id"]
if source_id not in DEVICE_MAP:
raise ValueError(f"unknown device: {source_id}")
observed_at = datetime.fromisoformat(message["ts"])
if observed_at.tzinfo is None:
raise ValueError("timestamp must include a timezone offset")
return {
"device_id": DEVICE_MAP[source_id],
"observed_at": observed_at.astimezone(timezone.utc).isoformat(),
"measurements": {
"temperature": {
"value": fahrenheit_to_celsius(float(message["T_101"])),
"unit": "Cel",
},
"pressure": {
"value": float(message["P_101"]),
"unit": "kPa",
},
},
"quality": "good",
"source": "mqtt",
}
if __name__ == "__main__":
raw = {
"device_id": "PLC-A-001",
"ts": "2026-03-18T14:30:00+08:00",
"T_101": 176.0,
"P_101": 238.4,
"quality": "good",
}
print(json.dumps(normalize(raw), ensure_ascii=False, indent=2))
将代码保存为 adapter.py 后可运行：
python3 adapter.py
输出中的温度应为 80.0 Cel，时间则转换为 UTC。接入真实平台时，可以把 normalize() 的结果交给 HTTP、MQTT 或平台 SDK 客户端，但要避免把认证信息写入仓库。令牌和端点应通过环境变量注入，例如：
export PLATFORM_ENDPOINT='https://platform.example/api/telemetry'
export PLATFORM_TOKEN='replace-with-a-test-token'
python3 adapter.py
这里的地址只是集成占位符，不代表 open supOS 的真实端点。提交 PR 前，应以赛题仓库中的接口说明和测试环境为准。
让 PR 能被验证，而不只是被阅读
工业连接器经常依赖现场设备，维护者很难复现贡献者的环境。因此，PR 最好附带脱敏样例、明确的运行命令和自动化测试。至少覆盖以下情况：
- 正常数据能够完成字段、单位和时区转换。
- 未知设备不会被静默写入错误资产。
- 缺少时区的时间戳被拒绝或按明确规则补全。
bad、uncertain等质量状态不会伪装成正常读数。- 重试不会为同一条遥测数据制造重复记录。
- 日志不会输出令牌、密码或敏感生产数据。
还应控制 PR 的范围。一个 PR 同时加入新协议、重构公共模块并修改数据模型，会显著增加评审成本。更稳妥的拆法是先提交数据契约和样例，再提交适配器，最后补充平台集成与性能优化。每一步都应保持可运行，并说明兼容性影响。
从“接上”走向“可长期运行”
工业 AI 规模化依赖的不是一次成功调用，而是一条可以审计、恢复和演进的数据链路。选择赛题切入点时，可以优先检查：是否解决了真实的 IT/OT 翻译问题，数据语义是否明确，异常行为是否可观察，测试是否能在没有现场设备的环境中运行。
一份范围清楚、测试完整的 PR，可能只增加一个映射器或修复一个时间戳问题，却能直接提升上层智能体获得上下文的可靠性。对工业开源项目而言，这类基础工作往往比再包装一个模型演示更接近规模化落地。