优化代码，增加更新测试用例功能

main.py

@@ -1,26 +1,33 @@
 """main.py - MIL SDK 测试示例"""
 from pathlib import Path
-from src.core import setup_logging, read_excel_data,read_excel_case,create_excel_case
+
+from src.core import (
+    setup_logging,
+    read_excel_data,
+    read_excel_case,
+    create_excel_case,
+    update_case_excel,
+)
 
 
-def main():
+def main() -> None:
+    """MIL SDK 示例程序主函数"""
     setup_logging()
 
     sample_path = Path(__file__).parent / "sample.xlsx"
     case_path = Path(__file__).parent / "case.xlsx"
     excel_path = Path(__file__).parent
 
-
+    print(f"读取文件: {sample_path}")
-    # print(f"读取文件: {sample_path}")
+    data_dict = read_excel_data("new_data.xlsx", return_object=False)
-    data_dict = read_excel_data(sample_path, return_object=False)
-    pass
     # print(data_dict)
-    case_dict = read_excel_case(case_path,data_dict,True)
-    pass
-    create_excel_case(excel_path,data_dict,case_dict)
-    # print(case_dict)
-    pass
 
+    # case_dict = read_excel_case(case_path, data_dict, True)
+    # create_excel_case(excel_path, data_dict, case_dict)
+    # print(case_dict)
+
+    old_data_dict = read_excel_data("old_data.xlsx", return_object=False)
+    update_case_excel("new_case.xlsx", old_data_dict, data_dict)
 
 
 if __name__ == "__main__":

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
 dev = [
     "pytest>=7.0.0",
     "mypy>=1.0.0",
+    "ruff>=0.1.0",
 ]
 
 [build-system]
@@ -28,3 +29,30 @@ python_version = "3.10"
 warn_return_any = true
 warn_unused_configs = true
 disallow_untyped_defs = true
+
+[tool.ruff]
+target-version = "py310"
+line-length = 100
+
+[tool.ruff.lint]
+select = [
+    "E",     # pycodestyle errors
+    "W",     # pycodestyle warnings
+    "F",     # pyflakes
+    "I",     # isort
+    "UP",    # pyupgrade
+    "B",     # flake8-bugbear
+    "C4",    # flake8-comprehensions
+]
+ignore = [
+    "E501",  # line too long (handled by formatter)
+]
+
+[tool.ruff.lint.isort]
+known-first-party = ["src"]
+
+[tool.ruff.format]
+quote-style = "double"
+indent-style = "space"
+skip-magic-trailing-comma = false
+line-ending = "auto"

src/core/__init__.py

@@ -14,6 +14,8 @@ from .base import DataLog, SignalData, ExcelDataResult
 from .mil_read_data_excel import read_excel_data, get_data_log, ExcelReaderConfig
 from .mil_read_case_excel import read_excel_case
 from .mil_create_data_excel import create_excel_case
+from .mil_update_excel import update_case_excel
+
 from .exceptions import (
     MILSDKError,
     ExcelReadError,
@@ -23,11 +25,6 @@ from .exceptions import (
 )
 from .logging_config import setup_logging, get_logger
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
-
 __all__ = [
     "DataLog",
     "SignalData",

src/core/mil_create_data_excel.py

@@ -11,12 +11,15 @@ import logging
 from typing import Any
 
 from .base import DataLog
-from .exceptions import ExcelReadError, ExcelFormatError, CaseDataError, ExcelWriteError
+from .exceptions import CaseDataError, ExcelWriteError
-from openpyxl import Workbook, load_workbook
+from openpyxl import Workbook
 from openpyxl.worksheet.worksheet import Worksheet
 
 logger = logging.getLogger(__name__)
 
+CaseResultDict = dict[str, Any]
+
+
 def create_excel_case(
     excel_path: str,
     data_dict: dict,
@@ -44,7 +47,7 @@ def create_excel_case(
 
     data_dict_copy = {k: v for k, v in data_dict.items()
                       if k not in ('wb', 'sheet', 'source_row')}
-    datas_dict = __analysis_case(data_dict_copy, sheets_dict)
+    datas_dict = _analysis_case(data_dict_copy, sheets_dict)
 
     case_count = 0
     for name in datas_dict.keys():
@@ -54,16 +57,14 @@ def create_excel_case(
         sheet = new_wb["Scenario1"]
 
         generate_path = f"{excel_path}/{name}.xlsx"
-        __write_excel_data(sheet, datas_dict[name], source_row)
+        _write_excel_data(sheet, datas_dict[name], source_row)
-        # new_wb.save(generate_path)
 
         logger.info(f"已生成用例: {name}，路径: {generate_path}")
 
     logger.info(f"测试用例生成完成，共 {case_count} 个用例")
 
 
-
+def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> None:
-def __write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> None:
     """写入 Excel 数据"""
     for name in data_dict.keys():
         column = data_dict[name]["column"]
@@ -93,8 +94,7 @@ def __write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> No
         sheet.cell(source_row + row - 1, column).data_type = "float"
 
 
-
+def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict:
-def __analysis_case(data_dict: dict, sheets_dict: dict) -> dict:
     """分析用例字典，将用例中的信号名替换为信号值
 
     Args:
@@ -104,7 +104,7 @@ def __analysis_case(data_dict: dict, sheets_dict: dict) -> dict:
     Returns:
         处理后的用例字典
     """
-    datas_dict = {}
+    datas_dict: CaseResultDict = {}
     for sheet_name in sheets_dict.keys():
         sheet_dict = sheets_dict[sheet_name]
 
@@ -116,12 +116,12 @@ def __analysis_case(data_dict: dict, sheets_dict: dict) -> dict:
                 continue
 
             datas_dict[case] = copy.deepcopy(data_dict)
-            __init_data_log(datas_dict[case])
+            _init_data_log(datas_dict[case])
-            __analysis_step(sheet_dict[case]["step"], datas_dict[case])
+            _analysis_step(sheet_dict[case]["step"], datas_dict[case])
     return datas_dict
 
 
-def __analysis_step(step_dict: dict, data_dict: dict) -> None:
+def _analysis_step(step_dict: dict, data_dict: dict) -> None:
     """分析步骤字典，将步骤中的信号名替换为信号值
 
     Args:
@@ -130,14 +130,14 @@ def __analysis_step(step_dict: dict, data_dict: dict) -> None:
     """
     for step_key in step_dict.keys():
         if "action" in step_dict[step_key].keys():
-            __analysis_data(step_dict[step_key]["action"], data_dict, step_dict[step_key]["time"])
+            _analysis_data(step_dict[step_key]["action"], data_dict, step_dict[step_key]["time"])
         else:
             for name in data_dict.keys():
                 datalog = data_dict[name]["datalog"]
                 datalog.append(DataLog(step_dict[step_key]["time"], datalog[-1].value))
 
 
-def __analysis_data(action_dict: dict, data_dict: dict, time: float) -> None:
+def _analysis_data(action_dict: dict, data_dict: dict, time: float) -> None:
     """分析数据字典，将数据字典中的信号名替换为信号值
 
     Args:
@@ -148,9 +148,9 @@ def __analysis_data(action_dict: dict, data_dict: dict, time: float) -> None:
     Raises:
         CaseDataError: 信号不存在
     """
-    data_log_len = None
+    data_log_len: int | None = None
     for name in action_dict.keys():
-        matched_key = None
+        matched_key: str | None = None
         for key in data_dict.keys():
             if name.lower() == key.strip().lower():
                 matched_key = key
@@ -171,7 +171,7 @@ def __analysis_data(action_dict: dict, data_dict: dict, time: float) -> None:
             datalog.append(DataLog(time, datalog[-1].value))
 
 
-def __init_data_log(data_dict: dict) -> None:
+def _init_data_log(data_dict: dict) -> None:
     """初始化数据日志
 
     Args:

src/core/mil_read_case_excel.py

@@ -13,22 +13,29 @@ Excel 模板格式约定:
     - 列 4: 步骤名称
     - 列 5: 时间
 """
-import copy
 import logging
 from typing import Any
 
-from .base import DataLog
+from openpyxl import load_workbook
-from .exceptions import ExcelReadError, ExcelFormatError, CaseDataError, ExcelWriteError
-from openpyxl import Workbook, load_workbook
 from openpyxl.worksheet.worksheet import Worksheet
 
+from .base import DataLog
+from .exceptions import ExcelReadError, ExcelFormatError, CaseDataError
+
 logger = logging.getLogger(__name__)
 
-COL_INDEX_TITLE = 1
+CaseDict = dict[str, Any]
-COL_INDEX_STATUS = 2
+
-COL_INDEX_ACTION = 3
+
-COL_INDEX_NAME = 4
+class CaseColumns:
-COL_INDEX_TIME = 5
+    """用例 Excel 列索引常量"""
+    TITLE = 1
+    STATUS = 2
+    ACTION = 3
+    NAME = 4
+    TIME = 5
+
+
 STATUS_COMPLETE = "完成测试"
 
 
@@ -36,7 +43,7 @@ def read_excel_case(
     excel_path: str,
     data_dict: dict,
     addTimeEn: bool,
-) -> dict[str, Any]:
+) -> CaseDict:
     """读取 Excel 用例模板
 
     Args:
@@ -63,7 +70,7 @@ def read_excel_case(
         raise ExcelReadError(f"无法读取Excel文件 {excel_path}: {str(e)}")
 
     try:
-        version = __get_template_version(wb["Atech-Hefei"])
+        version = _get_template_version(wb["Atech-Hefei"])
         if version is None:
             logger.error("缺少模板版本号")
             raise ExcelFormatError("Excel格式错误，缺少模板版本号")
@@ -73,7 +80,7 @@ def read_excel_case(
         logger.error(f"缺少 'Atech-Hefei' 表")
         raise ExcelFormatError(f"Excel文件 {excel_path} 格式错误，缺少 Atech-Hefei 表")
 
-    result_dict = {}
+    result_dict: CaseDict = {}
     sheet_count = 0
     for sheet in wb.worksheets:
         sheet_count += 1
@@ -88,11 +95,11 @@ def read_excel_case(
         row = 2
         old_time = 0.0
         while True:
-            if sheet.cell(row=row, column=COL_INDEX_TITLE).value is not None:
+            if sheet.cell(row=row, column=CaseColumns.TITLE).value is not None:
-                new_head = sheet.cell(row=row, column=COL_INDEX_TITLE).value
+                new_head = sheet.cell(row=row, column=CaseColumns.TITLE).value
             if new_head is None:
                 raise CaseDataError(
-                    f"Excel文件 {excel_path} 格式错误，{sheet.title} 第 {row} 行第 {COL_INDEX_TITLE} 列必须有标题名"
+                    f"Excel文件 {excel_path} 格式错误，{sheet.title} 第 {row} 行第 {CaseColumns.TITLE} 列必须有标题名"
                 )
             if new_head != old_head:
                 step_id = 0
@@ -102,20 +109,20 @@ def read_excel_case(
                 result_dict[sheet.title][new_head] = {}
                 result_dict[sheet.title][new_head]["step"] = {}
 
-                if sheet.cell(row=row, column=COL_INDEX_STATUS).value == STATUS_COMPLETE:
+                if sheet.cell(row=row, column=CaseColumns.STATUS).value == STATUS_COMPLETE:
                     result_dict[sheet.title][new_head]["enable"] = True
                 else:
                     result_dict[sheet.title][new_head]["enable"] = False
 
-            step_name = sheet.cell(row, COL_INDEX_NAME).value
+            step_name = sheet.cell(row, CaseColumns.NAME).value
-            step_time = sheet.cell(row, COL_INDEX_TIME).value
+            step_time = sheet.cell(row, CaseColumns.TIME).value
             if step_time is None:
                 break
             try:
                 step_time = float(step_time)
             except (ValueError, TypeError):
                 raise CaseDataError(
-                    f"Excel文件 {excel_path} 格式错误，{sheet.title} 第 {row} 行第 {COL_INDEX_TIME} 列时间必须是数字"
+                    f"Excel文件 {excel_path} 格式错误，{sheet.title} 第 {row} 行第 {CaseColumns.TIME} 列时间必须是数字"
                 )
 
             result_dict[sheet.title][new_head]["step"][f"step{step_id}"] = {}
@@ -125,10 +132,10 @@ def read_excel_case(
                 old_time += step_time
                 result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["time"] = old_time
 
-            strings = sheet.cell(row, COL_INDEX_ACTION).value
+            strings = sheet.cell(row, CaseColumns.ACTION).value
             if strings is not None:
                 result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["action"] = \
-                    __analysis_action(sheet, row, COL_INDEX_ACTION, strings, data_dict, excel_path)
+                    _analysis_action(sheet, row, CaseColumns.ACTION, strings, data_dict, excel_path)
             step_id += 1
             row += 1
 
@@ -136,8 +143,7 @@ def read_excel_case(
     return result_dict
 
 
-
+def _get_template_version(sheet: Worksheet) -> str | None:
-def __get_template_version(sheet: Worksheet) -> str | None:
     """获取 Excel 模板版本号
 
     Args:
@@ -148,13 +154,13 @@ def __get_template_version(sheet: Worksheet) -> str | None:
     """
     row = 2
     version = None
-    while sheet.cell(row=row, column=COL_INDEX_TITLE).value is not None:
+    while sheet.cell(row=row, column=CaseColumns.TITLE).value is not None:
-        version = sheet.cell(row=row, column=COL_INDEX_TITLE).value
+        version = sheet.cell(row=row, column=CaseColumns.TITLE).value
         row += 1
     return version
 
 
-def __analysis_action(
+def _analysis_action(
     sheet: Worksheet,
     row: int,
     column: int,
@@ -207,4 +213,3 @@ def __analysis_action(
         action_dict[matched_key] = signal_value
 
     return action_dict
-

src/core/mil_read_data_excel.py

@@ -19,6 +19,9 @@ from .exceptions import ExcelReadError, ExcelFormatError
 
 logger = logging.getLogger(__name__)
 
+SignalDict = dict[str, SignalData]
+DataDict = dict[str, Any]
+
 
 @dataclass
 class ExcelReaderConfig:
@@ -43,12 +46,11 @@ class ExcelReaderConfig:
     block_path_row: int = 4
 
 
-
 def read_excel_data(
     excel_path: str,
     return_object: bool = False,
     config: ExcelReaderConfig | None = None
-) -> dict[str, Any] | ExcelDataResult:
+) -> DataDict | ExcelDataResult:
     """读取 MIL 仿真 Excel 文件并解析信号数据
 
     Args:
@@ -102,7 +104,7 @@ def read_excel_data(
 
     logger.debug(f"找到 {config.source_header} 标记，行号: {source_row}")
 
-    signals: dict[str, SignalData] = {}
+    signals: SignalDict = {}
     column = 2
     current_header = None
     while True:
@@ -119,7 +121,7 @@ def read_excel_data(
             if name == "time":
                 column += 1
                 continue
-            if name == "Parameter:" or name == "Value" or name == "BlockPath":
+            if name in ("Parameter:", "Value", "BlockPath"):
                 column += 1
                 continue
             sig_type = sheet.cell(config.type_row, column).value
@@ -130,9 +132,6 @@ def read_excel_data(
                 datalog=get_data_log(sheet, source_row + config.data_start_row_offset, column, config.time_column)
             )
             logger.debug(f"解析信号: {name}, 数据点: {len(signals[name].datalog)}")
-        # elif current_header == config.output_header:
-        #     column += 1
-        #     continue
         else:
             break
         column += 1
@@ -147,7 +146,7 @@ def read_excel_data(
             signals=signals
         )
 
-    data: dict[str, Any] = {
+    data: DataDict = {
         "wb": wb,
         "sheet": sheet,
         "source_row": source_row
@@ -157,6 +156,7 @@ def read_excel_data(
 
     return data
 
+
 def get_data_log(
     sheet: Worksheet,
     row: int,

src/core/mil_update_excel.py

@@ -0,0 +1,122 @@
+"""MIL 测试用例 Excel 更新模块
+
+主要功能:
+- 比较新旧数据差异
+- 添加新信号列或删除旧信号列
+"""
+import logging
+from typing import Any
+
+from .base import SignalData
+from .exceptions import CaseDataError
+from openpyxl import Workbook
+from openpyxl.worksheet.worksheet import Worksheet
+
+logger = logging.getLogger(__name__)
+
+
+def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None:
+    """更新 Excel 文件中的信号列
+
+    Args:
+        filename: Excel 文件路径
+        old_data: 旧数据字典
+        new_data: 新数据字典
+
+    Raises:
+        CaseDataError: 数据类型错误
+    """
+    logger.info(f"开始更新 Excel 文件 {filename}")
+
+    old_wb = old_data.pop('wb')
+    old_sheet = old_data.pop('sheet')
+    source_row = old_data.pop('source_row')
+
+    new_data.pop('wb')
+    new_data.pop('sheet')
+    new_data.pop('source_row')
+
+    if not isinstance(old_wb, Workbook) or not isinstance(old_sheet, Worksheet):
+        logger.error("数据类型错误，Workbook 或 Worksheet 类型不匹配")
+        raise CaseDataError("数据异常，Workbook 或 Worksheet 类型错误")
+
+    add_input = _get_add_input(old_data, new_data)
+    del_input = _get_del_input(old_data, new_data)
+    datalog_len = _get_datalog_len(old_data)
+
+    for name, info in reversed(del_input.items()):
+        old_sheet.delete_cols(info['column'])
+        logger.info(f"删除第{info['column']}列\t信号名: {name}")
+
+    for name, info in add_input.items():
+        old_sheet.insert_cols(info['column'])
+
+        old_sheet.cell(1, info['column']).value = name
+        old_sheet.cell(1, info['column']).data_type = "str"
+
+        old_sheet.cell(3, info['column']).value = info['type']
+        old_sheet.cell(3, info['column']).data_type = "str"
+
+        for index in range(source_row + 1, datalog_len + source_row + 1):
+            try:
+                value = int(info['datalog'][0].value)
+                old_sheet.cell(index, info['column']).value = value
+                old_sheet.cell(index, info['column']).data_type = "int"
+            except (ValueError, TypeError):
+                value = str(info['datalog'][0].value)
+                old_sheet.cell(index, info['column']).value = value
+                old_sheet.cell(index, info['column']).data_type = "str"
+        logger.info(f"新增第{info['column']}列\t信号名: {name}")
+
+    old_wb.save(filename)
+    logger.info(f"更新 Excel 文件 {filename} 完成")
+
+
+def _get_add_input(old_data: dict, new_data: dict) -> dict[str, SignalData]:
+    """获取新增输入数据
+
+    Args:
+        old_data: 旧数据字典
+        new_data: 新数据字典
+
+    Returns:
+        新增信号字典
+    """
+    add_input: dict[str, SignalData] = {}
+
+    for name in new_data.keys():
+        if name not in old_data.keys():
+            add_input[name] = new_data[name]
+    return add_input
+
+
+def _get_del_input(old_data: dict, new_data: dict) -> dict[str, SignalData]:
+    """获取删除输入数据
+
+    Args:
+        old_data: 旧数据字典
+        new_data: 新数据字典
+
+    Returns:
+        删除信号字典
+    """
+    del_input: dict[str, SignalData] = {}
+
+    for name in old_data.keys():
+        if name not in new_data.keys():
+            del_input[name] = old_data[name]
+    return del_input
+
+
+def _get_datalog_len(old_data: dict) -> int:
+    """获取数据日志长度
+
+    Args:
+        old_data: 数据字典
+
+    Returns:
+        数据日志长度
+    """
+    for name in old_data.keys():
+        return len(old_data[name]['datalog'])
+    return 0

tests/test_mil_case_excel.py

@@ -4,14 +4,14 @@ from openpyxl import Workbook
 
 from src.core.mil_read_case_excel import (
     read_excel_case,
-    __get_template_version,
+    _get_template_version,
-    __analysis_action,
+    _analysis_action,
 )
 from src.core.mil_create_data_excel import (
     create_excel_case,
-    __init_data_log,
+    _init_data_log,
-    __analysis_case,
+    _analysis_case,
-    __analysis_step,
+    _analysis_step,
 )
 from src.core.exceptions import ExcelReadError, ExcelFormatError, CaseDataError
 from src.core.base import DataLog
@@ -69,7 +69,7 @@ def test_get_template_version():
     sheet.cell(2, 1, "v1.0.0")
     sheet.cell(3, 1, None)
 
-    version = __get_template_version(sheet)
+    version = _get_template_version(sheet)
     assert version == "v1.0.0"
 
 
@@ -78,7 +78,7 @@ def test_get_template_version_empty():
     wb = Workbook()
     sheet = wb.active
 
-    version = __get_template_version(sheet)
+    version = _get_template_version(sheet)
     assert version is None
 
 
@@ -88,7 +88,7 @@ def test_analysis_action_valid(sample_data_dict):
     sheet = wb.create_sheet("TestSheet")
     sheet.cell(1, 1, "Test")
 
-    action = __analysis_action(sheet, 1, 3, "signal1=newvalue", sample_data_dict, "test.xlsx")
+    action = _analysis_action(sheet, 1, 3, "signal1=newvalue", sample_data_dict, "test.xlsx")
     assert "signal1" in action
     assert action["signal1"] == "newvalue"
 
@@ -99,7 +99,7 @@ def test_analysis_action_multiple(sample_data_dict):
     sheet = wb.create_sheet("TestSheet")
     sheet.cell(1, 1, "Test")
 
-    action = __analysis_action(sheet, 1, 3, "signal1=v1; signal2=v2", sample_data_dict, "test.xlsx")
+    action = _analysis_action(sheet, 1, 3, "signal1=v1; signal2=v2", sample_data_dict, "test.xlsx")
     assert "signal1" in action
     assert "signal2" in action
     assert action["signal1"] == "v1"
@@ -113,7 +113,7 @@ def test_analysis_action_invalid_signal(sample_data_dict):
     sheet.cell(1, 1, "Test")
 
     with pytest.raises(CaseDataError, match="信号 .* 不存在"):
-        __analysis_action(sheet, 1, 3, "invalid_signal=value", sample_data_dict, "test.xlsx")
+        _analysis_action(sheet, 1, 3, "invalid_signal=value", sample_data_dict, "test.xlsx")
 
 
 def test_init_data_log(sample_data_dict):
@@ -122,7 +122,7 @@ def test_init_data_log(sample_data_dict):
     for name in data_copy:
         data_copy[name]["datalog"] = data_copy[name]["datalog"].copy()
 
-    __init_data_log(data_copy)
+    _init_data_log(data_copy)
 
     for name in data_copy:
         assert len(data_copy[name]["datalog"]) == 1
@@ -143,7 +143,7 @@ def test_analysis_step_with_action(sample_data_dict):
     for name in data_copy:
         data_copy[name]["datalog"] = data_copy[name]["datalog"].copy()
 
-    __analysis_step(step_dict, data_copy)
+    _analysis_step(step_dict, data_copy)
 
     assert len(data_copy["signal1"]["datalog"]) == 3
     assert data_copy["signal1"]["datalog"][-1].time == 1.5
@@ -164,14 +164,14 @@ def test_analysis_step_without_action(sample_data_dict):
         data_copy[name]["datalog"] = data_copy[name]["datalog"].copy()
 
     initial_length = len(data_copy["signal1"]["datalog"])
-    __analysis_step(step_dict, data_copy)
+    _analysis_step(step_dict, data_copy)
 
     assert len(data_copy["signal1"]["datalog"]) == initial_length + 1
 
 
 def test_analysis_case(sample_data_dict, sample_sheets_dict):
     """验证用例分析"""
-    result = __analysis_case(sample_data_dict, sample_sheets_dict)
+    result = _analysis_case(sample_data_dict, sample_sheets_dict)
 
     assert "TestCase1" in result
     assert "TestCase2" not in result
@@ -188,7 +188,7 @@ def test_analysis_case_skips_enabled_cases(sample_data_dict):
         }
     }
 
-    result = __analysis_case(sample_data_dict, sheets_dict)
+    result = _analysis_case(sample_data_dict, sheets_dict)
     assert "EnabledCase" not in result