上海“54321“智能制造目标下，基础级智能工厂的“核心数据实时采集“要采什么、怎么采（附架构与 Python 实现）

摘要：《上海市加快发展新一代智能制造行动方案》提出到 2028 年基础级智能工厂累计达 1600 家，并把“实现核心数据实时采集”写进了基础级的准入要求。这篇文章从工程角度拆解三件事：核心数据到底分哪三层、设备数据为什么可以不必先过 MES，以及一条从 PLC / CNC 到时序数据库的最小可用采集链路（含可运行的 Python 实现与三个现场坑）。

本文目录

一、先给结论

先把判断摆出来，后面再展开论据：

二、政策原文里，和"数据采集"直接相关的三句话

《上海市加快发展新一代智能制造行动方案》由上海市经济信息化委于 2026 年 10 月 27 日印发，10 月 28 日经中国新闻网、证券时报、澎湃新闻等多家媒体报道。与本文主题直接相关的表述有三句：

第一句，写在基础级智能工厂的培育要求里：引导制造企业合理部署智能制造装备、软件和系统，实现核心数据实时采集、关键生产工序自动化、生产与经营管理信息化。

第二句，关于量级：到 2028 年实现基础级智能工厂累计达 1600 家；推动基础级智能工厂 AI 普及率达 100%，先进级、卓越级、领航级智能工厂的 AI 应用场景比例分别达 10%、30%、70%。

第三句，关于梯度：基础级 1600 家、先进级 500 家、卓越级 50 家、领航级 5 家，构成"54321"目标里的工厂梯度部分。

需要说明的是，同批发布的 2000 个 AI 应用场景、10 个垂域大模型、300 个以上高水平智能体、工业机器人密度 600 台/万人等指标，主要落在先进级及以上。本文只讨论基础级这一层——也就是大多数中小制造企业真正要跨过去的那一步。

三、"核心数据"到底指哪些：一张三级清单

这是我被问得最多的问题。政策文件不会给你点位表，但工程上可以拆得很清楚：

【配图位置1】请在此上传 图1_核心数据三层与采集链路.png

关键判断在这里：三层都要采，但存储策略完全不同。

L1 和 L2 是时间序列，写入频率高、量大，应该进时序数据库；L3 是业务事实，天然要跟工单、批次绑定，应该进业务库。我见过不少项目把三层混在一张表里，半年后数据量上来，一张报表查三分钟，最后没人愿意用，系统就废在那里。

四、采集架构：设备数据为什么可以不必先过 MES

这是本文最想说清的一点。

MES 的定位是工单管理、报工、质量追溯和过程管控，它的数据粒度是件、批、班次。而设备点位是秒级甚至亚秒级的。把后者直接写进前者的库里，会带来三个具体问题：

• 写入放大：一台设备 50 个点位、1 秒一次，一天就是 432 万条记录。二十台设备一个月就上亿，MES 数据库扛不住。
• 表结构膨胀：为塞下高频点位而加的宽表，会让原本清晰的工单模型变得难以维护。
• 拖垮生产系统：MES 是车间每天都要用的系统，任何查询变慢都会直接影响排产和报工，这个代价没人愿意承担。

更合理的分层是：

设备层：PLC / CNC / 传感器 / 仪表
↓ （Modbus、OPC UA、串口、IO 采集）
边缘层：边缘网关 → 点位规范化 → 本地缓存（断网续传）
↓ （MQTT 上行，厂内网络）
数据层：时序数据库（秒级原始 + 分钟级聚合）
↓ （标准接口 / 聚合视图）
应用层：MES 取工单所需结果、看板取状态、AI 取特征

这套分层的另一个好处是边界清晰：设备原始数据、时序库存、模型推理都放在厂内，只有聚合后的结果按需外送。对那些明确要求数据不出厂的企业来说，这条链路天然满足要求，不需要事后打补丁。

MES 与时序库的对接：按需取聚合结果

分层之后，MES 不再直连设备点位，而是通过聚合视图或标准接口按工单、批次取数。典型做法是：MES 侧只保存工单号、批次号、设备编号和起止时间，取数时把这段时间窗口和点位前缀传给时序库，由时序库返回分钟级聚合结果。

下面是一个 SQL 聚合查询示例，从 agg_1m 表按时间段和点位前缀取某工单周期的节拍与主轴负载均值：

— 输入：工单周期起止时间（毫秒）、设备点位前缀
— 输出：该工单周期内每分钟的节拍与主轴负载均值
SELECT
a.ts AS minute_ts,
a.point AS point,
a.v_avg AS avg_value,
a.v_min AS min_value,
a.v_max AS max_value,
a.v_cnt AS sample_cnt
FROM agg_1m a
WHERE a.ts >= :work_order_start_ms
AND a.ts < :work_order_end_ms
AND a.point LIKE :point_prefix || '%'
AND (
a.point LIKE '%.PROCESS.cycle_time'
OR a.point LIKE '%.PROCESS.spindle_load'
)
ORDER BY a.point, a.ts;

如果 MES 需要的是工单级的单一数值，而不是逐分钟明细，可以在外层再套一层聚合：

— 工单级汇总：节拍均值、主轴负载均值与样本数
SELECT
SUM(CASE WHEN point LIKE '%.PROCESS.cycle_time' THEN v_avg * v_cnt END)
/ NULLIF(SUM(CASE WHEN point LIKE '%.PROCESS.cycle_time' THEN v_cnt END), 0) AS avg_cycle_time,
SUM(CASE WHEN point LIKE '%.PROCESS.spindle_load' THEN v_avg * v_cnt END)
/ NULLIF(SUM(CASE WHEN point LIKE '%.PROCESS.spindle_load' THEN v_cnt END), 0) AS avg_spindle_load,
SUM(v_cnt) AS total_samples
FROM agg_1m
WHERE ts >= :work_order_start_ms
AND ts < :work_order_end_ms
AND point LIKE :point_prefix || '%'
AND (point LIKE '%.PROCESS.cycle_time' OR point LIKE '%.PROCESS.spindle_load');

注意这里用的是 v_avg * v_cnt 加权，而不是直接对 v_avg 求平均——分钟级聚合表的每一行代表不同数量的原始样本，直接平均会让采样稀疏的分钟被过度放大。

为什么这种「按需取聚合结果」比 MES 直连设备点位更合理，可以从三个角度看：

• 数据量差了两个数量级：一台设备一天 432 万条原始点位，聚合到分钟级只剩 1440 条。MES 每次取数只扫聚合表，查询从分钟级降到毫秒级，工单报工、批次追溯这些高频操作不会被拖慢。
• 职责边界清晰：时序库负责高频写入、降采样和保留策略，MES 负责工单、批次和业务语义。MES 不需要理解点位命名规则，只需要知道「这个工单、这台设备、这段时间」；点位前缀和聚合口径由数据层统一维护，改一次全厂生效。
• 故障隔离：设备抖动、网关断连、补发洪峰都发生在时序库这一侧，MES 的库不会被写入放大波及。反过来，MES 升级或报表高峰也不会影响采集链路，两边各自演进。

落地时建议把取数逻辑封装成一个只读接口（比如 GET /api/oee/points?work_order=…&prefix=…），MES 只调接口、不写 SQL。这样聚合口径调整、时序库从 SQLite 换成 TDengine 或 InfluxDB 时，MES 侧一行代码都不用改。

五、工程实现：从 PLC / CNC 到时序库的一条最小链路

下面给一段可以直接跑起来的参考实现。为了让示例不依赖特定硬件，采集侧做了可替换的封装：有 OPC UA 服务器就走 OPC UA，没有就用模拟数据先看通链路。

5.1 采集与上行

# -*- coding: utf-8 -*-
"""
设备数据采集最小链路示例：采集 → 点位规范化 → MQTT 上行
依赖：pip install asyncua paho-mqtt
说明：生产环境请补全异常处理、断网续传与鉴权配置。
"""
import asyncio
import json
import time
import random
from datetime import datetime
———- 1. 点位命名规范化 ———-
命名规则：{厂区}.{产线}.{设备类型}.{设备编号}.{信号类别}.{信号名}
这一步必须最先做。同一台设备在三个系统里三个名字，是后期最难收拾的坑。
def make_point(workshop, line, dev_type, dev_no, category, signal):
return f"{workshop}.{line}.{dev_type}.{dev_no}.{category}.{signal}"
def normalize(value, scale=1.0, offset=0.0, precision=2):
"""量程换算 + 精度收敛，避免浮点噪声污染时序库"""
return round(value * scale + offset, precision)
———- 2. 采集封装 ———-
class MockSource:
"""模拟数据源：没有真实设备时先把链路跑通（降级方案）"""
def init(self, device_no):
self.device_no = device_no
self._state = "RUN"
async def read(self):
# 真实场景替换为 OPC UA / Modbus 读取
if random.random() &lt; 0.03:
self._state = random.choice(["RUN", "IDLE", "ALARM"])
return {
"state": self._state, # 设备状态（离散量）
"spindle_load": normalize(random.uniform(20, 85)), # 主轴负载 %
"spindle_speed": normalize(random.uniform(800, 3000), precision=0), # 转速 rpm
"cycle_time": normalize(random.uniform(45, 75)), # 节拍 秒
}
class OpcUaSource:
"""
真实 OPC UA 采集：连接服务器 → 按 NodeId 批量读取 → 断线重连 + 读取超时
关键参数说明：
endpoint : OPC UA 服务器地址，如 opc.tcp://192.168.1.10:4840
node_map : 信号名 → NodeId 的映射，NodeId 形如 "ns=2;s=CNC01.SpindleLoad"
read_timeout : 单次批量读取超时（秒），超时即判定本次采集失败，避免主循环被卡死
reconnect_interval : 断线后重连间隔（秒），指数退避的起始值
max_reconnect_interval : 重连间隔上限（秒），防止无限增长
"""
def init(self, endpoint, node_map, read_timeout=3.0,
reconnect_interval=1.0, max_reconnect_interval=30.0):
self.endpoint = endpoint
self.node_map = node_map
self.read_timeout = read_timeout
self.reconnect_interval = reconnect_interval
self.max_reconnect_interval = max_reconnect_interval
self._client = None
self._nodes = {} # 信号名 → Node 对象，连接成功后一次性建好
self._connected = False
async def connect(self):
"""建立连接并缓存 Node 对象；失败时抛出异常由上层重连逻辑处理"""
from asyncua import Client
self._client = Client(url=self.endpoint, timeout=self.read_timeout)
await self._client.connect()
# 批量预取 Node 对象，避免每次读取都重新解析 NodeId
self._nodes = {
signal: self._client.get_node(node_id)
for signal, node_id in self.node_map.items()
}
self._connected = True
print(f"[opcua] 已连接 {self.endpoint}，点位 {len(self._nodes)} 个")

async def read(self):
"""按 NodeId 批量读取；断线或超时返回 None，由主循环决定是否降级"""
if not self._connected:
return None
try:
# read_values 一次请求读多个节点，比逐个 read_value 少一个数量级的往返
values = await asyncio.wait_for(
self._nodes_read(), timeout=self.read_timeout
)
return values
except asyncio.TimeoutError:
print(f"[opcua] 读取超时（&gt;{self.read_timeout}s），本次采集跳过")
return None
except Exception as e:
print(f"[opcua] 读取异常：{e}，标记断线")
self._connected = False
return None

async def _nodes_read(self):
"""内部：按 node_map 顺序批量读取并组装为 {信号名: 值}"""
signals = list(self._nodes.keys())
nodes = [self._nodes[s] for s in signals]
raw_values = await self._client.read_values(nodes)
result = {}
for signal, value in zip(signals, raw_values):
# 离散量（state）保持字符串语义，模拟量与 MockSource 对齐
if signal == "state":
result[signal] = "RUN" if value else "IDLE"
else:
result[signal] = normalize(float(value))
return result

async def ensure_connected(self):
"""断线重连：指数退避，直到连上或达到上限"""
delay = self.reconnect_interval
while not self._connected:
try:
await self.connect()
return True
except Exception as e:
print(f"[opcua] 重连失败：{e}，{delay:.1f}s 后重试")
await asyncio.sleep(delay)
delay = min(delay * 2, self.max_reconnect_interval)
return True

async def close(self):
if self._client:
try:
await self._client.disconnect()
except Exception:
pass
self._connected = False
———- 3. MQTT 上行 ———-
class MqttUploader:
def init(self, host, port=1883, topic="factory/l1/telemetry"):
import paho.mqtt.client as mqtt
self.topic = topic
self.client = mqtt.Client(client_id=f"gw-{int(time.time())}")
self.client.connect(host, port, keepalive=60)
self.client.loop_start()
def publish(self, payload: dict):
# qos=1 保证至少一次送达；时序数据允许少量重复，不允许丢
self.client.publish(self.topic, json.dumps(payload), qos=1)
———- 4. 主循环 ———-
async def main():
workshop, line = "SH01", "LINE_A"
devices = [("CNC", "01"), ("CNC", "02")]
# 有 OPC UA 服务器就走真实采集，没有则降级为 MockSource
USE_OPCUA = True
OPCUA_ENDPOINT = "opc.tcp://192.168.1.10:4840"
NODE_MAP = {
"state": "ns=2;s=CNC01.Status.State",
"spindle_load": "ns=2;s=CNC01.Process.SpindleLoad",
"spindle_speed": "ns=2;s=CNC01.Process.SpindleSpeed",
"cycle_time": "ns=2;s=CNC01.Process.CycleTime",
}

sources = {}
for dev_type, dev_no in devices:
if USE_OPCUA:
src = OpcUaSource(OPCUA_ENDPOINT, NODE_MAP)
await src.ensure_connected()
sources[dev_no] = src
else:
sources[dev_no] = MockSource(dev_no)

# 生产环境替换为真实 broker 地址
up = MqttUploader(host="127.0.0.1")

while True:
ts = datetime.utcnow().isoformat(timespec="milliseconds") + "Z"
batch = []
for dev_type, dev_no in devices:
src = sources[dev_no]
raw = await src.read()
# 读取失败：OPC UA 断线时降级到 MockSource，保证链路不断流
if raw is None:
if isinstance(src, OpcUaSource):
print(f"[{dev_no}] OPC UA 不可用，降级为模拟数据")
raw = await MockSource(dev_no).read()
asyncio.create_task(src.ensure_connected()) # 后台重连
else:
continue
for signal, value in raw.items():
category = "STATUS" if signal == "state" else "PROCESS"
batch.append({
"ts": ts,
"point": make_point(workshop, line, dev_type, dev_no, category, signal),
"value": value,
})
# 批量上报，避免逐点发送把 broker 打满
up.publish({"ts": ts, "data": batch})
print(f"[{ts}] 上报 {len(batch)} 个点位")
await asyncio.sleep(1)
if name == "main":
asyncio.run(main())

断网续传与异常处理

上面那段代码有一个致命假设：网络一直是通的。车间里断网是常态——交换机重启、无线 AP 掉线、broker 升级，任何一次抖动都会让 publish 抛异常或者静默丢包。所以边缘网关必须自带一块本地缓存，断连时写本地、重连后按时间顺序补发。

这里用 SQLite 做环形缓存队列，零外部依赖，掉电也不丢。三个设计要点：

• 环形队列：表里只保留最近 N 条（比如 50 万条），超出后按 ts 淘汰最老的，防止磁盘被写满把网关搞崩。
• 按时间顺序补发：补发时严格按 ts 升序取，保证时序库里的数据不乱序，否则降采样和趋势查询都会出问题。
• 幂等去重：qos=1 是“至少一次”，重连补发、broker 重投都会产生重复。落库侧用 (ts, point) 做联合主键，INSERT OR REPLACE 幂等写入，重复数据自然被覆盖。

# -*- coding: utf-8 -*-
"""
断网续传：SQLite 环形缓存队列 + 重连补发
依赖：pip install paho-mqtt
思路：发送前先落本地，发送成功再删除；断连时只写不删，重连后按 ts 升序补发。
"""
import json
import sqlite3
import time
import threading
CACHE_DB = "edge_cache.db"
MAX_ROWS = 500_000 # 环形队列容量，按磁盘和断网时长估算
BATCH_SIZE = 500 # 单次补发批量，避免一次性把 broker 打满
class LocalQueue:
"""本地环形缓存队列：先落盘，再发送，成功才删除"""
def __init__(self, db_path=CACHE_DB, max_rows=MAX_ROWS):
self.max_rows = max_rows
self._lock = threading.Lock()
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute("PRAGMA journal_mode=WAL") # 读写并发更稳，掉电更安全
self.conn.execute("""
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL, — ISO8601 毫秒时间戳，补发排序依据
topic TEXT NOT NULL,
payload TEXT NOT NULL
)
""")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_outbox_ts ON outbox(ts)")
self.conn.commit()
def put(self, ts, topic, payload):
"""入队：写入本地，同时做环形淘汰"""
with self._lock:
self.conn.execute(
"INSERT INTO outbox (ts, topic, payload) VALUES (?, ?, ?)",
(ts, topic, json.dumps(payload, ensure_ascii=False)),
)
# 环形淘汰：只保留最近 max_rows 条
self.conn.execute(
"DELETE FROM outbox WHERE id <= (SELECT MAX(id) – ? FROM outbox)",
(self.max_rows,),
)
self.conn.commit()
def peek_batch(self, limit=BATCH_SIZE):
"""按时间顺序取一批待补发数据"""
with self._lock:
cur = self.conn.execute(
"SELECT id, ts, topic, payload FROM outbox ORDER BY ts ASC, id ASC LIMIT ?",
(limit,),
)
return cur.fetchall()
def ack(self, ids):
"""发送成功后删除，ack 之前崩溃会重发，靠下游幂等兜底"""
if not ids:
return
with self._lock:
self.conn.executemany("DELETE FROM outbox WHERE id = ?", [(i,) for i in ids])
self.conn.commit()
def pending(self):
with self._lock:
return self.conn.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
class ReliableUploader:
"""带断网续传的 MQTT 上行封装"""
def init(self, host, port=1883, topic="factory/l1/telemetry"):
import paho.mqtt.client as mqtt
self.topic = topic
self.queue = LocalQueue()
self.online = False
self.client = mqtt.Client(client_id=f"gw-{int(time.time())}")
self.client.on_connect = self._on_connect
self.client.on_disconnect = self._on_disconnect
self.client.connect_async(host, port, keepalive=60) # 异步连接，断网不阻塞主循环
self.client.loop_start()
def _on_connect(self, client, userdata, flags, rc):
self.online = (rc == 0)
if self.online:
print("[mqtt] 已连接，开始补发本地缓存")
threading.Thread(target=self._flush, daemon=True).start()
def _on_disconnect(self, client, userdata, rc):
self.online = False
print(f"[mqtt] 连接断开 rc={rc}，转入本地缓存模式")
def publish(self, ts, payload):
"""统一入口：无论在线与否，先落本地，再尝试发送"""
self.queue.put(ts, self.topic, payload)
if self.online:
self._flush()
def _flush(self):
"""按时间顺序补发，发送成功才 ack 删除"""
while self.online:
rows = self.queue.peek_batch()
if not rows:
return
ok_ids = []
for rid, ts, topic, payload in rows:
info = self.client.publish(topic, payload, qos=1)
# wait_for_publish 确认 broker 已接收，避免“假成功”丢数据
try:
info.wait_for_publish(timeout=5)
if info.is_published():
ok_ids.append(rid)
else:
break
except Exception:
self.online = False
break
self.queue.ack(ok_ids)
if not self.online:
return
———- 落库侧：qos=1 重复数据的幂等去重 ———-
def init_sink(conn):
"""ts + point 联合主键，天然幂等：重复写入只会覆盖，不会产生脏数据"""
conn.execute("""
CREATE TABLE IF NOT EXISTS raw (
ts TEXT NOT NULL, — 与上行 payload 中的 ts 完全一致
point TEXT NOT NULL,
value REAL,
PRIMARY KEY (ts, point)
)
""")
conn.commit()
def write_idempotent(conn, rows):
"""rows: [(ts, point, value), …]
qos=1 下同一条数据可能被投递多次，INSERT OR REPLACE 保证结果一致。
"""
conn.executemany(
"INSERT OR REPLACE INTO raw (ts, point, value) VALUES (?, ?, ?)", rows
)
conn.commit()
if name == "main":
up = ReliableUploader(host="127.0.0.1")
while True:
ts = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + ".000Z"
up.publish(ts, {"ts": ts, "data": [{"point": "SH01.LINE_A.CNC.01.PROCESS.spindle_load", "value": 62.5}]})
print(f"[{ts}] 本地待补发 {up.queue.pending()} 条")
time.sleep(1)

几个容易忽略的细节：

• 先落盘再发送，而不是发送失败再落盘。后者在进程被 kill 或掉电时会丢掉正在发送的那一批。
• ack 之前崩溃会重发，这是有意为之——宁可重复，不可丢失。重复由下游 (ts, point) 主键兜住。
• 补发要限速。断网两小时可能积压几十万条，重连后一次性灌进去会把 broker 和时序库打满，所以用 BATCH_SIZE 分批，必要时加 sleep。
• 环形容量要算清楚。50 万条按每秒 100 条算约 1.4 小时，如果现场断网可能超过这个时长，就要调大 MAX_ROWS 或降低采集频率。

5.2 落库与降采样

时序数据落地后，真正决定这套系统能活多久的是降采样策略。原始数据保留短期，聚合数据保留长期，否则半年后查询性能会崩掉。

# -*- coding: utf-8 -*-
"""
时序数据落库与降采样示例（SQLite 版，零外部依赖即可运行）
生产环境可替换为 InfluxDB / TDengine，逻辑一致。
"""
import sqlite3
import time
from datetime import datetime, timedelta
DB = "telemetry.db"
def init_db(conn):
conn.execute("""
CREATE TABLE IF NOT EXISTS raw (
ts INTEGER NOT NULL, — 毫秒时间戳
point TEXT NOT NULL, — 点位全名
value REAL,
PRIMARY KEY (ts, point)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS agg_1m (
ts INTEGER NOT NULL, — 对齐到分钟
point TEXT NOT NULL,
v_avg REAL, v_min REAL, v_max REAL, v_cnt INTEGER,
PRIMARY KEY (ts, point)
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_raw_point ON raw(point, ts)")
conn.commit()
def write_batch(conn, rows):
"""rows: [(ts_ms, point, value), …]"""
conn.executemany(
"INSERT OR REPLACE INTO raw (ts, point, value) VALUES (?, ?, ?)", rows
)
conn.commit()
def rollup_to_1m(conn, end_ms, window_min=1):
"""把窗口内的原始数据聚合为分钟级，并删除已聚合的原始数据"""
start_ms = end_ms – window_min * 60 * 1000
cur = conn.execute("""
SELECT point,
AVG(value), MIN(value), MAX(value), COUNT(*)
FROM raw WHERE ts >= ? AND ts < ? GROUP BY point
""", (start_ms, end_ms))
agg = cur.fetchall()
minute_ts = (end_ms // 60000) * 60000
conn.executemany(
"INSERT OR REPLACE INTO agg_1m (ts, point, v_avg, v_min, v_max, v_cnt) "
"VALUES (?, ?, ?, ?, ?, ?)",
[(minute_ts, p, a, mi, ma, c) for p, a, mi, ma, c in agg]
)
conn.execute("DELETE FROM raw WHERE ts >= ? AND ts < ?", (start_ms, end_ms))
conn.commit()
return len(agg)
def query_trend(conn, point, hours=8):
"""查询近 N 小时趋势：优先走聚合表，这是看板能秒开的原因"""
since = int((datetime.utcnow() – timedelta(hours=hours)).timestamp() * 1000)
cur = conn.execute(
"SELECT ts, v_avg FROM agg_1m WHERE point = ? AND ts >= ? ORDER BY ts",
(point, since)
)
return cur.fetchall()
if name == "main":
conn = sqlite3.connect(DB)
init_db(conn)
# 模拟写入 120 秒数据（每秒一个点）
now = int(time.time() * 1000)
rows = []
for i in range(120):
ts = now – (120 – i) * 1000
rows.append((ts, "SH01.LINE_A.CNC.01.PROCESS.spindle_load", 40 + i % 30))
write_batch(conn, rows)
print("写入原始点位数:", len(rows))
聚合为分钟级
n = rollup_to_1m(conn, now, window_min=5)
print("聚合点位:", n)
查询趋势
trend = query_trend(conn, "SH01.LINE_A.CNC.01.PROCESS.spindle_load")
print("趋势点数:", len(trend), "示例:", trend[:3])
conn.close()</code></pre>
这段代码跑起来不需要任何外部服务，建议先在笔记本上把链路走一遍，再接到真实设备上。顺序很重要：先验证链路，再接设备，反过来做会在现场浪费大量时间。
六、三个真实会踩的坑
坑一：老旧设备没有网口。
十年前的 CNC 只有 RS-232 串口，或者连 PLC 都没有。可行做法有三档：有串口的走 Modbus RTU 转网关；只有 IO 信号的用 IO 模块采运行状态；连状态都采不到的，用外接电流互感器测整机功率，反推运行/待机/停机——这不是取巧，而是很多老产线唯一能落地的方案，精度也够用来算 OEE（设备综合效率，Overall Equipment Effectiveness）。
坑二：点位命名没有标准。
同一台注塑机，在设备台账里叫"IM-03"，在电工口里叫"三号机"，在 MES 里叫"INJ_003"。三个名字指向同一台设备，任何跨系统对账都要人工翻译。先建点位字典再采集，命名规则参考前文 make_point()，全厂统一一次，后面所有系统都受益。
坑三：没有降采样策略。
前面给了 rollup_to_1m() 的实现。没有它，秒级原始数据存一年，查询会从秒级退化到分钟级，看板打不开，工程师就开始怀疑这套系统的价值。原始 30 天 + 分钟级 2 年 + 小时级 5 年，是大多数中小厂够用的配比。
七、落地检查清单
【配图位置2】请在此上传 图2_落地检查清单.png
按这个顺序推进，可以少走很多弯路：
盘点设备清单：型号、出厂年份、控制器类型、可提供的接口（网口 / 串口 / IO / 无）
确定三层数据范围：L1 采哪些、L2 采哪些、L3 怎么跟工单绑定，先做减法
建立点位字典：全厂统一命名，一次到位
确认采集频率：L1 秒级、L2 秒到分钟级、L3 事件触发，不要一律最高频
选定存储方案：时序库选型 + 降采样配比 + 备份策略
明确数据边界：哪些在厂内、哪些可外送、谁能访问
先跑通一条产线：一条线跑顺再复制，不要全厂铺开
验收标准写进合同：采集覆盖率、数据完整率、断网续传时长，能量化的都量化
八、常见问题
Q1：一定要先上 MES 才能算基础级智能工厂吗？
不是。政策对基础级的表述是"核心数据实时采集、关键生产工序自动化、生产与经营管理信息化"三件事并列，采集是其中之一，并不以 MES 为前置条件。实际工程中，先把设备数据采上来、存好、能用，反而能为后续 MES 选型提供更真实的依据——至少你知道自己到底要管什么。
Q2：采集频率是不是越高越好？
不是。频率要匹配业务问题：算 OEE 用秒级足够，分析工艺漂移分钟级足够，做刀具寿命预测可能要亚秒级。按问题定频率，高频数据的存储与计算成本是持续的，不是一次性的。
Q3：要求数据不出厂，还能做 AI 分析吗？
能，而且这是更常见的部署方式。采集、存储、模型推理全部放在厂内，只有需要跨厂协同的聚合结果（比如集团级产能看板）才外送。私有化部署的技术栈已经很成熟，需要权衡的是算力成本，不是可行性。
Q4：1600 家基础级的"核心数据实时采集"有没有量化标准？
目前公开表述中没有给出点位数量、采集频率的具体阈值，只明确了"实时采集"这个性质要求。工程上的判断依据是：采到的数据能否回答"设备现在能不能干活""这道工序稳不稳""这批活干得怎么样"这三个问题。能回答，就达到了基础级这一步的实质要求。
九、总结
回到政策那三句话。基础级 1600 家这个量级意味着，绝大多数工厂要跨的第一步不是选大模型、不是买机器人，而是把车间里那些已经存在了十年、二十年的设备数据，稳定、完整、可追溯地采上来。
这一层的门槛不高，但很实：需要的是点位字典、采集链路、时序存储和降采样策略，而不是概念。把这一层做扎实，上面那些 AI 场景才有着落；跳过这一层直接谈智能体，最后往往停在演示里。
参考与说明
政策来源：《上海市加快发展新一代智能制造行动方案》，上海市经济信息化委，2026 年 10 月 27 日印发；2026 年 10 月 28 日经中国新闻网、证券时报、澎湃新闻等媒体报道。文中引用的数字均来自上述公开报道。
利益相关：本人从事工厂 AI 系统的落地工作，文中架构与代码来自实际部署过的项目，客户名称与具体数据因未获公开授权已做脱敏，示例代码为脱敏后的简化版本。
说明：本文写作过程中使用了 AI 工具辅助整理资料与校对，技术判断与现场经验由本人负责。文中的政策引用可自行核验。

5.3 从采集数据算 OEE

数据采上来、存好了，接下来要回答一个老板最关心的问题：这套设备到底有多少有效产出。OEE（设备综合效率，Overall Equipment Effectiveness）就是标准答案，它由三个因子相乘得到：

OEE = 可用率（Availability） × 性能率（Performance） × 良品率（Quality）

三个因子分别对应前文三级清单里的不同层，这也是为什么三层都要采：

一句话记法：L1 决定机器转不转，L2 决定转得快不快，L3 决定转出来的活好不好。缺任何一层，OEE 都算不出来——只有 L1 能算可用率，只有 L1+L2 能算性能和可用率，三层齐全才是完整的 OEE。

下面这段代码直接复用 5.2 里的 raw 与 agg_1m 两张表，零外部依赖即可运行：

# -*- coding: utf-8 -*-
"""
从采集数据计算 OEE：可用率 × 性能率 × 良品率
依赖：仅标准库，复用 5.2 的 raw / agg_1m 两张表
说明：L3 的产出与质量数据在真实项目里来自 MES 报工或视觉检测，
这里用一张 production 表模拟，字段与 MES 侧保持一致。
"""
import sqlite3
import time
DB = "telemetry.db"
POINT_PREFIX = "SH01.LINE_A.CNC.01"
IDEAL_CYCLE_TIME = 50.0 # 理论节拍（秒/件），来自工艺文件或设备额定值
PLANNED_MINUTES = 60 # 统计窗口：计划生产 60 分钟
def init_production(conn):
"""L3 产出与质量结果：真实场景由 MES 报工或视觉检测写入"""
conn.execute("""
CREATE TABLE IF NOT EXISTS production (
ts INTEGER NOT NULL, — 报工时间（毫秒）
device TEXT NOT NULL, — 设备编号
good_count INTEGER NOT NULL, — 合格数
bad_count INTEGER NOT NULL, — 不合格数
PRIMARY KEY (ts, device)
)
""")
conn.commit()
def calc_availability(conn, prefix, start_ms, end_ms, planned_minutes):
"""可用率 A = 实际运行时间 / 计划生产时间
数据来源：L1 设备运行状态，从 agg_1m 里取 state 的分钟级样本。
约定：state 点位 value=1 表示 RUN，0 表示非 RUN（IDLE/ALARM/停机）。
"""
cur = conn.execute("""
SELECT SUM(v_avg * v_cnt) / NULLIF(SUM(v_cnt), 0)
FROM agg_1m
WHERE point = ? AND ts >= ? AND ts < ?
""", (f"{prefix}.STATUS.state", start_ms, end_ms))
row = cur.fetchone()
run_ratio = row[0] if row and row[0] is not None else 0.0
planned_ms = planned_minutes * 60 * 1000
run_ms = run_ratio * planned_ms
return run_ms, planned_ms, (run_ms / planned_ms if planned_ms else 0.0)
def calc_performance(conn, prefix, start_ms, end_ms, run_ms, ideal_cycle):
"""性能率 P = 理论节拍 × 实际产出 / 实际运行时间
数据来源：L2 工序过程量，实际节拍取 agg_1m 的加权均值。
实际产出 = 实际运行时间 / 实际平均节拍。
"""
cur = conn.execute("""
SELECT SUM(v_avg * v_cnt) / NULLIF(SUM(v_cnt), 0)
FROM agg_1m
WHERE point = ? AND ts >= ? AND ts < ?
""", (f"{prefix}.PROCESS.cycle_time", start_ms, end_ms))
row = cur.fetchone()
actual_cycle = row[0] if row and row[0] is not None else 0.0
if actual_cycle <= 0 or run_ms <= 0:
return 0.0, 0.0, 0.0
actual_output = run_ms / 1000.0 / actual_cycle # 实际产出件数
performance = (ideal_cycle * actual_output) / (run_ms / 1000.0)
return actual_cycle, actual_output, min(performance, 1.0) # 上限截断到 1
def calc_quality(conn, device, start_ms, end_ms):
"""良品率 Q = 合格数 / 总产出数
数据来源：L3 产出与质量结果，来自 production 表（MES 报工）。
"""
cur = conn.execute("""
SELECT SUM(good_count), SUM(good_count + bad_count)
FROM production
WHERE device = ? AND ts >= ? AND ts < ?
""", (device, start_ms, end_ms))
row = cur.fetchone()
good = row[0] or 0
total = row[1] or 0
return good, total, (good / total if total else 0.0)
def calc_oee(conn, prefix, device, end_ms, planned_minutes=PLANNED_MINUTES,
ideal_cycle=IDEAL_CYCLE_TIME):
"""OEE = A × P × Q，返回完整明细便于排查是哪一项拖后腿"""
start_ms = end_ms – planned_minutes * 60 * 1000
run_ms, planned_ms, a = calc_availability(
conn, prefix, start_ms, end_ms, planned_minutes)
actual_cycle, output, p = calc_performance(
conn, prefix, start_ms, end_ms, run_ms, ideal_cycle)
good, total, q = calc_quality(conn, device, start_ms, end_ms)
oee = a * p * q
return {
"window": (start_ms, end_ms),
"run_minutes": round(run_ms / 60000, 2),
"planned_minutes": planned_minutes,
"actual_cycle": round(actual_cycle, 2),
"ideal_cycle": ideal_cycle,
"output": round(output, 1),
"good": good,
"total": total,
"A": round(a, 4),
"P": round(p, 4),
"Q": round(q, 4),
"OEE": round(oee, 4),
}
if name == "main":
conn = sqlite3.connect(DB)
init_production(conn)
now = int(time.time() * 1000)
start = now – PLANNED_MINUTES * 60 * 1000
—- 模拟写入 L1 / L2 的分钟级聚合数据（真实场景由 5.2 的 rollup_to_1m 生成）—-
rows = []
for i in range(PLANNED_MINUTES):
ts = start + i * 60 * 1000
# 前 45 分钟 RUN，后 15 分钟停机 → 可用率约 0.75
state = 1.0 if i &lt; 45 else 0.0
rows.append((ts, f"{POINT_PREFIX}.STATUS.state", state, state, state, 60))
# 实际节拍 55 秒，理论 50 秒 → 性能率约 0.909
rows.append((ts, f"{POINT_PREFIX}.PROCESS.cycle_time", 55.0, 54.0, 56.0, 60))
conn.executemany(
"INSERT OR REPLACE INTO agg_1m (ts, point, v_avg, v_min, v_max, v_cnt) "
"VALUES (?, ?, ?, ?, ?, ?)", rows)
—- 模拟写入 L3 产出与质量（真实场景来自 MES 报工）—-
conn.execute(
"INSERT OR REPLACE INTO production (ts, device, good_count, bad_count) "
"VALUES (?, ?, ?, ?)", (now, "CNC.01", 48, 2))
conn.commit()
result = calc_oee(conn, POINT_PREFIX, "CNC.01", now)
print("=" * 46)
print(f"统计窗口：{PLANNED_MINUTES} 分钟 设备：CNC.01")
print(f"运行时间：{result['run_minutes']} / {result['planned_minutes']} 分钟")
print(f"实际节拍：{result['actual_cycle']} 秒 理论节拍：{result['ideal_cycle']} 秒")
print(f"产出：{result['output']} 件 合格：{result['good']} / {result['total']}")
print("-" * 46)
print(f"可用率 A = {result['A']:.2%}")
print(f"性能率 P = {result['P']:.2%}")
print(f"良品率 Q = {result['Q']:.2%}")
print(f"OEE = {result['OEE']:.2%}")
print("=" * 46)
conn.close()</code></pre>
示例输出（注释形式，便于对照）：
统计窗口：60 分钟 设备：CNC.01
运行时间：45.0 / 60 分钟
实际节拍：55.0 秒 理论节拍：50.0 秒
产出：49.1 件 合格：48 / 50
可用率 A = 75.00%
性能率 P = 90.91%
良品率 Q = 96.00%
OEE = 65.45%
这个 65.45% 就是典型的「能看但不够好」：可用率 75% 说明停机时间占了四分之一，是最大的短板；性能率 90.91% 说明节拍比理论值慢了约 10%，可能是工艺参数偏保守或刀具磨损；良品率 96% 已经不错。排查顺序应该是先看 A、再看 P、最后看 Q——停机损失通常最容易改善，也最容易被忽视。
几个落地时要注意的点：
理论节拍必须来自工艺文件，不能拿历史最快值当基准，否则性能率永远接近 100%，失去诊断意义。
计划生产时间要扣掉计划停机（换班、计划保养、无排产），否则可用率会被系统性低估。
L3 数据要能对上工单。OEE 按班次、按工单、按设备三个维度都要能切，这要求 production 表带上工单号和班次字段，而不是只记一个总数。
OEE 是诊断指标，不是考核指标。一旦拿来考核，现场就会开始挑数据——把报警改成待机、把不良品挪到下个班次，指标好看了，问题还在。