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

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
-CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
-CLF-01,CLF-01.PRES,炉压,kPa,float,1000,,ns=2;s=CLF.Pres
+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,opcua
+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
-CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
-CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres
+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,opcua
+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,