From 4457b531702c5e8029b4b9cbb061827d68a1405d Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:28:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(#63):=20=E7=82=B9=E4=BD=8D=E5=AD=97?= =?UTF-8?q?=E5=85=B8=20CSV=20=E5=AF=BC=E5=85=A5+=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E6=A0=A1=E9=AA=8C=EF=BC=88=E5=A4=8D=E7=94=A8=E5=86=85=E6=A0=B8?= =?UTF-8?q?=20point=5Fdict=20=E6=A0=A1=E9=AA=8C=E5=99=A8=EF=BC=8C=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=20OPC=20=E8=8A=82=E7=82=B9/=E6=A8=A1=E6=9D=BF?= =?UTF-8?q?=E7=BA=A7=E6=A0=A1=E9=AA=8C=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/template-console/point_importer.py | 318 ++++++++++++++++++ .../tests/test_point_importer.py | 228 +++++++++++++ 2 files changed, 546 insertions(+) create mode 100644 core/template-console/point_importer.py create mode 100644 core/template-console/tests/test_point_importer.py diff --git a/core/template-console/point_importer.py b/core/template-console/point_importer.py new file mode 100644 index 0000000..667f645 --- /dev/null +++ b/core/template-console/point_importer.py @@ -0,0 +1,318 @@ +# -*- coding: utf-8 -*- +"""⑤.7 点位字典 CSV 导入 + 自动校验页面 —— issue #63 / PRD ⑤.7。 + +配置台的"导入页面"要解决：实施工程师拿着 DCS 点表（Excel 转 CSV）粘进配置台， +**一次性看到所有问题**（表头错/量纲错/重复点号/采样率非正/协议非法/OPC 节点 +格式错），而不是改一条报一条。本模块是导入页面的后端引擎。 + +**复用而非重造**：点位字典的 schema/加载/校验（量纲/数据类型/采样率/重复点号/ +协议）已由内核 ``core/edge-gateway/point_dict``（loader/validator/schema）实现 +并被边缘网关正式使用。本模块在其基础上增加**配置台专属**校验维度： + +1. OPC 节点格式校验（OPC UA 节点须形如 ``ns=<数字>;s=<名>`` 或 PLC 寄存器 + ``holding:<数字>`` / ``coil:<数字>``，与 simulator/opcua 驱动约定一致）； +2. 表头列顺序严格对齐（实施工程师照表填列，列序错位是高频错误）； +3. 行级结果聚合为 ``ImportRowIssue``（行号 + 严重级别 + 问题 + 修复建议）， + 供配置台前端逐行渲染、按严重级别过滤； +4. 模板选择（resin/ti）：不同行业模板的合法量纲集合不同（如树脂含 rpm/mmol·g⁻¹）， + 导入时按模板切换校验基线。 + +零运行时依赖：复用 ``point_dict`` 子包（纯标准库 csv/dataclass）。 +""" +from __future__ import annotations + +import os +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, List, Optional, Tuple + +# 复用内核 edge-gateway 的点位字典加载/校验（schema/validator/loader） +# 在 tests/_bootstrap.py 中已把 core/edge-gateway 加入 sys.path， +# 故此处以顶层包 point_dict 引用（与 edge-gateway 自身测试一致）。 +from point_dict import ( # noqa: E402 + CSV_HEADERS, + VALID_PROTOCOLS, + VALID_UNITS, + Point, + PointDict, + load_point_dict_csv, +) +from point_dict import schema as _pd_schema # noqa: E402 + + +class Severity(str, Enum): + """问题严重级别（配置台前端据此着色/过滤）。""" + + ERROR = "error" # 阻断：不修复无法入库 + WARN = "warn" # 警告：可入库但建议复核（如 OPC 节点为空） + + +class TemplateKind(str, Enum): + """行业模板（决定合法量纲等校验基线）。""" + + TI = "ti" # 氯化/化工通用（templates/ti-cl4） + RESIN = "resin" # 吸附树脂（templates/resin，含 rpm / mmol/g） + + +# OPC 节点格式（与 simulator/opcua/s7 驱动约定一致）： +# ns=2;s=CLF.Temp —— OPC UA 节点（namespace + 字符串 id） +# holding:40010 / coil:1 —— Modbus 寄存器（保持/线圈 + 地址） +_OPC_UA_RE = re.compile(r"^ns=\d+;s=[^\s,]+$") +_MODBUS_RE = re.compile(r"^(holding|coil|input|discrete):(\d+)$") + + +def _unit_set(template: TemplateKind) -> set: + """按模板返回合法量纲集合（树脂含 rpm/mmol·g⁻¹ 等扩展）。""" + base = set(VALID_UNITS) + if template == TemplateKind.RESIN: + # VALID_UNITS 已含树脂扩展（rpm/mmol/g），直接复用 + return base + # ti 模板：移除树脂专属量纲，避免化工模板误用树脂量纲 + base.discard("rpm") + base.discard("mmol/g") + return base + + +@dataclass +class ImportRowIssue: + """导入页一行的问题（行号 + 严重级别 + 问题 + 修复建议，可解释）。""" + + row: int # CSV 行号（表头=1，数据从 2 起） + severity: Severity + code: str # 错误码（对齐 point_dict.validator 的 code + 本模块扩展） + message: str # 问题描述 + suggestion: str = "" # 修复建议（供配置台"一键修复"提示） + + def to_dict(self) -> dict: + return { + "row": self.row, "severity": self.severity.value, + "code": self.code, "message": self.message, + "suggestion": self.suggestion, + } + + +@dataclass +class ImportReport: + """导入校验报告（配置台导入页面数据模型）。""" + + template: TemplateKind + total_rows: int = 0 # 数据行数 + issues: List[ImportRowIssue] = field(default_factory=list) + loaded_points: int = 0 # 成功加载的点数 + file_path: str = "" + + @property + def ok(self) -> bool: + """无 ERROR 级问题即可入库（WARN 不阻断）。""" + return not any(i.severity == Severity.ERROR for i in self.issues) + + @property + def error_count(self) -> int: + return sum(1 for i in self.issues if i.severity == Severity.ERROR) + + @property + def warn_count(self) -> int: + return sum(1 for i in self.issues if i.severity == Severity.WARN) + + def summary(self) -> str: + """人类可读汇总（配置台导入结果横幅）。""" + status = "通过" if self.ok else "未通过" + return (f"导入校验{status}：{self.loaded_points} 点 / " + f"{self.total_rows} 行，错误 {self.error_count}，警告 {self.warn_count}") + + def to_dict(self) -> dict: + return { + "template": self.template.value, + "total_rows": self.total_rows, + "loaded_points": self.loaded_points, + "ok": self.ok, + "error_count": self.error_count, + "warn_count": self.warn_count, + "summary": self.summary(), + "issues": [i.to_dict() for i in self.issues], + } + + +# --------------------------------------------------------------------------- +# 校验扩展 +# --------------------------------------------------------------------------- + +def _validate_opc_node(point: Point) -> List[ImportRowIssue]: + """OPC 节点格式校验（配置台扩展维度）。 + + 约定（与驱动注册表对齐）： + - 协议 opcua：节点须匹配 ``ns=<数字>;s=<名>``； + - 协议 modbus：节点须匹配 ``holding/coil/input/discrete:<数字>``； + - 协议 simulator：节点可空，或任意上述格式（演示用，宽松）； + - 节点为空：WARN（可入库但运行时无法采集，建议补全）。 + """ + out: List[ImportRowIssue] = [] + node = (point.opc_node or "").strip() + if not node: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.WARN, code="empty_opc_node", + message=f"第{point.row_number}行 opcNode 为空", + suggestion="运行时无法采集，建议补全 OPC UA 节点或 PLC 寄存器地址", + )) + return out + + proto = (point.protocol or "").lower() + ok_ua = bool(_OPC_UA_RE.match(node)) + ok_mb = bool(_MODBUS_RE.match(node)) + if proto == "opcua" and not ok_ua: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.ERROR, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 不符合 OPC UA " + f"格式 ns=;s="), + suggestion="示例：ns=2;s=CLF.Temp", + )) + elif proto == "modbus" and not ok_mb: + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.ERROR, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 不符合 Modbus " + f"格式 holding/coil/input/discrete:"), + suggestion="示例：holding:40010", + )) + elif proto not in ("opcua", "modbus") and not (ok_ua or ok_mb): + # simulator/s7/... 节点为空已 WARN；非空但格式都不符则 WARN（宽松） + out.append(ImportRowIssue( + row=point.row_number, severity=Severity.WARN, code="bad_opc_node", + message=(f"第{point.row_number}行 opcNode '{node}' 既非 OPC UA 也非 " + f"Modbus 格式"), + suggestion="确认节点格式或清空（演示协议可空）", + )) + return out + + +def _validate_header_order(headers: List[str]) -> List[ImportRowIssue]: + """表头列顺序严格对齐（列序错位是实施工程师高频错误）。""" + out: List[ImportRowIssue] = [] + if not headers: + out.append(ImportRowIssue( + row=1, severity=Severity.ERROR, code="empty_header", + message="CSV 缺少表头行", + suggestion=f"表头应为：{','.join(CSV_HEADERS)}", + )) + return out + missing = [h for h in CSV_HEADERS if h not in headers] + for h in missing: + out.append(ImportRowIssue( + row=1, severity=Severity.ERROR, code="missing_column", + message=f"表头缺少必填列：{h}", + suggestion=f"补列 {h}（完整表头：{','.join(CSV_HEADERS)}）", + )) + if headers[: len(CSV_HEADERS)] != CSV_HEADERS and not missing: + out.append(ImportRowIssue( + row=1, severity=Severity.WARN, code="bad_column_order", + message=f"表头列顺序与标准不一致：{headers}", + suggestion=f"标准顺序：{','.join(CSV_HEADERS)}", + )) + return out + + +def _convert_validator_issues(report: "object", severity_for: Dict[str, Severity]) -> List[ImportRowIssue]: + """把内核 validator.ValidationReport.issues 转成 ImportRowIssue。""" + out: List[ImportRowIssue] = [] + for it in getattr(report, "issues", []): + sev = severity_for.get(it.code, Severity.ERROR) + out.append(ImportRowIssue( + row=it.row, severity=sev, code=it.code, message=it.message, + )) + return out + + +# 内核 validator 错误码 → 严重级别映射 +_SEVERITY_MAP: Dict[str, Severity] = { + "missing_column": Severity.ERROR, + "missing_field": Severity.ERROR, + "bad_unit": Severity.ERROR, + "bad_data_type": Severity.ERROR, + "bad_sample_rate": Severity.ERROR, + "bad_protocol": Severity.ERROR, + "dup_point": Severity.ERROR, +} + + +# --------------------------------------------------------------------------- +# 导入入口 +# --------------------------------------------------------------------------- + +def import_csv( + path: str, + template: TemplateKind = TemplateKind.TI, + extra_unit_check: bool = True, +) -> Tuple[PointDict, ImportReport]: + """导入并校验点位字典 CSV（配置台导入页面后端入口）。 + + Args: + path: CSV 文件路径（UTF-8，9 列表头）； + template: 行业模板（决定合法量纲集合，resin/ti）； + extra_unit_check: 是否按模板收窄量纲集合做额外校验。 + + Returns: + (PointDict, ImportReport)：加载的点位模型 + 校验报告。 + 报告 ``ok`` 为 True 即可入库；WARN 不阻断。 + """ + report = ImportReport(template=template, file_path=path) + + # 1) 表头校验（先读表头行） + import csv as _csv + with open(path, "r", encoding="utf-8-sig") as fh: + reader = _csv.reader(fh) + rows = list(reader) + headers = [c.strip() for c in rows[0]] if rows else [] + report.issues.extend(_validate_header_order(headers)) + + # 2) 加载 + 内核校验（量纲/数据类型/采样率/重复点号/协议） + point_dict = load_point_dict_csv(path) + report.loaded_points = len(point_dict) + report.total_rows = len(point_dict.points) + + from point_dict.validator import validate_point_dict + kernel_report = validate_point_dict(point_dict, headers) + report.issues.extend(_convert_validator_issues(kernel_report, _SEVERITY_MAP)) + + # 3) 模板级量纲收窄（resin 才允许 rpm/mmol·g⁻¹） + if extra_unit_check: + allowed_units = _unit_set(template) + for p in point_dict.points: + if p.unit and p.unit not in allowed_units: + # 内核 validator 已按全集校验过；这里只补充模板级差异提示 + if p.unit in ("rpm", "mmol/g") and template == TemplateKind.TI: + report.issues.append(ImportRowIssue( + row=p.row_number, severity=Severity.ERROR, + code="template_unit_mismatch", + message=(f"第{p.row_number}行 量纲 '{p.unit}' 为树脂模板专属，" + f"当前导入的是 {template.value} 模板"), + suggestion="切换模板为 resin，或修正量纲", + )) + + # 4) OPC 节点格式校验（配置台扩展） + for p in point_dict.points: + report.issues.extend(_validate_opc_node(p)) + + # 行号排序，便于配置台逐行展示 + report.issues.sort(key=lambda i: (i.row, i.code)) + return point_dict, report + + +def import_csv_string( + content: str, + template: TemplateKind = TemplateKind.TI, + encoding: str = "utf-8", +) -> Tuple[PointDict, ImportReport]: + """从 CSV 文本导入（配置台粘贴框场景，落临时文件后复用 import_csv）。""" + import tempfile + tmp = tempfile.NamedTemporaryFile( + mode="w", encoding=encoding, suffix=".csv", delete=False) + try: + tmp.write(content) + tmp.flush() + tmp.close() + return import_csv(tmp.name, template=template) + finally: + try: + os.unlink(tmp.name) + except OSError: + pass diff --git a/core/template-console/tests/test_point_importer.py b/core/template-console/tests/test_point_importer.py new file mode 100644 index 0000000..a2553f9 --- /dev/null +++ b/core/template-console/tests/test_point_importer.py @@ -0,0 +1,228 @@ +# -*- coding: utf-8 -*- +"""点位字典 CSV 导入 + 自动校验测试（issue #63）。 + +覆盖： +1. 合法 CSV 导入通过（ti / resin 两套模板）； +2. 表头校验（缺失列 / 列序错位）； +3. 内核校验复用（量纲/数据类型/采样率/重复点号/协议）； +4. OPC 节点格式校验（opcua/modbus/空）； +5. 模板级量纲收窄（rpm 仅 resin 允许）； +6. 报告 ok/汇总/字典化 + 粘贴框入口。 +""" +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.point_importer import ( # noqa: E402 + ImportReport, + ImportRowIssue, + Severity, + TemplateKind, + import_csv, + import_csv_string, +) + +GOOD_TI = """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,opcua +""" + +GOOD_RESIN = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol +R-801,R-801.TEMP,反应釜温度,℃,float,1000,true,ns=2;s=R801.Temp,opcua +R-801,R-801.AGIT,搅拌转速,rpm,float,1000,true,ns=2;s=R801.Agit,opcua +""" + +BAD_MULTI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol +CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,badnode,opcua +CLF-01,CLF-01.TEMP,炉压,kPa,badtype,0,true,ns=2;s=CLF.Pres,opcua +CLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus +""" + + +class _TmpCsv: + """临时 CSV 文件助手。""" + + def __init__(self, content): + self._tmp = tempfile.mkdtemp() + self.path = os.path.join(self._tmp, "points.csv") + with open(self.path, "w", encoding="utf-8") as fh: + fh.write(content) + + def cleanup(self): + import shutil + shutil.rmtree(self._tmp, ignore_errors=True) + + +class GoodImportTest(unittest.TestCase): + """合法 CSV 导入。""" + + def test_good_ti_imports_ok(self): + f = _TmpCsv(GOOD_TI) + try: + pd, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertTrue(rep.ok, rep.summary()) + self.assertEqual(rep.loaded_points, 2) + self.assertEqual(rep.error_count, 0) + finally: + f.cleanup() + + def test_good_resin_imports_ok_with_rpm(self): + f = _TmpCsv(GOOD_RESIN) + try: + pd, rep = import_csv(f.path, template=TemplateKind.RESIN) + self.assertTrue(rep.ok, rep.summary()) + # rpm 在 resin 模板合法 + self.assertEqual(rep.error_count, 0) + finally: + f.cleanup() + + def test_report_summary_and_dict(self): + f = _TmpCsv(GOOD_TI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("通过", rep.summary()) + d = rep.to_dict() + self.assertTrue(d["ok"]) + self.assertEqual(d["template"], "ti") + self.assertEqual(d["loaded_points"], 2) + finally: + f.cleanup() + + +class HeaderValidationTest(unittest.TestCase): + """表头校验。""" + + def test_missing_column_is_error(self): + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertFalse(rep.ok) + codes = [i.code for i in rep.issues if i.row == 1] + self.assertIn("missing_column", codes) + finally: + f.cleanup() + + def test_wrong_column_order_is_warn(self): + # 列齐全但顺序错（name 提前）→ WARN，不阻断 + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,protocol,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,opcua,ns=2;s=CLF.Temp\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("bad_column_order", [i.code for i in rep.issues]) + finally: + f.cleanup() + + +class KernelValidationTest(unittest.TestCase): + """复用内核校验（量纲/数据类型/采样率/重复点号）。""" + + def test_dup_point_detected(self): + bad = GOOD_TI + "CLF-01,CLF-01.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF.Temp2,opcua\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertFalse(rep.ok) + self.assertIn("dup_point", [i.code for i in rep.issues]) + finally: + f.cleanup() + + def test_bad_data_type_and_sample_rate(self): + f = _TmpCsv(BAD_MULTI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + codes = [i.code for i in rep.issues] + self.assertIn("bad_data_type", codes) + self.assertIn("bad_sample_rate", codes) + finally: + f.cleanup() + + def test_bad_protocol_detected(self): + bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,unknownproto\n" + f = _TmpCsv(bad) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertIn("bad_protocol", [i.code for i in rep.issues]) + finally: + f.cleanup() + + +class OpcNodeValidationTest(unittest.TestCase): + """OPC 节点格式校验（配置台扩展维度）。""" + + def test_bad_opcua_node_is_error(self): + # BAD_MULTI 第1行 opcNode=badnode 协议 opcua → ERROR + f = _TmpCsv(BAD_MULTI) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + opc_issues = [i for i in rep.issues if i.code == "bad_opc_node"] + self.assertTrue(any(i.severity == Severity.ERROR for i in opc_issues)) + finally: + f.cleanup() + + def test_valid_modbus_node_ok(self): + # BAD_MULTI 第3行 holding:40010 modbus → 不报 bad_opc_node + f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus\n") + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + self.assertNotIn("bad_opc_node", [i.code for i in rep.issues + if i.severity == Severity.ERROR]) + finally: + f.cleanup() + + def test_empty_opc_node_is_warn(self): + f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,simulator\n") + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + empties = [i for i in rep.issues if i.code == "empty_opc_node"] + self.assertEqual(len(empties), 1) + self.assertEqual(empties[0].severity, Severity.WARN) + # 警告不阻断 + self.assertTrue(rep.ok) + finally: + f.cleanup() + + +class TemplateUnitTest(unittest.TestCase): + """模板级量纲收窄。""" + + def test_rpm_rejected_in_ti_template(self): + f = _TmpCsv(GOOD_RESIN) + try: + _, rep = import_csv(f.path, template=TemplateKind.TI) + # rpm 是树脂专属，ti 模板应报 template_unit_mismatch + self.assertIn("template_unit_mismatch", [i.code for i in rep.issues]) + self.assertFalse(rep.ok) + finally: + f.cleanup() + + def test_rpm_allowed_in_resin_template(self): + f = _TmpCsv(GOOD_RESIN) + try: + _, rep = import_csv(f.path, template=TemplateKind.RESIN) + self.assertNotIn("template_unit_mismatch", [i.code for i in rep.issues]) + self.assertTrue(rep.ok, rep.summary()) + finally: + f.cleanup() + + +class ImportStringTest(unittest.TestCase): + """粘贴框入口（import_csv_string）。""" + + def test_import_from_string(self): + pd, rep = import_csv_string(GOOD_TI, template=TemplateKind.TI) + self.assertTrue(rep.ok) + self.assertEqual(len(pd), 2) + + def test_import_string_bad_csv(self): + bad = "device_id,point_id\nCLF-01,CLF-01.TEMP\n" # 缺列 + _, rep = import_csv_string(bad, template=TemplateKind.TI) + self.assertFalse(rep.ok) + + +if __name__ == "__main__": + unittest.main()