diff --git a/main.py b/main.py index b7efb87..6d41f7c 100644 --- a/main.py +++ b/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}") - data_dict = read_excel_data(sample_path, return_object=False) - pass + print(f"读取文件: {sample_path}") + data_dict = read_excel_data("new_data.xlsx", return_object=False) # 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__": diff --git a/pyproject.toml b/pyproject.toml index f1f92d9..fb9772d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,7 @@ dependencies = [ dev = [ "pytest>=7.0.0", "mypy>=1.0.0", + "ruff>=0.1.0", ] [build-system] @@ -27,4 +28,31 @@ python_functions = ["test_*"] python_version = "3.10" warn_return_any = true warn_unused_configs = true -disallow_untyped_defs = true \ No newline at end of file +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" \ No newline at end of file diff --git a/src/core/__init__.py b/src/core/__init__.py index b38dc75..7e5beba 100644 --- a/src/core/__init__.py +++ b/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", diff --git a/src/core/base.py b/src/core/base.py index 69f9fa6..51aff39 100644 --- a/src/core/base.py +++ b/src/core/base.py @@ -14,7 +14,7 @@ class DataLog: time: float = 0.0 value: str = "" - + @dataclass class SignalData: """信号数据封装 diff --git a/src/core/mil_create_data_excel.py b/src/core/mil_create_data_excel.py index 0798f5e..435886f 100644 --- a/src/core/mil_create_data_excel.py +++ b/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 openpyxl import Workbook, load_workbook +from .exceptions import CaseDataError, ExcelWriteError +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) - # new_wb.save(generate_path) + _write_excel_data(sheet, datas_dict[name], source_row) 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) -> dict: +def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict: """分析用例字典，将用例中的信号名替换为信号值 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]) - __analysis_step(sheet_dict[case]["step"], datas_dict[case]) + _init_data_log(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,11 +171,11 @@ 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: data_dict: 数据字典 """ for name in data_dict.keys(): - data_dict[name]["datalog"] = data_dict[name]["datalog"][:1] + data_dict[name]["datalog"] = data_dict[name]["datalog"][:1] \ No newline at end of file diff --git a/src/core/mil_read_case_excel.py b/src/core/mil_read_case_excel.py index 0a89f32..ba76c85 100644 --- a/src/core/mil_read_case_excel.py +++ b/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 .exceptions import ExcelReadError, ExcelFormatError, CaseDataError, ExcelWriteError -from openpyxl import Workbook, load_workbook +from openpyxl import 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 -COL_INDEX_STATUS = 2 -COL_INDEX_ACTION = 3 -COL_INDEX_NAME = 4 -COL_INDEX_TIME = 5 +CaseDict = dict[str, Any] + + +class CaseColumns: + """用例 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: - new_head = sheet.cell(row=row, column=COL_INDEX_TITLE).value + if sheet.cell(row=row, column=CaseColumns.TITLE).value is not None: + 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_time = sheet.cell(row, COL_INDEX_TIME).value + step_name = sheet.cell(row, CaseColumns.NAME).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: - version = sheet.cell(row=row, column=COL_INDEX_TITLE).value + while sheet.cell(row=row, column=CaseColumns.TITLE).value is not None: + 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, @@ -206,5 +212,4 @@ def __analysis_action( action_dict[matched_key] = signal_value - return action_dict - + return action_dict \ No newline at end of file diff --git a/src/core/mil_read_data_excel.py b/src/core/mil_read_data_excel.py index 9441c56..4fbc0dd 100644 --- a/src/core/mil_read_data_excel.py +++ b/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, diff --git a/src/core/mil_update_excel.py b/src/core/mil_update_excel.py new file mode 100644 index 0000000..fbcfc4a --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/test_mil_case_excel.py b/tests/test_mil_case_excel.py index 09e9336..1ca9c75 100644 --- a/tests/test_mil_case_excel.py +++ b/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, - __analysis_action, + _get_template_version, + _analysis_action, ) from src.core.mil_create_data_excel import ( create_excel_case, - __init_data_log, - __analysis_case, - __analysis_step, + _init_data_log, + _analysis_case, + _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