上传初版
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
"""MIL SDK 核心模块
|
||||
|
||||
提供 MIL 仿真数据读取功能
|
||||
|
||||
使用示例:
|
||||
from src.core import read_excel_data
|
||||
|
||||
result = read_excel_data("simulation.xlsx")
|
||||
"""
|
||||
import logging
|
||||
|
||||
from .base import DataLog, SignalData, ExcelDataResult
|
||||
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
|
||||
from .mil_update_excel import update_case_excel
|
||||
|
||||
from .exceptions import (
|
||||
MILSDKError,
|
||||
ExcelReadError,
|
||||
ExcelFormatError,
|
||||
CaseDataError,
|
||||
ExcelWriteError,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"DataLog",
|
||||
"SignalData",
|
||||
"ExcelDataResult",
|
||||
"ExcelReaderConfig",
|
||||
"read_excel_data",
|
||||
"read_excel_case",
|
||||
"create_excel_case",
|
||||
"update_case_excel",
|
||||
"MILSDKError",
|
||||
"ExcelReadError",
|
||||
"ExcelFormatError",
|
||||
"CaseDataError",
|
||||
"ExcelWriteError",
|
||||
"setup_logging",
|
||||
"get_logger",
|
||||
]
|
||||
@@ -0,0 +1,92 @@
|
||||
"""MIL SDK 核心数据模型"""
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class DataLog:
|
||||
"""仿真数据日志记录
|
||||
|
||||
Attributes:
|
||||
time: 时间戳（秒）
|
||||
value: 信号值（可以是任意类型）
|
||||
"""
|
||||
time: float = 0.0
|
||||
value: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class SignalData:
|
||||
"""信号数据封装
|
||||
|
||||
Attributes:
|
||||
signal_type: 信号类型
|
||||
column: 列索引
|
||||
datalog: 数据日志列表
|
||||
"""
|
||||
# signal_type: str | None = None
|
||||
column: int = 0
|
||||
attributes: dict[str, str] = field(default_factory=dict)
|
||||
datalog: list[DataLog] = field(default_factory=list)
|
||||
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""转换为字典格式
|
||||
|
||||
Returns:
|
||||
包含 signal_type、column、datalog 的字典
|
||||
"""
|
||||
return {
|
||||
"column": self.column,
|
||||
"attributes": self.attributes,
|
||||
"datalog": self.datalog
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExcelDataResult:
|
||||
"""Excel 数据读取结果封装
|
||||
|
||||
提供对 Excel 数据的类型安全访问，隐藏内部实现细节
|
||||
|
||||
Attributes:
|
||||
sheet_name: 工作表名称
|
||||
source_row: Source: Input 所在行号
|
||||
signals: 信号名称到信号数据的映射
|
||||
"""
|
||||
sheet_name: str = "Scenario1"
|
||||
source_row: int = 0
|
||||
signals: dict[str, SignalData] = field(default_factory=dict)
|
||||
|
||||
def get_signal(self, name: str) -> SignalData | None:
|
||||
"""获取指定信号的数据
|
||||
|
||||
Args:
|
||||
name: 信号名称
|
||||
|
||||
Returns:
|
||||
信号数据对象，如果不存在返回 None
|
||||
"""
|
||||
return self.signals.get(name)
|
||||
|
||||
def get_signal_names(self) -> list[str]:
|
||||
"""获取所有信号名称
|
||||
|
||||
Returns:
|
||||
信号名称列表
|
||||
"""
|
||||
return list(self.signals.keys())
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""转换为字典格式（兼容旧 API）
|
||||
|
||||
Returns:
|
||||
包含所有信号的字典，保留原有的数据结构
|
||||
"""
|
||||
result = {
|
||||
"sheet_name": self.sheet_name,
|
||||
"source_row": self.source_row
|
||||
}
|
||||
for name, signal in self.signals.items():
|
||||
result[name] = signal.to_dict()
|
||||
return result
|
||||
@@ -0,0 +1,26 @@
|
||||
"""MIL SDK 自定义异常模块"""
|
||||
|
||||
|
||||
class MILSDKError(Exception):
|
||||
"""MIL SDK 基础异常类"""
|
||||
pass
|
||||
|
||||
|
||||
class ExcelReadError(MILSDKError):
|
||||
"""Excel 文件读取错误（文件不存在、权限问题等）"""
|
||||
pass
|
||||
|
||||
|
||||
class ExcelFormatError(MILSDKError):
|
||||
"""Excel 格式错误（缺少 Sheet、格式不匹配等）"""
|
||||
pass
|
||||
|
||||
|
||||
class CaseDataError(MILSDKError):
|
||||
"""用例数据错误（信号不存在、类型错误等）"""
|
||||
pass
|
||||
|
||||
|
||||
class ExcelWriteError(MILSDKError):
|
||||
"""Excel 文件写入错误（权限问题、保存失败等）"""
|
||||
pass
|
||||
@@ -0,0 +1,188 @@
|
||||
"""MIL 测试用例 Excel 生成模块
|
||||
|
||||
主要功能:
|
||||
- 根据数据和用例模板生成测试用例 Excel 文件
|
||||
- 将用例中的信号名替换为实际信号值
|
||||
|
||||
生成的 Excel 文件遵循与读取相同的格式约定。
|
||||
"""
|
||||
import copy
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from .base import DataLog
|
||||
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,
|
||||
sheets_dict: dict,
|
||||
) -> None:
|
||||
"""生成测试用例 Excel 文件
|
||||
|
||||
Args:
|
||||
excel_path: 输出目录路径
|
||||
data_dict: 数据字典
|
||||
sheets_dict: 工作表字典
|
||||
|
||||
Raises:
|
||||
CaseDataError: 数据异常
|
||||
ExcelWriteError: 文件写入错误
|
||||
"""
|
||||
logger.info(f"开始生成测试用例，输出目录: {excel_path}")
|
||||
wb = data_dict.get('wb')
|
||||
sheet = data_dict.get('sheet')
|
||||
source_row = data_dict.get('source_row')
|
||||
|
||||
if not isinstance(wb, Workbook) or not isinstance(sheet, Worksheet):
|
||||
logger.error("数据类型错误，Workbook 或 Worksheet 类型不匹配")
|
||||
raise CaseDataError("数据异常，Workbook 或 Worksheet 类型错误")
|
||||
|
||||
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}")
|
||||
new_wb = copy.deepcopy(wb)
|
||||
sheet = new_wb["Scenario1"]
|
||||
|
||||
generate_path = f"{excel_path}/{name}.xlsx"
|
||||
_write_excel_data(sheet, datas_dict[name], source_row)
|
||||
|
||||
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:
|
||||
"""写入 Excel 数据
|
||||
|
||||
Args:
|
||||
sheet: Worksheet 对象
|
||||
data_dict: 数据字典
|
||||
source_row: Source: Input 所在行号
|
||||
"""
|
||||
for name in data_dict.keys():
|
||||
column = data_dict[name]["column"]
|
||||
datalog = data_dict[name]["datalog"]
|
||||
for index, data in enumerate(datalog):
|
||||
try:
|
||||
value = int(data.value)
|
||||
sheet.cell(row=source_row + index, column=column).data_type = "int"
|
||||
except (ValueError, TypeError):
|
||||
value = data.value
|
||||
sheet.cell(row=source_row + index, column=column).data_type = "str"
|
||||
finally:
|
||||
sheet.cell(row=source_row + index, column=column).value = value
|
||||
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"
|
||||
|
||||
column = 2
|
||||
while sheet.cell(1, column).value != "time":
|
||||
column += 1
|
||||
row = 1
|
||||
while sheet.cell(source_row + row, column).value is not None:
|
||||
row += 1
|
||||
stop_time = sheet.cell(source_row + row - 1, column).value
|
||||
if stop_time < datalog[-1].time:
|
||||
sheet.cell(source_row + row - 1, column).value = datalog[-1].time
|
||||
sheet.cell(source_row + row - 1, column).data_type = "float"
|
||||
|
||||
|
||||
def _analysis_case(data_dict: dict, sheets_dict: dict) -> CaseResultDict:
|
||||
"""分析用例字典，将用例中的信号名替换为信号值
|
||||
|
||||
Args:
|
||||
data_dict: 数据字典
|
||||
sheets_dict: 工作表字典
|
||||
|
||||
Returns:
|
||||
处理后的用例字典
|
||||
"""
|
||||
datas_dict: CaseResultDict = {}
|
||||
for sheet_name in sheets_dict.keys():
|
||||
sheet_dict = sheets_dict[sheet_name]
|
||||
|
||||
for case in sheet_dict.keys():
|
||||
try:
|
||||
if sheet_dict[case]["enable"]:
|
||||
continue
|
||||
except KeyError:
|
||||
continue
|
||||
|
||||
datas_dict[case] = copy.deepcopy(data_dict)
|
||||
_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:
|
||||
"""分析步骤字典，将步骤中的信号名替换为信号值
|
||||
|
||||
Args:
|
||||
step_dict: 步骤字典
|
||||
data_dict: 数据字典
|
||||
"""
|
||||
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:
|
||||
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:
|
||||
"""分析数据字典，将数据字典中的信号名替换为信号值
|
||||
|
||||
Args:
|
||||
action_dict: 操作字典
|
||||
data_dict: 数据字典
|
||||
time: 时间戳
|
||||
|
||||
Raises:
|
||||
CaseDataError: 信号不存在
|
||||
"""
|
||||
data_log_len: int | None = None
|
||||
for name in action_dict.keys():
|
||||
matched_key: str | None = None
|
||||
for key in data_dict.keys():
|
||||
if name.lower() == key.strip().lower():
|
||||
matched_key = key
|
||||
break
|
||||
if matched_key is None:
|
||||
raise CaseDataError(f"数据字典中不存在信号 {name}")
|
||||
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:
|
||||
datalog.append(DataLog(time, datalog[-1].value))
|
||||
|
||||
|
||||
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]
|
||||
@@ -0,0 +1,215 @@
|
||||
"""MIL 测试用例 Excel 读取模块
|
||||
|
||||
主要功能:
|
||||
- 读取 Excel 用例模板
|
||||
- 解析测试步骤和操作
|
||||
|
||||
Excel 模板格式约定:
|
||||
- Sheet 'Atech-Hefei': 存储模板版本号
|
||||
- 其他 Sheet: 存储测试用例数据
|
||||
- 列 1: 用例标题
|
||||
- 列 2: 状态（完成测试/未完成）
|
||||
- 列 3: 操作描述（signal1=value1; signal2=value2）
|
||||
- 列 4: 步骤名称
|
||||
- 列 5: 时间
|
||||
"""
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
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__)
|
||||
|
||||
CaseDict = dict[str, Any]
|
||||
|
||||
|
||||
class CaseColumns:
|
||||
"""用例 Excel 列索引常量"""
|
||||
TITLE = 1
|
||||
STATUS = 2
|
||||
ACTION = 3
|
||||
NAME = 4
|
||||
TIME = 5
|
||||
|
||||
|
||||
STATUS_COMPLETE = "完成测试"
|
||||
|
||||
|
||||
def read_excel_case(
|
||||
excel_path: str,
|
||||
data_dict: dict,
|
||||
addTimeEn: bool,
|
||||
) -> CaseDict:
|
||||
"""读取 Excel 用例模板
|
||||
|
||||
Args:
|
||||
excel_path: Excel 模板文件路径
|
||||
data_dict: 信号数据字典
|
||||
addTimeEn: 是否累加时间
|
||||
|
||||
Returns:
|
||||
用例字典
|
||||
|
||||
Raises:
|
||||
ExcelReadError: 文件读取失败
|
||||
ExcelFormatError: 格式错误
|
||||
CaseDataError: 用例数据错误
|
||||
"""
|
||||
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)}")
|
||||
except FileNotFoundError:
|
||||
raise ExcelReadError(f"Excel文件 {excel_path} 不存在")
|
||||
except Exception as e:
|
||||
logger.error(f"读取 Excel 文件失败: {e}")
|
||||
raise ExcelReadError(f"无法读取Excel文件 {excel_path}: {str(e)}")
|
||||
|
||||
try:
|
||||
version = _get_template_version(wb["Atech-Hefei"])
|
||||
if version is None:
|
||||
logger.error("缺少模板版本号")
|
||||
raise ExcelFormatError("Excel格式错误，缺少模板版本号")
|
||||
logger.info(f"用例模板版本号: {version}")
|
||||
del wb["Atech-Hefei"]
|
||||
except KeyError:
|
||||
logger.error(f"缺少 'Atech-Hefei' 表")
|
||||
raise ExcelFormatError(f"Excel文件 {excel_path} 格式错误，缺少 Atech-Hefei 表")
|
||||
|
||||
result_dict: CaseDict = {}
|
||||
sheet_count = 0
|
||||
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:
|
||||
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
|
||||
old_head = new_head
|
||||
|
||||
result_dict[sheet.title][new_head] = {}
|
||||
result_dict[sheet.title][new_head]["step"] = {}
|
||||
|
||||
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, 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} 行第 {CaseColumns.TIME} 列时间必须是数字"
|
||||
)
|
||||
|
||||
result_dict[sheet.title][new_head]["step"][f"step{step_id}"] = {}
|
||||
result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["name"] = step_name
|
||||
|
||||
if addTimeEn:
|
||||
old_time += step_time
|
||||
result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["time"] = old_time
|
||||
|
||||
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, CaseColumns.ACTION, strings, data_dict, excel_path)
|
||||
step_id += 1
|
||||
row += 1
|
||||
|
||||
logger.info(f"用例模板读取完成，共 {sheet_count} 个表，{sum(len(v) for v in result_dict.values())} 个用例")
|
||||
return result_dict
|
||||
|
||||
|
||||
def _get_template_version(sheet: Worksheet) -> str | None:
|
||||
"""获取 Excel 模板版本号
|
||||
|
||||
Args:
|
||||
sheet: Excel 工作表
|
||||
|
||||
Returns:
|
||||
模板版本号，如果未找到返回 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
|
||||
return version
|
||||
|
||||
|
||||
def _analysis_action(
|
||||
sheet: Worksheet,
|
||||
row: int,
|
||||
column: int,
|
||||
strings: str,
|
||||
data_dict: dict,
|
||||
excel_path: str
|
||||
) -> dict[str, str]:
|
||||
"""解析操作字符串
|
||||
|
||||
Args:
|
||||
sheet: Excel 工作表
|
||||
row: 行号
|
||||
column: 列号
|
||||
strings: 操作字符串
|
||||
data_dict: 数据字典
|
||||
excel_path: Excel 文件路径
|
||||
|
||||
Returns:
|
||||
操作字典
|
||||
|
||||
Raises:
|
||||
CaseDataError: 信号不存在或格式错误
|
||||
"""
|
||||
action_dict: dict[str, str] = {}
|
||||
strings = strings.replace("；", ";").replace("\n", "")
|
||||
|
||||
for item in strings.split(";"):
|
||||
item = item.strip()
|
||||
if not item:
|
||||
continue
|
||||
|
||||
item = item.replace(" ", "")
|
||||
if "=" not in item:
|
||||
continue
|
||||
|
||||
signal_name, signal_value = item.split("=", 1)
|
||||
signal_name = signal_name.strip()
|
||||
|
||||
matched_key: str | None = None
|
||||
for key in data_dict.keys():
|
||||
if signal_name.lower() == key.strip().lower():
|
||||
matched_key = key
|
||||
break
|
||||
|
||||
if matched_key is None:
|
||||
raise CaseDataError(
|
||||
f"{sheet.title} 第 {row} 行第 {column} 列操作中信号 {signal_name} 不存在"
|
||||
)
|
||||
|
||||
action_dict[matched_key] = signal_value
|
||||
|
||||
return action_dict
|
||||
@@ -0,0 +1,207 @@
|
||||
"""MIL 仿真数据 Excel 解析模块
|
||||
|
||||
Excel 文件格式约定:
|
||||
- Sheet 名称: Scenario1
|
||||
- 行 1: 信号名称（time, signal1, signal2...）
|
||||
- 行 2: 预留行（通常为空）
|
||||
- 行 3: "Source: Input" 标记行，标识数据列开始
|
||||
- 行 4+: 时间-值数据对
|
||||
"""
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.worksheet.worksheet import Worksheet
|
||||
|
||||
from .base import DataLog, SignalData, ExcelDataResult
|
||||
from .exceptions import ExcelReadError, ExcelFormatError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SignalDict = dict[str, SignalData]
|
||||
DataDict = dict[str, Any]
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExcelReaderConfig:
|
||||
"""Excel 读取配置
|
||||
|
||||
Attributes:
|
||||
sheet_name: 工作表名称，默认 "Scenario1"
|
||||
source_header: 数据源标记行文本，默认 "Source: Input"
|
||||
time_column: 时间列索引，默认 1（A 列）
|
||||
header_row: 信号名称所在行号，默认 1
|
||||
type_row: 信号类型所在行号，默认 3
|
||||
data_start_row_offset: 相对于 source_row 的数据起始行偏移量，默认 1
|
||||
"""
|
||||
sheet_name: str = "Scenario1"
|
||||
source_header: str = "Source: Input"
|
||||
time_column: int = 1
|
||||
header_row: int = 1
|
||||
type_row: int = 3
|
||||
interp_row: int = 6
|
||||
data_start_row_offset: int = 1
|
||||
|
||||
output_header: str = "Source: Output"
|
||||
block_path_row: int = 4
|
||||
|
||||
|
||||
def read_excel_data(
|
||||
excel_path: str,
|
||||
return_object: bool = False,
|
||||
config: ExcelReaderConfig | None = None
|
||||
) -> DataDict | ExcelDataResult:
|
||||
"""读取 MIL 仿真 Excel 文件并解析信号数据
|
||||
|
||||
Args:
|
||||
excel_path: Excel 文件路径
|
||||
return_object: 是否返回 ExcelDataResult 对象（推荐），默认 False 保持向后兼容
|
||||
config: Excel 读取配置，默认使用 ExcelReaderConfig()
|
||||
|
||||
Returns:
|
||||
return_object=False 时：包含以下键的字典（向后兼容）:
|
||||
- wb: Workbook 对象
|
||||
- sheet: Worksheet 对象
|
||||
- source_row: Source: Input 所在行号
|
||||
- <信号名>: 信号元数据（type, column, datalog）
|
||||
|
||||
return_object=True 时：ExcelDataResult 对象（推荐）
|
||||
- sheet_name: 工作表名称
|
||||
- source_row: Source: Input 所在行号
|
||||
- signals: 信号名称到信号数据的映射
|
||||
|
||||
Raises:
|
||||
ExcelReadError: 文件不存在或读取失败
|
||||
ExcelFormatError: Excel 格式不符合预期
|
||||
"""
|
||||
config = config or ExcelReaderConfig()
|
||||
|
||||
logger.info(f"开始读取文件: {excel_path}")
|
||||
try:
|
||||
wb = load_workbook(excel_path)
|
||||
logger.info(f"文件加载成功，Sheet 数量: {len(wb.sheetnames)}")
|
||||
except FileNotFoundError:
|
||||
logger.error(f"文件不存在: {excel_path}")
|
||||
raise ExcelReadError(f"Excel文件 {excel_path} 不存在")
|
||||
except Exception as e:
|
||||
logger.error(f"读取文件失败: {e}")
|
||||
raise ExcelReadError(f"读取Excel文件 {excel_path} 失败: {e}")
|
||||
|
||||
try:
|
||||
sheet = wb[config.sheet_name]
|
||||
except KeyError:
|
||||
logger.error(f"缺少 '{config.sheet_name}' 表")
|
||||
raise ExcelFormatError(f"Excel文件 {excel_path} 缺少 '{config.sheet_name}' 表")
|
||||
|
||||
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}")
|
||||
|
||||
signals: SignalDict = {}
|
||||
column = 2
|
||||
current_header = None
|
||||
while True:
|
||||
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:
|
||||
break
|
||||
column += 1
|
||||
|
||||
signal_count = len(signals)
|
||||
logger.info(f"解析完成，共 {signal_count} 个信号")
|
||||
|
||||
if return_object:
|
||||
return ExcelDataResult(
|
||||
sheet_name=config.sheet_name,
|
||||
source_row=source_row,
|
||||
signals=signals
|
||||
)
|
||||
|
||||
data: DataDict = {
|
||||
"wb": wb,
|
||||
"sheet": sheet,
|
||||
"source_row": source_row
|
||||
}
|
||||
for name, signal in signals.items():
|
||||
data[name] = signal.to_dict()
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def __get_signal_attributes(
|
||||
sheet: Worksheet,
|
||||
column: int,
|
||||
max_row: int
|
||||
) -> dict[int, str]:
|
||||
"""从指定位置读取信号属性字典
|
||||
|
||||
Args:
|
||||
sheet: Worksheet 对象
|
||||
column: 列号
|
||||
max_row: 最大行号
|
||||
|
||||
Returns:
|
||||
信号属性字典，键为行号，值为单元格值
|
||||
"""
|
||||
attributes: dict[int, str] = {}
|
||||
for row in range(1, max_row):
|
||||
attributes[row] = sheet.cell(row=row, column=column).value
|
||||
return attributes
|
||||
|
||||
def __get_data_log(
|
||||
sheet: Worksheet,
|
||||
row: int,
|
||||
column: int,
|
||||
time_column: int = 1
|
||||
) -> list[DataLog]:
|
||||
"""从指定位置读取时间-值数据对列表
|
||||
|
||||
Args:
|
||||
sheet: Worksheet 对象
|
||||
row: 起始行号
|
||||
column: 数据列号
|
||||
time_column: 时间列索引，默认 1
|
||||
|
||||
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:
|
||||
break
|
||||
data_log.append(DataLog(time, value))
|
||||
row += 1
|
||||
return data_log
|
||||
@@ -0,0 +1,124 @@
|
||||
"""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')
|
||||
old_source_row = old_data.pop('source_row')
|
||||
|
||||
new_data.pop('wb')
|
||||
new_data.pop('sheet')
|
||||
new_source_row = 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'])
|
||||
|
||||
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]
|
||||
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"
|
||||
|
||||
for index in range(old_source_row + 1, datalog_len + old_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
|
||||
Reference in New Issue
Block a user