diff --git a/mil.json b/mil.json new file mode 100644 index 0000000..1da7587 --- /dev/null +++ b/mil.json @@ -0,0 +1,8 @@ +{ + "DataPath": "", + "FilePath": "", + "AddTimeEn": true, + "GeratePath": true, + "CurrProject": "", + "ItemConfigs": {} +} \ No newline at end of file diff --git a/mil/__init__.py b/mil/__init__.py index 6de39c9..0a88dec 100644 --- a/mil/__init__.py +++ b/mil/__init__.py @@ -1,5 +1,4 @@ from .ui.mil_tool import FrameMILTool -from openpyxl import Workbook def create_mil_tool(parent=None): return FrameMILTool(parent) diff --git a/mil/core/__init__.py b/mil/core/__init__.py index 770647c..76fdfad 100644 --- a/mil/core/__init__.py +++ b/mil/core/__init__.py @@ -7,9 +7,8 @@ result = read_excel_data("simulation.xlsx") """ -import logging -from .base import DataLog, SignalData, ExcelDataResult +from .base import DataLog, SignalData, ExcelDataResult,Config from .mil_read_data_excel import read_excel_data, ExcelReaderConfig from .mil_read_case_excel import read_excel_case from .mil_create_data_excel import create_excel_case @@ -23,6 +22,9 @@ from .exceptions import ( ExcelWriteError, ) + + + __all__ = [ "DataLog", "SignalData", @@ -37,6 +39,5 @@ __all__ = [ "ExcelFormatError", "CaseDataError", "ExcelWriteError", - "setup_logging", - "get_logger", + "Config" ] \ No newline at end of file diff --git a/mil/core/base.py b/mil/core/base.py index 340a232..5c1019c 100644 --- a/mil/core/base.py +++ b/mil/core/base.py @@ -1,7 +1,105 @@ """MIL SDK 核心数据模型""" -from dataclasses import dataclass, field +import os +import json +import logging + +from dataclasses import dataclass, field, fields from typing import Any +logger = logging.getLogger(__name__) + +@dataclass +class Config: + """MIL SDK 全局配置 + + 承载测试工程相关的全局开关与路径信息，是 UI 层与 core 层之间的 + 配置传输载体。序列化使用 JSON 文件持久化。 + + Attributes: + DataPath: 仿真数据目录（Excel 原始文件所在路径） + FilePath: 当前打开的 Excel 文件路径 + AddTimeEn: 用例步骤时间是否按累加方式记录 + GeratePath: 是否为生成的用例另存新文件 + CurrProject: 当前工程名称 + ItemConfigs: 各模块/用例项的细粒度配置 + """ + DataPath:str = "" + FilePath:str = "" + AddTimeEn:bool = True + GeratePath:bool = True + CurrProject:str = "" + # dict 是可变类型，必须用 default_factory 显式实例化，避免多个 Config 共享同一对象 + ItemConfigs: dict = field(default_factory=dict) + + + def to_dict(self): + """将 Config 实例序列化为 dict，便于写入 JSON 文件。""" + return { + "DataPath":self.DataPath, + "FilePath":self.FilePath, + "AddTimeEn":self.AddTimeEn, + "GeratePath":self.GeratePath, + "CurrProject":self.CurrProject, + "ItemConfigs":self.ItemConfigs + } + + def load_config(self, config_path: str) -> "Config": + """从 JSON 文件读取配置并填充到当前实例的各个字段。 + + 解析规则： + - JSON 中存在的字段会被回写到 Config 的对应字段； + - JSON 中缺失的字段保持当前 Config 实例的默认值； + - JSON 中多余字段被忽略（向前兼容：增加字段不会破坏老配置）。 + + Args: + config_path: 配置文件路径。 + + Returns: + self：填充后的 Config 实例，便于链式调用。 + + Raises: + FileNotFoundError: 配置文件不存在。 + json.JSONDecodeError: 文件内容不是合法 JSON。 + """ + try: + # 以 UTF-8 读取 JSON，避免 Windows 默认编码带来的乱码问题 + with open(config_path, 'r', encoding='utf-8') as f: + raw = json.load(f) + except FileNotFoundError: + # 显式记录并重抛，遵循"错误必须显式处理，禁止静默失败" + logger.error(f"配置文件{config_path}未找到") + raise + + if not isinstance(raw, dict): + # 文件不合法（非 dict 根节点）→ 用一个空 dict 填充，保持实例可用 + raw = {} + + # 用反射取出本类已声明的字段名白名单，避免 JSON 脏字段污染 + allowed = {f.name for f in fields(self.__class__)} + for key, value in raw.items(): + if key in allowed: + setattr(self, key, value) + return self + + def save_config(self,config_path:str): + """将当前配置以 JSON 格式写入磁盘。 + + Args: + config_path: 配置文件路径。 + + Raises: + Exception: 写入失败时记录日志并原样抛出异常。 + """ + try: + with open(config_path,'w',encoding='utf-8') as f: + json.dump(self.to_dict(), f, indent=4, ensure_ascii=False) + except Exception as e: + logger.error(f"配置文件{config_path}写入失败{e.args}") + + + + + @dataclass class DataLog: @@ -19,10 +117,15 @@ class DataLog: class SignalData: """信号数据封装 + 描述 Excel 中某一列信号的完整信息： + - 所在列号（column） + - 列上方的属性行（attributes，例如 Signal Name / BlockPath 等） + - 时间-值采样序列（datalog） + Attributes: - signal_type: 信号类型 - column: 列索引 - datalog: 数据日志列表 + column: 信号在 Excel 工作表中的列索引（1-based） + attributes: 列上方各属性行（行号 -> 文本）的字典映射 + datalog: 时间戳-数值采样点列表 """ # signal_type: str | None = None column: int = 0 @@ -31,10 +134,10 @@ class SignalData: def to_dict(self) -> dict[str, Any]: - """转换为字典格式 + """将 SignalData 转为 dict，便于跨层传输与持久化。 Returns: - 包含 signal_type、column、datalog 的字典 + 包含 column、attributes、datalog 字段的字典。 """ return { "column": self.column, diff --git a/mil/core/exceptions.py b/mil/core/exceptions.py index 27a3cb7..34d0679 100644 --- a/mil/core/exceptions.py +++ b/mil/core/exceptions.py @@ -1,26 +1,33 @@ -"""MIL SDK 自定义异常模块""" +"""MIL SDK 自定义异常模块 + +约定： +- 所有 SDK 主动抛出的异常均继承自 MILSDKError，便于上层统一捕获； +- 异常命名体现"出错阶段"（读取 / 格式 / 用例数据 / 写入）， + 调用方只需按需捕获细分异常，或在外层兜底捕获 MILSDKError。 +""" class MILSDKError(Exception): - """MIL SDK 基础异常类""" + """MIL SDK 所有自定义异常的根类，UI 层可针对此类型统一兜底。""" pass class ExcelReadError(MILSDKError): - """Excel 文件读取错误（文件不存在、权限问题等）""" + """Excel 读取阶段错误：文件不存在、权限不足、被占用、底层解析失败等。""" pass class ExcelFormatError(MILSDKError): - """Excel 格式错误（缺少 Sheet、格式不匹配等）""" + """Excel 内容格式错误：缺少约定的工作表、缺少 'Source: Input' 标记、 + 模板版本号缺失等结构性异常。""" pass class CaseDataError(MILSDKError): - """用例数据错误（信号不存在、类型错误等）""" + """用例数据语义错误：信号名未在数据字典中找到、步骤时间非数值等。""" pass class ExcelWriteError(MILSDKError): - """Excel 文件写入错误（权限问题、保存失败等）""" + """Excel 写入阶段错误：磁盘权限、文件被占用、保存失败等。""" pass \ No newline at end of file diff --git a/mil/core/mil_create_data_excel.py b/mil/core/mil_create_data_excel.py index 4883ff3..0ee6b46 100644 --- a/mil/core/mil_create_data_excel.py +++ b/mil/core/mil_create_data_excel.py @@ -24,18 +24,23 @@ def create_excel_case( data_dict: dict, sheets_dict: dict, ) -> None: - """生成测试用例 Excel 文件 + """根据读入的数据字典与用例模板，批量生成测试用例 Excel 文件。 + + 每个用例以"用例名.xlsx"的形式输出到 excel_path 目录下，文件格式 + 与 read_excel_data 读入的模板保持一致。 Args: - excel_path: 输出目录路径 - data_dict: 数据字典 - sheets_dict: 工作表字典 + excel_path: 输出目录路径。 + data_dict: read_excel_data 返回的字典，需要包含 wb / sheet / source_row + 以及每个信号的 column / datalog。 + sheets_dict: read_excel_case 返回的用例字典。 Raises: - CaseDataError: 数据异常 - ExcelWriteError: 文件写入错误 + CaseDataError: 传入的数据字典类型不符合 Workbook/Worksheet。 + ExcelWriteError: 保存失败（路径非法、权限不足等）。 """ logger.info(f"开始生成测试用例，输出目录: {excel_path}") + # 取出原始工作簿的引用；后续每个用例都基于原 wb 做 deepcopy，避免相互污染 wb = data_dict.get('wb') sheet = data_dict.get('sheet') source_row = data_dict.get('source_row') @@ -44,11 +49,13 @@ def create_excel_case( logger.error("数据类型错误，Workbook 或 Worksheet 类型不匹配") raise CaseDataError("数据异常，Workbook 或 Worksheet 类型错误") + # 拆出"信号 -> 数据"部分，剥离 Workbook 元数据，便于对每个用例独立处理 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) case_count = 0 + # 对每个用例：拷贝原工作簿 → 写入信号值 → 另存为独立文件 for name in datas_dict.keys(): case_count += 1 logger.debug(f"正在生成用例: {name}") @@ -58,7 +65,8 @@ def create_excel_case( generate_path = f"{excel_path}/{name}.xlsx" _write_excel_data(sheet, datas_dict[name], source_row) - wb.save(generate_path) + # 注意必须保存副本，不能回写源 wb（修复后的关键点） + new_wb.save(generate_path) logger.info(f"已生成用例: {name}，路径: {generate_path}") @@ -66,16 +74,17 @@ def create_excel_case( def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> None: - """写入 Excel 数据 + """将每个信号的 datalog 序列写回到对应单元格，并校正末尾时间戳。 Args: - sheet: Worksheet 对象 - data_dict: 数据字典 - source_row: Source: Input 所在行号 + sheet: 目标 Worksheet 对象。 + data_dict: 当前用例的"信号 -> {column, datalog}"字典。 + source_row: "Source: Input" 所在行号（数据起始行）。 """ for name in data_dict.keys(): column = data_dict[name]["column"] datalog = data_dict[name]["datalog"] + # 逐点写入：根据值类型设置 data_type，便于后续读取时类型还原 for index, data in enumerate(datalog): try: value = int(data.value) @@ -85,10 +94,12 @@ def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> Non sheet.cell(row=source_row + index, column=column).data_type = "str" finally: sheet.cell(row=source_row + index, column=column).value = value + # 第 2 列（首个信号列）额外把 time 写进 A 列，保持和原模板一致 if column == 2: sheet.cell(row=source_row + index, column=1).value = data.time sheet.cell(row=source_row + index, column=1).data_type = "float" + # 校正尾部时间戳：当最后一行的时间小于信号最后采样的 time 时，沿用信号末尾时间 column = 2 while sheet.cell(1, column).value != "time": column += 1 @@ -102,14 +113,18 @@ def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> Non def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict: - """分析用例字典，将用例中的信号名替换为信号值 + """遍历每个用例模板，把"信号名"展开成"信号值"序列，得到每个用例的最终数据字典。 + + 处理逻辑： + - enable=True 的用例跳过（不生成）； + - 缺失 enable 字段也跳过，避免模板脏数据导致运行期错误。 Args: - data_dict: 数据字典 - sheets_dict: 工作表字典 + data_dict: 去除 wb/sheet/source_row 后的数据字典。 + sheets_dict: 用例字典（read_excel_case 的返回）。 Returns: - 处理后的用例字典 + 用例名 -> 用例专属数据字典 的映射。 """ datas_dict: CaseResultDict = {} for sheet_name in sheets_dict.keys(): @@ -118,10 +133,13 @@ def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict: for case in sheet_dict.keys(): try: if sheet_dict[case]["enable"]: + # 已完成的用例跳过生成 continue except KeyError: + # 缺少 enable 字段：保守跳过 continue + # 拷贝数据并重置 datalog，然后按步骤逐条填充 datas_dict[case] = copy.deepcopy(data_dict) _init_data_log(datas_dict[case]) _analysis_step(sheet_dict[case]["step"], datas_dict[case]) @@ -129,34 +147,40 @@ def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict: def _analysis_step(step_dict: dict, data_dict: dict) -> None: - """分析步骤字典，将步骤中的信号名替换为信号值 + """按步骤字典依次处理每一步：有 action 走 action；无 action 走"信号保持"。 Args: - step_dict: 步骤字典 - data_dict: 数据字典 + step_dict: 步骤字典 {"step0": {...}, "step1": {...}, ...}。 + data_dict: 当前用例的数据字典（in-place 修改）。 """ 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"]) else: + # 无 action：对每个信号按"前一帧值保持"在时间轴上插一个数据点 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: - """分析数据字典，将数据字典中的信号名替换为信号值 + """对单步 action 展开：为每个被赋值的信号追加或修改 datalog 中的当前时间点。 + + 注意：data_log_len 用于本步结束后将"未参与动作的其他信号"补齐到同等长度， + 保持所有信号在同一时间轴上对齐。 Args: - action_dict: 操作字典 - data_dict: 数据字典 - time: 时间戳 + action_dict: 操作字典（信号名 -> 字符串值）。 + data_dict: 当前用例的数据字典（in-place 修改）。 + time: 本步骤的时间戳。 Raises: - CaseDataError: 信号不存在 + CaseDataError: action 中存在 data_dict 找不到的信号。 """ data_log_len: int | None = None for name in action_dict.keys(): + # 大小写不敏感匹配：避免 Excel 中信号名大小写差异导致的 KeyError matched_key: str | None = None for key in data_dict.keys(): if name.lower() == key.strip().lower(): @@ -167,11 +191,14 @@ def _analysis_data(action_dict: dict, data_dict: dict, time: float) -> None: else: datalog = data_dict[matched_key]["datalog"] if datalog[-1].time == time: + # 同一时间戳已存在 → 修改末点值；多步合并到同一时刻 datalog[-1].value = action_dict[name] data_log_len = len(datalog) else: + # 追加新的采样点 datalog.append(DataLog(time, action_dict[name])) data_log_len = len(datalog) + # 对未参与动作的信号补点（保持上一帧值），保证时间轴对齐 for name in data_dict.keys(): datalog = data_dict[name]["datalog"] if len(datalog) != data_log_len: @@ -179,10 +206,10 @@ def _analysis_data(action_dict: dict, data_dict: dict, time: float) -> None: def _init_data_log(data_dict: dict) -> None: - """初始化数据日志 + """把每个信号的 datalog 重置为只保留首点，作为各用例的初始状态。 Args: - data_dict: 数据字典 + data_dict: 数据字典（in-place 修改）。 """ for name in data_dict.keys(): data_dict[name]["datalog"] = data_dict[name]["datalog"][:1] \ No newline at end of file diff --git a/mil/core/mil_read_case_excel.py b/mil/core/mil_read_case_excel.py index ba76c85..45e4060 100644 --- a/mil/core/mil_read_case_excel.py +++ b/mil/core/mil_read_case_excel.py @@ -28,7 +28,13 @@ CaseDict = dict[str, Any] class CaseColumns: - """用例 Excel 列索引常量""" + """用例 Excel 各列含义的常量定义。 + TITLE: 用例标题（用于分组与启停判断） + STATUS: 用例状态（"完成测试" 等） + ACTION: 操作描述（如 "signal1=value1; signal2=value2"） + NAME: 当前步骤的名称 + TIME: 当前步骤的时间戳（秒，浮点） + """ TITLE = 1 STATUS = 2 ACTION = 3 @@ -36,6 +42,7 @@ class CaseColumns: TIME = 5 +# STATUS 列等于此值表示该用例已被勾选为"已完成测试" STATUS_COMPLETE = "完成测试" @@ -44,22 +51,35 @@ def read_excel_case( data_dict: dict, addTimeEn: bool, ) -> CaseDict: - """读取 Excel 用例模板 + """读取 Excel 用例模板并解析为内存中的结构化用例字典。 Args: - excel_path: Excel 模板文件路径 - data_dict: 信号数据字典 - addTimeEn: 是否累加时间 + excel_path: Excel 模板文件路径。 + data_dict: 数据字典（来自 read_excel_data），用于校验 action 中的信号名。 + addTimeEn: True 表示按"累加"方式记录每一步的时间戳， + False 表示直接使用步骤中填写的时间。 Returns: - 用例字典 + 用例字典： + { + sheet_title: { + case_name: { + "enable": bool, + "step": { + "step0": {"name": ..., "time": ..., "action": {...} 可选}, + ... + } + } + } + } Raises: - ExcelReadError: 文件读取失败 - ExcelFormatError: 格式错误 - CaseDataError: 用例数据错误 + ExcelReadError: 文件读取失败（不存在 / 权限问题）。 + ExcelFormatError: 缺少 Atech-Hefei 表 / 缺少模板版本号。 + CaseDataError: 标题缺失 / 时间非数值 / action 信号未在 data_dict 中找到。 """ logger.info(f"开始读取用例模板: {excel_path}") + # 用只读 + 取值模式打开，避免触发公式重算、降低内存占用 try: wb = load_workbook(excel_path, read_only=True, data_only=True) logger.debug(f"Excel 文件加载成功，Sheet 数量: {len(wb.sheetnames)}") @@ -69,6 +89,7 @@ def read_excel_case( logger.error(f"读取 Excel 文件失败: {e}") raise ExcelReadError(f"无法读取Excel文件 {excel_path}: {str(e)}") + # 版本表 Atech-Hefei 必须存在；解析后将其从工作簿移除，避免进入后续遍历 try: version = _get_template_version(wb["Atech-Hefei"]) if version is None: @@ -82,25 +103,31 @@ def read_excel_case( result_dict: CaseDict = {} sheet_count = 0 + # 按 sheet 维度组织用例：每个 sheet 内连续的行通过 "标题" 列切换归属 for sheet in wb.worksheets: sheet_count += 1 logger.debug(f"正在读取用例模板表: {sheet.title}") result_dict[sheet.title] = {} + # 当前行所属标题 / 上一次见到过的标题，用于检测标题变化、新用例的开始 new_head = None old_head = None step_id = 0 row = 2 + # 累加模式下的当前累计时间 old_time = 0.0 while True: + # 自上而下扫描 TITLE 列；遇到 None 才视为结束 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} 行第 {CaseColumns.TITLE} 列必须有标题名" ) + # 标题切换：意味着进入新用例，重置步骤序号与累加时间 if new_head != old_head: step_id = 0 old_time = 0.0 @@ -109,6 +136,7 @@ def read_excel_case( result_dict[sheet.title][new_head] = {} result_dict[sheet.title][new_head]["step"] = {} + # 是否启用（仅看该用例首行的 STATUS 即可） if sheet.cell(row=row, column=CaseColumns.STATUS).value == STATUS_COMPLETE: result_dict[sheet.title][new_head]["enable"] = True else: @@ -116,8 +144,10 @@ def read_excel_case( 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): @@ -129,9 +159,11 @@ def read_excel_case( result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["name"] = step_name if addTimeEn: + # 累加模式下：用 old_time 累加，写出"绝对时间" old_time += step_time result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["time"] = old_time + # 解析 action 字符串（信号名=信号值;...），缺省视为无动作 strings = sheet.cell(row, CaseColumns.ACTION).value if strings is not None: result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["action"] = \ @@ -144,16 +176,17 @@ def read_excel_case( def _get_template_version(sheet: Worksheet) -> str | None: - """获取 Excel 模板版本号 + """读取 'Atech-Hefei' 表中 TITLE 列自第 2 行起的全部版本号文本。 Args: - sheet: Excel 工作表 + sheet: Excel 工作表对象。 Returns: - 模板版本号，如果未找到返回 None + 模板版本号字符串；若整列均为空，返回 None。 """ row = 2 version = None + # 持续向下读取直到遇到空单元格；保留最后一个非空值作为版本号 while sheet.cell(row=row, column=CaseColumns.TITLE).value is not None: version = sheet.cell(row=row, column=CaseColumns.TITLE).value row += 1 @@ -168,23 +201,30 @@ def _analysis_action( data_dict: dict, excel_path: str ) -> dict[str, str]: - """解析操作字符串 + """将 "signal1=value1; signal2=value2" 形式的操作字符串解析为字典。 + + 解析规则： + - 兼容中文分号 "；" 与换行符； + - 容忍空格； + - 缺 "=" 的片段直接跳过； + - 信号名按"忽略大小写 + 去空格"在 data_dict 中匹配，匹配失败抛 CaseDataError。 Args: - sheet: Excel 工作表 - row: 行号 - column: 列号 - strings: 操作字符串 - data_dict: 数据字典 - excel_path: Excel 文件路径 + sheet: Excel 工作表（用于错误信息中显示 sheet 名）。 + row: 当前所在行号。 + column: 当前所在列号。 + strings: 原始操作字符串。 + data_dict: 来自 read_excel_data 的信号字典。 + excel_path: 仅用于错误信息中上下文。 Returns: - 操作字典 + 信号名 -> 信号值的字典。 Raises: - CaseDataError: 信号不存在或格式错误 + CaseDataError: 信号在 data_dict 中找不到。 """ action_dict: dict[str, str] = {} + # 中文分号替换为半角；同时剔除换行避免空片段 strings = strings.replace("；", ";").replace("\n", "") for item in strings.split(";"): @@ -192,13 +232,16 @@ def _analysis_action( if not item: continue + # 容忍信号与值之间的空格，统一去掉 item = item.replace(" ", "") if "=" not in item: + # 没有 "=" 视为非法片段，直接跳过（保持容错） continue signal_name, signal_value = item.split("=", 1) signal_name = signal_name.strip() + # 大小写不敏感地在 data_dict 中查找原始 key（保留原拼写用于回写） matched_key: str | None = None for key in data_dict.keys(): if signal_name.lower() == key.strip().lower(): diff --git a/mil/core/mil_read_data_excel.py b/mil/core/mil_read_data_excel.py index de949f9..cb3b9c8 100644 --- a/mil/core/mil_read_data_excel.py +++ b/mil/core/mil_read_data_excel.py @@ -33,7 +33,10 @@ class ExcelReaderConfig: time_column: 时间列索引，默认 1（A 列） header_row: 信号名称所在行号，默认 1 type_row: 信号类型所在行号，默认 3 + interp_row: 插值策略所在行号，默认 6 data_start_row_offset: 相对于 source_row 的数据起始行偏移量，默认 1 + output_header: 输出数据标记行文本，默认 "Source: Output" + block_path_row: BlockPath 属性所在行号，默认 4 """ sheet_name: str = "Scenario1" source_header: str = "Source: Input" @@ -75,9 +78,11 @@ def read_excel_data( ExcelReadError: 文件不存在或读取失败 ExcelFormatError: Excel 格式不符合预期 """ + # 未显式传入配置时，退回到默认约定值 config = config or ExcelReaderConfig() logger.info(f"开始读取文件: {excel_path}") + # 1) 加载底层 Excel 文件：openpyxl 自身抛出的两类异常分别映射到自定义异常类 try: wb = load_workbook(excel_path) logger.info(f"文件加载成功，Sheet 数量: {len(wb.sheetnames)}") @@ -88,59 +93,61 @@ def read_excel_data( logger.error(f"读取文件失败: {e}") raise ExcelReadError(f"读取Excel文件 {excel_path} 失败: {e}") + # 2) 取出约定名称的工作表，KeyError 表示工作表缺失，视为格式错误 try: sheet = wb[config.sheet_name] except KeyError: logger.error(f"缺少 '{config.sheet_name}' 表") raise ExcelFormatError(f"Excel文件 {excel_path} 缺少 '{config.sheet_name}' 表") + # 3) 自第 1 行起纵向扫描 "Source: Input" 标记行 column = 2 source_row = 1 while sheet.cell(row=source_row, column=column).value != config.source_header: source_row += 1 + # 扫描到表格末尾仍未命中，认定为格式错误 if source_row >= sheet.max_row: logger.error(f"缺少 {config.source_header} 标记") raise ExcelFormatError(f"Excel格式错误,缺少 {config.source_header}") logger.debug(f"找到 {config.source_header} 标记，行号: {source_row}") + # 4) 沿 source_row 横向遍历每一列，识别并解析出每个有效信号 signals: SignalDict = {} column = 2 - current_header = None while True: + # 当列所在单元格的标记不再是 source_header 时，说明信号区域结束 value = sheet.cell(source_row, column).value - if value is not None: - current_header = value - if current_header is None: - logger.error(f"缺少 {config.source_header} 标记") - raise ExcelFormatError(f"Excel格式错误,缺少 {config.source_header}") - if current_header == config.source_header: - name = sheet.cell(config.header_row, column).value - if name is None: - break - if name == "time": - column += 1 - continue - if name in ("Parameter:", "Value", "BlockPath"): - column += 1 - continue - attributes = __get_signal_attributes(sheet, column, source_row) - if attributes is None: - break - signals[name] = SignalData( - column=column, - attributes=attributes, - datalog=__get_data_log(sheet, source_row + config.data_start_row_offset, column, config.time_column) - ) - logger.debug(f"解析信号: {name}, 数据点: {len(signals[name].datalog)}") - else: + if value != config.source_header: break + # 取 header_row 行的列名作为信号名 + name = sheet.cell(config.header_row, column).value + if name is None: + break + # 时间列与若干固定属性名跳过 + if name == "time": + column += 1 + continue + if name in ("Parameter:", "Value", "BlockPath"): + column += 1 + continue + # 收集列上方属性 + 数据序列，写入 SignalData + attributes = __get_signal_attributes(sheet, column, source_row) + if attributes is None: + break + signals[name] = SignalData( + column=column, + attributes=attributes, + datalog=__get_data_log(sheet, source_row + config.data_start_row_offset, column, config.time_column) + ) + logger.debug(f"解析信号: {name}, 数据点: {len(signals[name].datalog)}") column += 1 signal_count = len(signals) logger.info(f"解析完成，共 {signal_count} 个信号") + # 新接口：返回强类型 ExcelDataResult（推荐） if return_object: return ExcelDataResult( sheet_name=config.sheet_name, @@ -148,6 +155,7 @@ def read_excel_data( signals=signals ) + # 旧接口：保持返回 dict 以便调用方继续拿到 Workbook / Worksheet 引用 data: DataDict = { "wb": wb, "sheet": sheet, @@ -164,15 +172,15 @@ def __get_signal_attributes( column: int, max_row: int ) -> dict[int, str]: - """从指定位置读取信号属性字典 + """读取指定列在 [1, max_row) 范围内各属性行的内容。 Args: sheet: Worksheet 对象 column: 列号 - max_row: 最大行号 + max_row: 信号名行（不含）以内的最大行号 Returns: - 信号属性字典，键为行号，值为单元格值 + 信号属性字典，键为行号（int），值为该单元格文本（str） """ attributes: dict[int, str] = {} for row in range(1, max_row): @@ -185,19 +193,20 @@ def __get_data_log( column: int, time_column: int = 1 ) -> list[DataLog]: - """从指定位置读取时间-值数据对列表 + """从指定起始行向下读取 (time, value) 序列，遇到空时间戳即终止。 Args: sheet: Worksheet 对象 row: 起始行号 column: 数据列号 - time_column: 时间列索引，默认 1 + time_column: 时间列索引，默认 1（A 列） Returns: DataLog 对象列表，直到遇到空时间戳为止 """ data_log: list[DataLog] = [] while True: + # 同时读取时间列与数据列，时间列空即视为数据终止 time = sheet.cell(row=row, column=time_column).value value = sheet.cell(row=row, column=column).value if time is None: diff --git a/mil/core/mil_update_excel.py b/mil/core/mil_update_excel.py index 7fee7d1..d231809 100644 --- a/mil/core/mil_update_excel.py +++ b/mil/core/mil_update_excel.py @@ -16,18 +16,25 @@ logger = logging.getLogger(__name__) def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None: - """更新 Excel 文件中的信号列 + """将旧 Excel 同步到新 Excel 的信号列集合（增 / 删）。 + + 流程： + 1. 拆出 wb / sheet / source_row 等元数据； + 2. 算 diff：要新增的信号、要删除的信号； + 3. 先删后插（保持列号稳定）； + 4. 回写文件。 Args: - filename: Excel 文件路径 - old_data: 旧数据字典 - new_data: 新数据字典 + filename: 目标 Excel 文件路径。 + old_data: 旧数据字典（含 wb / sheet / source_row 及各信号）。 + new_data: 新数据字典（结构同上，用于 diff 比较）。 Raises: - CaseDataError: 数据类型错误 + CaseDataError: 旧数据字典中 Workbook/Worksheet 类型不匹配。 """ logger.info(f"开始更新 Excel 文件 {filename}") + # 注意：pop 会从字典中移除键，调用前请确认数据不再被复用 old_wb = old_data.pop('wb') old_sheet = old_data.pop('sheet') old_source_row = old_data.pop('source_row') @@ -40,25 +47,32 @@ def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None: 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}") - + + # 再插入新增列：先 insert_cols 占位，再填充属性行与数据点 for name, info in add_input.items(): old_sheet.insert_cols(info['column']) - for index in range(1,old_source_row): - if index == old_source_row -1 : - old_sheet.cell(index, info['column']).value = info['attributes'][new_source_row - 1] + # 1) 属性行（header 行以下至 source_row - 1） + for index in range(1, old_source_row): + if index == old_source_row - 1: + # source_header 行：从 attributes 字典里取出 source_header 文本 + # 注：attributes 的行号空间与 old_data 保持一致，应使用 old_source_row + old_sheet.cell(index, info['column']).value = info['attributes'][old_source_row - 1] old_sheet.cell(index, info['column']).data_type = "str" else: old_sheet.cell(index, info['column']).value = info['attributes'][index] old_sheet.cell(index, info['column']).data_type = "str" + # 2) 数据行：根据值类型设置 data_type for index in range(old_source_row + 1, datalog_len + old_source_row + 1): try: value = int(info['datalog'][0].value) @@ -75,14 +89,14 @@ def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None: def _get_add_input(old_data: dict, new_data: dict) -> dict[str, SignalData]: - """获取新增输入数据 + """计算"新增"信号集合：存在于 new_data 但不存在于 old_data 的信号。 Args: - old_data: 旧数据字典 - new_data: 新数据字典 + old_data: 旧数据字典。 + new_data: 新数据字典。 Returns: - 新增信号字典 + 新增信号字典。 """ add_input: dict[str, SignalData] = {} @@ -93,14 +107,14 @@ def _get_add_input(old_data: dict, new_data: dict) -> dict[str, SignalData]: def _get_del_input(old_data: dict, new_data: dict) -> dict[str, SignalData]: - """获取删除输入数据 + """计算"删除"信号集合：存在于 old_data 但不存在于 new_data 的信号。 Args: - old_data: 旧数据字典 - new_data: 新数据字典 + old_data: 旧数据字典。 + new_data: 新数据字典。 Returns: - 删除信号字典 + 删除信号字典。 """ del_input: dict[str, SignalData] = {} @@ -111,14 +125,18 @@ def _get_del_input(old_data: dict, new_data: dict) -> dict[str, SignalData]: def _get_datalog_len(old_data: dict) -> int: - """获取数据日志长度 + """获取旧数据中第一个信号的 datalog 长度，作为新增列填充行数的基准。 + + 假设 old_data 中各信号 datalog 长度一致（来自同一时间轴）。 Args: - old_data: 数据字典 + old_data: 旧数据字典。 Returns: - 数据日志长度 + 数据日志长度（行数）；空字典返回 0。 """ - for name in old_data.keys(): - return len(old_data[name]['datalog']) - return 0 \ No newline at end of file + if not old_data: + return 0 + # 取首项的 datalog 长度即可，无需遍历整张字典 + first_name = next(iter(old_data)) + return len(old_data[first_name]['datalog']) \ No newline at end of file diff --git a/mil/ui/mil_project.py b/mil/ui/mil_project.py new file mode 100644 index 0000000..e69de29 diff --git a/mil/ui/mil_project.ui b/mil/ui/mil_project.ui new file mode 100644 index 0000000..869731f --- /dev/null +++ b/mil/ui/mil_project.ui @@ -0,0 +1,113 @@ + + + FrameProject + + + + 0 + 0 + 740 + 95 + + + + Frame + + + + + + + + 模型项目： + + + + + + + 数据文件： + + + + + + + 测试用例： + + + + + + + + + + + + 0 + 0 + + + + + + + + + + + + 0 + 0 + + + + + + + + + 0 + 0 + + + + + + + + + + + + + 0 + 0 + + + + 选择数据文件 + + + + + + + + 0 + 0 + + + + 选择测试用例 + + + + + + + + + + diff --git a/mil/ui/mil_project_ui.py b/mil/ui/mil_project_ui.py new file mode 100644 index 0000000..b0288bf --- /dev/null +++ b/mil/ui/mil_project_ui.py @@ -0,0 +1,114 @@ +# -*- coding: utf-8 -*- + +################################################################################ +## Form generated from reading UI file 'mil_project.ui' +## +## Created by: Qt User Interface Compiler version 6.11.1 +## +## WARNING! All changes made in this file will be lost when recompiling UI file! +################################################################################ + +from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale, + QMetaObject, QObject, QPoint, QRect, + QSize, QTime, QUrl, Qt) +from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor, + QFont, QFontDatabase, QGradient, QIcon, + QImage, QKeySequence, QLinearGradient, QPainter, + QPalette, QPixmap, QRadialGradient, QTransform) +from PySide6.QtWidgets import (QApplication, QFrame, QGridLayout, QLabel, + QLineEdit, QPushButton, QSizePolicy, QWidget) + +class Ui_FrameProject(object): + def setupUi(self, FrameProject): + if not FrameProject.objectName(): + FrameProject.setObjectName(u"FrameProject") + FrameProject.resize(740, 95) + self.gridLayout_4 = QGridLayout(FrameProject) + self.gridLayout_4.setObjectName(u"gridLayout_4") + self.gridLayout = QGridLayout() + self.gridLayout.setObjectName(u"gridLayout") + self.label = QLabel(FrameProject) + self.label.setObjectName(u"label") + + self.gridLayout.addWidget(self.label, 0, 0, 1, 1) + + self.label_2 = QLabel(FrameProject) + self.label_2.setObjectName(u"label_2") + + self.gridLayout.addWidget(self.label_2, 1, 0, 1, 1) + + self.label_3 = QLabel(FrameProject) + self.label_3.setObjectName(u"label_3") + + self.gridLayout.addWidget(self.label_3, 2, 0, 1, 1) + + + self.gridLayout_4.addLayout(self.gridLayout, 0, 0, 1, 1) + + self.gridLayout_2 = QGridLayout() + self.gridLayout_2.setObjectName(u"gridLayout_2") + self.lineEditName = QLineEdit(FrameProject) + self.lineEditName.setObjectName(u"lineEditName") + sizePolicy = QSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred) + sizePolicy.setHorizontalStretch(0) + sizePolicy.setVerticalStretch(0) + sizePolicy.setHeightForWidth(self.lineEditName.sizePolicy().hasHeightForWidth()) + self.lineEditName.setSizePolicy(sizePolicy) + + self.gridLayout_2.addWidget(self.lineEditName, 0, 0, 1, 1) + + self.lineEditDataPath = QLineEdit(FrameProject) + self.lineEditDataPath.setObjectName(u"lineEditDataPath") + sizePolicy.setHeightForWidth(self.lineEditDataPath.sizePolicy().hasHeightForWidth()) + self.lineEditDataPath.setSizePolicy(sizePolicy) + + self.gridLayout_2.addWidget(self.lineEditDataPath, 1, 0, 1, 1) + + self.lineEditFilePath = QLineEdit(FrameProject) + self.lineEditFilePath.setObjectName(u"lineEditFilePath") + sizePolicy.setHeightForWidth(self.lineEditFilePath.sizePolicy().hasHeightForWidth()) + self.lineEditFilePath.setSizePolicy(sizePolicy) + + self.gridLayout_2.addWidget(self.lineEditFilePath, 2, 0, 1, 1) + + + self.gridLayout_4.addLayout(self.gridLayout_2, 0, 1, 1, 1) + + self.gridLayout_3 = QGridLayout() + self.gridLayout_3.setObjectName(u"gridLayout_3") + self.pushButtonDataPath = QPushButton(FrameProject) + self.pushButtonDataPath.setObjectName(u"pushButtonDataPath") + sizePolicy1 = QSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Expanding) + sizePolicy1.setHorizontalStretch(0) + sizePolicy1.setVerticalStretch(0) + sizePolicy1.setHeightForWidth(self.pushButtonDataPath.sizePolicy().hasHeightForWidth()) + self.pushButtonDataPath.setSizePolicy(sizePolicy1) + + self.gridLayout_3.addWidget(self.pushButtonDataPath, 0, 0, 1, 1) + + self.pushButtonFilePath = QPushButton(FrameProject) + self.pushButtonFilePath.setObjectName(u"pushButtonFilePath") + sizePolicy1.setHeightForWidth(self.pushButtonFilePath.sizePolicy().hasHeightForWidth()) + self.pushButtonFilePath.setSizePolicy(sizePolicy1) + + self.gridLayout_3.addWidget(self.pushButtonFilePath, 1, 0, 1, 1) + + + self.gridLayout_4.addLayout(self.gridLayout_3, 0, 2, 1, 1) + + + self.retranslateUi(FrameProject) + + QMetaObject.connectSlotsByName(FrameProject) + # setupUi + + def retranslateUi(self, FrameProject): + FrameProject.setWindowTitle(QCoreApplication.translate("FrameProject", u"Frame", None)) + self.label.setText(QCoreApplication.translate("FrameProject", u"\u6a21\u578b\u9879\u76ee\uff1a", None)) + self.label_2.setText(QCoreApplication.translate("FrameProject", u"\u6570\u636e\u6587\u4ef6\uff1a", None)) + self.label_3.setText(QCoreApplication.translate("FrameProject", u"\u6d4b\u8bd5\u7528\u4f8b\uff1a", None)) + self.lineEditName.setText("") + self.pushButtonDataPath.setText(QCoreApplication.translate("FrameProject", u"\u9009\u62e9\u6570\u636e\u6587\u4ef6", None)) + self.pushButtonFilePath.setText(QCoreApplication.translate("FrameProject", u"\u9009\u62e9\u6d4b\u8bd5\u7528\u4f8b", None)) + # retranslateUi + diff --git a/mil/ui/mil_tool.py b/mil/ui/mil_tool.py index 8f889c3..08e4d88 100644 --- a/mil/ui/mil_tool.py +++ b/mil/ui/mil_tool.py @@ -1,17 +1,26 @@ +import os import logging from .mil_tool_ui import Ui_FrameMILTool from PySide6.QtWidgets import QFrame +from mil.core import Config + + logger = logging.getLogger(__name__) +CONF = "mil.json" class FrameMILTool(QFrame,Ui_FrameMILTool): def __init__(self,parent=None): super().__init__(parent) self.setupUi(self) - def init_ui(self): - pass - # logger.info("FrameMILTool 初始化完成") + self.initUI() - + def initUI(self): + self.config = Config() + if os.path.exists(CONF): + self.config.load_config(CONF) + else: + self.config.save_config(CONF) + pass diff --git a/requirements.txt b/requirements.txt index 3b6476f..c0099c0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,4 +4,4 @@ openpyxl>=3.0.0 # 开发依赖 -pytest>=7.0.0 \ No newline at end of file +pytest>=7.0.0