更新优化

This commit is contained in:
2026-07-10 13:35:29 +08:00
parent 4c5fc718c1
commit 45e4f1ecf2
14 changed files with 573 additions and 122 deletions
+5 -4
View File
@@ -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"
]
+109 -6
View File
@@ -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,
+13 -6
View File
@@ -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
+52 -25
View File
@@ -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]
+64 -21
View File
@@ -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():
+40 -31
View File
@@ -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: 时间列索引,默认 1A 列)
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:
+41 -23
View File
@@ -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
if not old_data:
return 0
# 取首项的 datalog 长度即可,无需遍历整张字典
first_name = next(iter(old_data))
return len(old_data[first_name]['datalog'])