完成初版
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
"""MIL SDK 核心模块
|
||||
|
||||
提供 MIL 仿真数据读取功能
|
||||
|
||||
使用示例:
|
||||
from src.core import setup_logging, read_excel_data
|
||||
|
||||
setup_logging()
|
||||
result = read_excel_data("simulation.xlsx")
|
||||
"""
|
||||
import logging
|
||||
|
||||
from .base import DataLog, SignalData, ExcelDataResult
|
||||
from .mil_data_excel import read_excel_data, get_data_log, ExcelReaderConfig
|
||||
from .mil_case_excel import create_excel_case, read_excel_case
|
||||
from .exceptions import (
|
||||
MILSDKError,
|
||||
ExcelReadError,
|
||||
ExcelFormatError,
|
||||
CaseDataError,
|
||||
ExcelWriteError,
|
||||
)
|
||||
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",
|
||||
"ExcelDataResult",
|
||||
"ExcelReaderConfig",
|
||||
"read_excel_data",
|
||||
"get_data_log",
|
||||
"read_excel_case",
|
||||
"create_excel_case",
|
||||
"MILSDKError",
|
||||
"ExcelReadError",
|
||||
"ExcelFormatError",
|
||||
"CaseDataError",
|
||||
"ExcelWriteError",
|
||||
"setup_logging",
|
||||
"get_logger",
|
||||
]
|
||||
@@ -0,0 +1,90 @@
|
||||
"""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
|
||||
datalog: list[DataLog] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""转换为字典格式
|
||||
|
||||
Returns:
|
||||
包含 signal_type、column、datalog 的字典
|
||||
"""
|
||||
return {
|
||||
"type": self.signal_type,
|
||||
"column": self.column,
|
||||
"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,61 @@
|
||||
"""日志配置模块"""
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
|
||||
def setup_logging(
|
||||
level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO",
|
||||
log_file: str | Path | None = "logs/mil_sdk.log",
|
||||
console_output: bool = True,
|
||||
) -> logging.Logger:
|
||||
"""配置日志系统
|
||||
|
||||
Args:
|
||||
level: 日志级别，默认 INFO
|
||||
log_file: 日志文件路径，默认 logs/mil_sdk.log。设为 None 则不写入文件
|
||||
console_output: 是否输出到控制台，默认 True
|
||||
|
||||
Returns:
|
||||
根日志记录器
|
||||
"""
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel(level)
|
||||
|
||||
if logger.hasHandlers():
|
||||
logger.handlers.clear()
|
||||
|
||||
formatter = logging.Formatter(
|
||||
"%(asctime)s [%(levelname)s] %(name)s - %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
|
||||
if log_file is not None:
|
||||
log_path = Path(log_file)
|
||||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
file_handler = logging.FileHandler(log_path, encoding="utf-8")
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
file_handler.setFormatter(formatter)
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
if console_output:
|
||||
console_handler = logging.StreamHandler(sys.stdout)
|
||||
console_handler.setLevel(level)
|
||||
console_handler.setFormatter(formatter)
|
||||
logger.addHandler(console_handler)
|
||||
|
||||
return logger
|
||||
|
||||
|
||||
def get_logger(name: str) -> logging.Logger:
|
||||
"""获取指定名称的日志记录器
|
||||
|
||||
Args:
|
||||
name: 日志记录器名称，通常使用 __name__
|
||||
|
||||
Returns:
|
||||
日志记录器实例
|
||||
"""
|
||||
return logging.getLogger(name)
|
||||
@@ -0,0 +1,333 @@
|
||||
"""MIL 测试用例 Excel 生成模块"""
|
||||
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.worksheet.worksheet import Worksheet
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def read_excel_case(
|
||||
excel_path: str,
|
||||
data_dict: dict,
|
||||
addTimeEn: bool,
|
||||
) -> dict[str, Any]:
|
||||
"""读取 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 = {}
|
||||
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=1).value is not None:
|
||||
new_head = sheet.cell(row=row, column=1).value
|
||||
if new_head is None:
|
||||
raise CaseDataError(
|
||||
f"Excel文件 {excel_path} 格式错误，{sheet.title} 第 {row} 行第 1 列必须有标题名"
|
||||
)
|
||||
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=2).value == '完成测试':
|
||||
result_dict[sheet.title][new_head]["enable"] = True
|
||||
else:
|
||||
result_dict[sheet.title][new_head]["enable"] = False
|
||||
|
||||
step_name = sheet.cell(row, 4).value
|
||||
step_time = sheet.cell(row, 5).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} 行第 5 列时间必须是数字"
|
||||
)
|
||||
|
||||
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, 3).value
|
||||
if strings is not None:
|
||||
result_dict[sheet.title][new_head]["step"][f"step{step_id}"]["action"] = \
|
||||
__analysis_action(sheet, row, 3, 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 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)
|
||||
new_wb.save(generate_path)
|
||||
|
||||
logger.info(f"测试用例生成完成，共 {case_count} 个用例")
|
||||
|
||||
|
||||
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"]
|
||||
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"
|
||||
|
||||
time_column = 2
|
||||
while sheet.cell(1, time_column).value != "time":
|
||||
time_column += 1
|
||||
row = 1
|
||||
while sheet.cell(source_row + row, time_column).value is not None:
|
||||
row += 1
|
||||
stop_time = sheet.cell(source_row + row - 1, time_column).value
|
||||
if stop_time < datalog[-1].time:
|
||||
sheet.cell(source_row + row - 1, time_column).value = datalog[-1].time
|
||||
sheet.cell(source_row + row - 1, time_column).data_type = "float"
|
||||
|
||||
|
||||
def __get_template_version(sheet: Worksheet) -> str | None:
|
||||
"""获取 Excel 模板版本号
|
||||
|
||||
Args:
|
||||
sheet: Excel 工作表
|
||||
|
||||
Returns:
|
||||
模板版本号，如果未找到返回 None
|
||||
"""
|
||||
row = 2
|
||||
version = None
|
||||
while sheet.cell(row=row, column=1).value is not None:
|
||||
version = sheet.cell(row=row, column=1).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 = {}
|
||||
strings = strings.replace("；", ";")
|
||||
strings = strings.replace("\n", "").split(';')
|
||||
for string in strings:
|
||||
if string == '':
|
||||
continue
|
||||
string = string.replace(" ", "")
|
||||
signal_name = string[:string.find('=')]
|
||||
if signal_name.lower() not in [k.strip().lower() for k in data_dict.keys()]:
|
||||
raise CaseDataError(
|
||||
f"{sheet.title} 第 {row} 行第 3 列操作中信号 {signal_name} 不存在"
|
||||
)
|
||||
else:
|
||||
action_dict[signal_name] = string[string.find('=') + 1:]
|
||||
return action_dict
|
||||
|
||||
|
||||
def __analysis_case(data_dict: dict, sheets_dict: dict) -> dict:
|
||||
"""分析用例字典，将用例中的信号名替换为信号值
|
||||
|
||||
Args:
|
||||
data_dict: 数据字典
|
||||
sheets_dict: 工作表字典
|
||||
|
||||
Returns:
|
||||
处理后的用例字典
|
||||
"""
|
||||
datas_dict = {}
|
||||
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 = None
|
||||
for name in action_dict.keys():
|
||||
matched_key = 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,185 @@
|
||||
"""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__)
|
||||
|
||||
|
||||
@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
|
||||
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
|
||||
) -> dict[str, Any] | 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: dict[str, SignalData] = {}
|
||||
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 == "Parameter:" or name == "Value" or name == "BlockPath":
|
||||
column += 1
|
||||
continue
|
||||
sig_type = sheet.cell(config.type_row, column).value
|
||||
|
||||
signals[name] = SignalData(
|
||||
signal_type=sig_type,
|
||||
column=column,
|
||||
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
|
||||
|
||||
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: dict[str, Any] = {
|
||||
"wb": wb,
|
||||
"sheet": sheet,
|
||||
"source_row": source_row
|
||||
}
|
||||
for name, signal in signals.items():
|
||||
data[name] = signal.to_dict()
|
||||
|
||||
return data
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user