feat: 完成 issue #21 点位字典 CSV schema 增加协议(protocol)维度

core/edge-gateway/README.md

@@ -21,9 +21,9 @@
 edge-gateway/
 ├── config/gateway.example.yaml   # 采集配置外置示例（模板参数化）
 ├── point_dict/                   # 点位字典：CSV schema + 导入 + 自动校验
-│   ├── schema.py                 # 字段定义与约束（对齐 PRD 5.1 字段表）
+│   ├── schema.py                 # 字段定义与约束（对齐 PRD 5.1 字段表 + issue #21 协议维度）
 │   ├── loader.py                 # CSV → 内存模型
-│   └── validator.py              # 校验：缺失字段 / 量纲 / 重复点号 / 采样率
+│   └── validator.py              # 校验：缺失字段 / 量纲 / 重复点号 / 采样率 / 协议
 ├── drivers/                      # 协议可插拔驱动（只读）
 │   ├── base.py                   # 驱动抽象基类（唯一入口 read_points）
 │   ├── opcua_driver.py           # OPC UA（和利时 DCS 等）
@@ -61,7 +61,8 @@ python -m unittest discover -s tests
 
 ## 模板化说明（换行业只改配置，不改代码）
 
-- 点位范围 / 采样率 / 量纲：由**点位字典 CSV** 驱动（模板 → 行业点位集）。
+- 点位范围 / 采样率 / 量纲 / 协议：由**点位字典 CSV** 驱动（模板 → 行业点位集）。
+  协议列（可选）提供点位级协议覆盖；留空则按 `gateway.yaml` drivers 段前缀路由。
 - 协议选型与连接参数：由 `gateway.yaml` 的 `drivers` 段驱动（模板 → 行业协议栈）。
 - Kafka topic 命名：`{template}.{device}.points`，随配置模板变化。
 - 新增协议：实现 `drivers/base.py` 的 `Driver` 子类并注册即可，引擎与上行链路零改动。

core/edge-gateway/collector/engine.py

@@ -58,14 +58,27 @@ class CollectorEngine:
 
     # ------------------------------------------------------------------
     def _build_routing(self) -> Dict[str, Driver]:
-        """按设备前缀把点位路由到对应驱动实例（模板配置驱动）。"""
+        """点位 → 驱动实例路由。
+
+        匹配优先级（issue #21「协议」维度）：
+        1. 点位级：点位字典 CSV protocol 列精确匹配驱动槽位协议；
+        2. 模板级（缺省）：按设备前缀 device_prefixes 匹配；
+        3. 兜底：无前缀的空槽位接收未匹配点位。
+        """
         routing: Dict[str, Driver] = {}
+        slot_by_protocol: Dict[str, Driver] = {}
         for protocol, driver, prefixes in self.driver_slots:
+            if protocol:
+                slot_by_protocol.setdefault(protocol, driver)
             for p in self.point_dict.points:
                 if p.point_id in routing:
                     continue
                 if not prefixes or any(p.device_id.startswith(pre) for pre in prefixes):
                     routing[p.point_id] = driver
+        # 点位级协议覆盖：优先于前缀路由
+        for p in self.point_dict.points:
+            if p.protocol and p.protocol in slot_by_protocol:
+                routing[p.point_id] = slot_by_protocol[p.protocol]
         return routing
 
     # ------------------------------------------------------------------

core/edge-gateway/config/point_dict.example.csv

@@ -1,11 +1,11 @@
-device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode
+device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
-CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
-CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres
+CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,opcua
-CLF-01,CLF-01.FEED,进料量,t/h,float,1000,true,ns=2;s=CLF.Feed
+CLF-01,CLF-01.FEED,进料量,t/h,float,1000,true,ns=2;s=CLF.Feed,opcua
-CLF-02,CLF-02.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF2.Temp
+CLF-02,CLF-02.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF2.Temp,
-CLF-02,CLF-02.RUN,运行状态,%,bool,1000,true,ns=2;s=CLF2.Run
+CLF-02,CLF-02.RUN,运行状态,%,bool,1000,true,ns=2;s=CLF2.Run,
-S7-01,S7-01.PUMP_A,泵A频率,Hz,float,1000,true,DB100.0.0
+S7-01,S7-01.PUMP_A,泵A频率,Hz,float,1000,true,DB100.0.0,s7
-S7-01,S7-01.PUMP_B,泵B频率,Hz,float,1000,true,DB100.4.0
+S7-01,S7-01.PUMP_B,泵B频率,Hz,float,1000,true,DB100.4.0,s7
-W-01,W-01.WT,称重值,kg,float,1000,true,holding:40001
+W-01,W-01.WT,称重值,kg,float,1000,true,holding:40001,weighing
-E-01,E-01.PWR,电表功率,kW,float,1000,true,holding:40010
+E-01,E-01.PWR,电表功率,kW,float,1000,true,holding:40010,energy
-E-01,E-01.KWH,电表累计,kWh,float,1000,true,holding:40012
+E-01,E-01.KWH,电表累计,kWh,float,1000,true,holding:40012,energy

core/edge-gateway/point_dict/__init__.py

@@ -2,7 +2,7 @@
 """点位字典（Point Dictionary）模块：CSV schema + 加载 + 自动校验。"""
 
 from .loader import Point, PointDict, load_point_dict_csv
-from .schema import CSV_HEADERS, VALID_DATA_TYPES, VALID_UNITS
+from .schema import CSV_HEADERS, VALID_DATA_TYPES, VALID_PROTOCOLS, VALID_UNITS
 from .validator import ValidationIssue, ValidationReport, validate_point_dict, validate_point_dict_file
 
 __all__ = [
@@ -11,6 +11,7 @@ __all__ = [
     "load_point_dict_csv",
     "CSV_HEADERS",
     "VALID_DATA_TYPES",
+    "VALID_PROTOCOLS",
     "VALID_UNITS",
     "ValidationIssue",
     "ValidationReport",

core/edge-gateway/point_dict/loader.py

@@ -26,6 +26,7 @@ class Point:
     sample_rate: int
     quality_code: bool = True
     opc_node: Optional[str] = None
+    protocol: Optional[str] = None  # 点位级协议覆盖；空 = 按 YAML device_prefixes 路由
     row_number: int = 0  # CSV 行号（从 2 开始，表头为第 1 行），用于报错定位
 
     @property
@@ -88,6 +89,7 @@ def load_point_dict_csv(path: str) -> PointDict:
                     sample_rate=sample_rate,
                     quality_code=_to_bool(row["qualityCode"]) if (row.get("qualityCode") or "").strip() else True,
                     opc_node=((row.get("opcNode") or "").strip() or None),
+                    protocol=((row.get("protocol") or "").strip().lower() or None),
                     row_number=row_number,
                 )
             )

core/edge-gateway/point_dict/schema.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 """点位字典 CSV schema —— 对齐 PRD 5.1「边缘采集网关」字段规范表。
 
-字段表（PRD 5.1）:
+字段表（PRD 5.1 + issue #21「协议」维度）:
     device_id    string  必填，唯一  设备编号，如 CLF-01
     point_id     string  必填，唯一  测点编号，如 CLF-01.TEMP
     name         string  必填        中文名，如 炉温
@@ -10,6 +10,14 @@
     sampleRate   int     必填, >0    采集周期(ms)
     qualityCode  bool    默认 true   是否启用质量码
     opcNode      string  选填        OPC UA 节点路径
+    protocol     enum    选填        采集协议（opcua/s7/modbus/weighing/energy/simulator）。
+                                    空 = 由 gateway.yaml drivers 段按设备前缀路由（模板级默认）。
+
+协议维度说明（issue #21）:
+    点位字典 CSV schema 覆盖「点位/设备/量纲/采样率/协议」五维。
+    protocol 列提供**点位级协议覆盖**：同一模板内混接多种协议时，
+    可在 CSV 中逐点位/逐设备显式指定协议；为空时保持模板级
+    （gateway.yaml drivers 段 device_prefixes）路由，向后兼容。
 """
 from __future__ import annotations
 
@@ -24,13 +32,16 @@ VALID_UNITS: List[str] = [
 # 合法数据类型集合
 VALID_DATA_TYPES: List[str] = ["float", "int", "bool"]
 
+# 合法采集协议集合（与 drivers/__init__.py 注册表对齐）
+VALID_PROTOCOLS: List[str] = ["opcua", "s7", "modbus", "weighing", "energy", "simulator"]
+
 # 必填字段
 REQUIRED_FIELDS: List[str] = ["device_id", "point_id", "name", "unit", "dataType", "sampleRate"]
 
 # CSV 表头（列顺序固定，便于实施工程师对照 DCS 点表填写）
 CSV_HEADERS: List[str] = [
     "device_id", "point_id", "name", "unit", "dataType", "sampleRate",
-    "qualityCode", "opcNode",
+    "qualityCode", "opcNode", "protocol",
 ]
 
 
@@ -44,3 +55,8 @@ def is_valid_unit(unit: str) -> bool:
 def is_valid_data_type(dtype: str) -> bool:
     """数据类型合法性校验。"""
     return dtype in VALID_DATA_TYPES
+
+
+def is_valid_protocol(protocol: str) -> bool:
+    """采集协议合法性校验（小写，须在驱动注册表内）。"""
+    return protocol in VALID_PROTOCOLS

core/edge-gateway/point_dict/validator.py

@@ -1,13 +1,15 @@
 # -*- coding: utf-8 -*-
 """点位字典自动校验器。
 
-校验维度（对齐 issue #3 与 PRD 5.1）：
+校验维度（对齐 issue #3 / #21 与 PRD 5.1）：
 1. 缺失字段：必填列缺失 / 必填值为空；
 2. 量纲：unit 不在合法量纲集合；
 3. 重复点号：point_id 重复（同一测点被定义两次）；
 4. 采样率：sampleRate 必须为正整数；
 5. 数据类型：dataType 必须为 float/int/bool；
-6. 表头：CSV 缺少必填列。
+6. 表头：CSV 缺少必填列；
+7. 协议：protocol 若填写必须为合法协议（opcua/s7/modbus/...），
+   为空表示按模板配置（YAML drivers 段）路由，不做约束。
 
 注：同一设备下多个测点行是正常场景（如 CLF-01 的炉温/炉压），
 因此 device_id 重复不做行级报错。
@@ -102,7 +104,13 @@ def validate_point_dict(point_dict: PointDict, headers: List[str]) -> Validation
                 ValidationIssue(code="bad_sample_rate", row=p.row_number,
                                 message=f"第{p.row_number}行 sampleRate 必须为正整数(ms)，当前: {p.sample_rate}")
             )
-        # 5) 重复点号（同一测点被定义两次）
+        # 5) 协议（可选列：填写则必须合法；空 = 按模板 YAML 路由）
+        if p.protocol and not schema.is_valid_protocol(p.protocol):
+            report.issues.append(
+                ValidationIssue(code="bad_protocol", row=p.row_number,
+                                message=f"第{p.row_number}行 非法协议: '{p.protocol}'（合法值见 schema.VALID_PROTOCOLS）")
+            )
+        # 6) 重复点号（同一测点被定义两次）
         if p.point_id:
             if p.point_id in seen_point_ids:
                 report.issues.append(

core/edge-gateway/tests/test_engine_metrics.py

@@ -30,6 +30,19 @@ def make_point_dict(n: int = 10) -> PointDict:
     return PointDict(points)
 
 
+def make_point_dict_with_protocol() -> PointDict:
+    """混接协议点位集：CLF 设备点位级指定 opcua，其余走前缀路由。"""
+    points = [
+        Point(device_id="CLF-01", point_id="CLF-01.TEMP", name="炉温", unit="℃",
+              data_type="float", sample_rate=1000, quality_code=True, protocol="opcua", row_number=2),
+        Point(device_id="CLF-01", point_id="CLF-01.PRES", name="炉压", unit="kPa",
+              data_type="float", sample_rate=1000, quality_code=True, row_number=3),
+        Point(device_id="S7-01", point_id="S7-01.PUMP_A", name="泵A频率", unit="Hz",
+              data_type="float", sample_rate=1000, quality_code=True, protocol="s7", row_number=4),
+    ]
+    return PointDict(points)
+
+
 class EngineMetricsTest(unittest.TestCase):
     def setUp(self):
         self._tmp = tempfile.mkdtemp()
@@ -56,6 +69,32 @@ class EngineMetricsTest(unittest.TestCase):
         # spool 已落盘
         self.assertEqual(spool.total_pending(), 10)
 
+    def test_protocol_column_overrides_prefix_routing(self):
+        """点位级 protocol 列优先于 YAML 前缀路由（issue #21）。"""
+        pd = make_point_dict_with_protocol()
+        spool = SpoolStore(os.path.join(self._tmp, "spool"))
+        metrics = HealthMetrics()
+        opcua_slot = SimulatorDriver()
+        s7_slot = SimulatorDriver()
+        engine = CollectorEngine(
+            point_dict=pd,
+            driver_slots=[
+                ("opcua", opcua_slot, ["CLF"]),
+                ("s7", s7_slot, ["S7"]),
+            ],
+            spool=spool,
+            metrics=metrics,
+            interval_ms=1000,
+        )
+        routing = engine._point_to_driver
+        # 点位级指定协议 → 精确匹配对应驱动实例
+        self.assertIs(routing["CLF-01.TEMP"], opcua_slot)
+        self.assertIs(routing["S7-01.PUMP_A"], s7_slot)
+        # 未指定协议 → 按设备前缀路由
+        self.assertIs(routing["CLF-01.PRES"], opcua_slot)
+        got = engine.collect_once()
+        self.assertEqual(got, 3)
+
     def test_failed_driver_counts_round(self):
         """驱动抛异常 → 该轮记为失败轮次，可用性下降。"""
 

core/edge-gateway/tests/test_loader.py

@@ -6,9 +6,9 @@ import unittest
 
 from point_dict.loader import load_point_dict_csv
 
-GOOD_CSV = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode
+GOOD_CSV = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
-CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
-CLF-01,CLF-01.PRES,炉压,kPa,float,1000,,ns=2;s=CLF.Pres
+CLF-01,CLF-01.PRES,炉压,kPa,float,1000,,ns=2;s=CLF.Pres,
 """
 
 
@@ -34,8 +34,15 @@ class LoaderTest(unittest.TestCase):
         self.assertEqual(p0.sample_rate, 1000)
         self.assertTrue(p0.quality_code)
         self.assertEqual(p0.opc_node, "ns=2;s=CLF.Temp")
+        self.assertEqual(p0.protocol, "opcua")
         self.assertEqual(p0.row_number, 2)
 
+    def test_protocol_blank_defaults_none(self):
+        """protocol 列留空时默认 None（走模板 YAML 前缀路由）。"""
+        path = self._write(GOOD_CSV)
+        pd = load_point_dict_csv(path)
+        self.assertIsNone(pd.points[1].protocol)
+
     def test_quality_code_default_true(self):
         """qualityCode 列留空时默认 true。"""
         path = self._write(GOOD_CSV)

core/edge-gateway/tests/test_validator.py

@@ -6,9 +6,9 @@ import unittest
 
 from point_dict.validator import validate_point_dict_file
 
-GOOD_CSV = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode
+GOOD_CSV = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
-CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
-CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres
+CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,
 """
 
 
@@ -86,6 +86,23 @@ CLF-01,CLF-01.TEMP,炉温,float,1000,true,
         codes = [i.code for i in report.issues]
         self.assertIn("missing_column", codes)
 
+    def test_bad_protocol(self):
+        """protocol 列填写了未注册协议时校验失败。"""
+        csv = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,http
+"""
+        report = validate_point_dict_file(self._write(csv))
+        codes = [i.code for i in report.issues]
+        self.assertIn("bad_protocol", codes)
+
+    def test_protocol_blank_ok(self):
+        """protocol 列为空时校验通过（模板级 YAML 路由）。"""
+        csv = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
+CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,
+"""
+        report = validate_point_dict_file(self._write(csv))
+        self.assertTrue(report.ok, report.summary())
+
     def test_multiple_issues_aggregated(self):
         csv = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode
 CLF-01,CLF-01.TEMP,炉温,摄氏度,float,0,true,