优化代码
This commit is contained in:
@@ -14,18 +14,9 @@ def main() -> None:
|
|||||||
"""MIL SDK 示例程序主函数"""
|
"""MIL SDK 示例程序主函数"""
|
||||||
setup_logging()
|
setup_logging()
|
||||||
|
|
||||||
sample_path = Path(__file__).parent / "sample.xlsx"
|
|
||||||
case_path = Path(__file__).parent / "case.xlsx"
|
|
||||||
excel_path = Path(__file__).parent
|
excel_path = Path(__file__).parent
|
||||||
|
|
||||||
print(f"读取文件: {sample_path}")
|
|
||||||
data_dict = read_excel_data("new_data.xlsx", return_object=False)
|
data_dict = read_excel_data("new_data.xlsx", return_object=False)
|
||||||
# print(data_dict)
|
|
||||||
|
|
||||||
# case_dict = read_excel_case(case_path, data_dict, True)
|
|
||||||
# create_excel_case(excel_path, data_dict, case_dict)
|
|
||||||
# print(case_dict)
|
|
||||||
|
|
||||||
old_data_dict = read_excel_data("old_data.xlsx", return_object=False)
|
old_data_dict = read_excel_data("old_data.xlsx", return_object=False)
|
||||||
update_case_excel("new_case.xlsx", old_data_dict, data_dict)
|
update_case_excel("new_case.xlsx", old_data_dict, data_dict)
|
||||||
|
|
||||||
|
|||||||
@@ -11,7 +11,7 @@
|
|||||||
import logging
|
import logging
|
||||||
|
|
||||||
from .base import DataLog, SignalData, ExcelDataResult
|
from .base import DataLog, SignalData, ExcelDataResult
|
||||||
from .mil_read_data_excel import read_excel_data, get_data_log, ExcelReaderConfig
|
from .mil_read_data_excel import read_excel_data, ExcelReaderConfig
|
||||||
from .mil_read_case_excel import read_excel_case
|
from .mil_read_case_excel import read_excel_case
|
||||||
from .mil_create_data_excel import create_excel_case
|
from .mil_create_data_excel import create_excel_case
|
||||||
from .mil_update_excel import update_case_excel
|
from .mil_update_excel import update_case_excel
|
||||||
@@ -31,9 +31,9 @@ __all__ = [
|
|||||||
"ExcelDataResult",
|
"ExcelDataResult",
|
||||||
"ExcelReaderConfig",
|
"ExcelReaderConfig",
|
||||||
"read_excel_data",
|
"read_excel_data",
|
||||||
"get_data_log",
|
|
||||||
"read_excel_case",
|
"read_excel_case",
|
||||||
"create_excel_case",
|
"create_excel_case",
|
||||||
|
"update_case_excel",
|
||||||
"MILSDKError",
|
"MILSDKError",
|
||||||
"ExcelReadError",
|
"ExcelReadError",
|
||||||
"ExcelFormatError",
|
"ExcelFormatError",
|
||||||
|
|||||||
+4
-2
@@ -24,10 +24,12 @@ class SignalData:
|
|||||||
column: 列索引
|
column: 列索引
|
||||||
datalog: 数据日志列表
|
datalog: 数据日志列表
|
||||||
"""
|
"""
|
||||||
signal_type: str | None = None
|
# signal_type: str | None = None
|
||||||
column: int = 0
|
column: int = 0
|
||||||
|
attributes: dict[str, str] = field(default_factory=dict)
|
||||||
datalog: list[DataLog] = field(default_factory=list)
|
datalog: list[DataLog] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
def to_dict(self) -> dict[str, Any]:
|
def to_dict(self) -> dict[str, Any]:
|
||||||
"""转换为字典格式
|
"""转换为字典格式
|
||||||
|
|
||||||
@@ -35,8 +37,8 @@ class SignalData:
|
|||||||
包含 signal_type、column、datalog 的字典
|
包含 signal_type、column、datalog 的字典
|
||||||
"""
|
"""
|
||||||
return {
|
return {
|
||||||
"type": self.signal_type,
|
|
||||||
"column": self.column,
|
"column": self.column,
|
||||||
|
"attributes": self.attributes,
|
||||||
"datalog": self.datalog
|
"datalog": self.datalog
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -59,13 +59,21 @@ def create_excel_case(
|
|||||||
generate_path = f"{excel_path}/{name}.xlsx"
|
generate_path = f"{excel_path}/{name}.xlsx"
|
||||||
_write_excel_data(sheet, datas_dict[name], source_row)
|
_write_excel_data(sheet, datas_dict[name], source_row)
|
||||||
|
|
||||||
|
wb.save(generate_path)
|
||||||
|
|
||||||
logger.info(f"已生成用例: {name},路径: {generate_path}")
|
logger.info(f"已生成用例: {name},路径: {generate_path}")
|
||||||
|
|
||||||
logger.info(f"测试用例生成完成,共 {case_count} 个用例")
|
logger.info(f"测试用例生成完成,共 {case_count} 个用例")
|
||||||
|
|
||||||
|
|
||||||
def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> None:
|
def _write_excel_data(sheet: Worksheet, data_dict: dict, source_row: int) -> None:
|
||||||
"""写入 Excel 数据"""
|
"""写入 Excel 数据
|
||||||
|
|
||||||
|
Args:
|
||||||
|
sheet: Worksheet 对象
|
||||||
|
data_dict: 数据字典
|
||||||
|
source_row: Source: Input 所在行号
|
||||||
|
"""
|
||||||
for name in data_dict.keys():
|
for name in data_dict.keys():
|
||||||
column = data_dict[name]["column"]
|
column = data_dict[name]["column"]
|
||||||
datalog = data_dict[name]["datalog"]
|
datalog = data_dict[name]["datalog"]
|
||||||
|
|||||||
@@ -40,6 +40,7 @@ class ExcelReaderConfig:
|
|||||||
time_column: int = 1
|
time_column: int = 1
|
||||||
header_row: int = 1
|
header_row: int = 1
|
||||||
type_row: int = 3
|
type_row: int = 3
|
||||||
|
interp_row: int = 6
|
||||||
data_start_row_offset: int = 1
|
data_start_row_offset: int = 1
|
||||||
|
|
||||||
output_header: str = "Source: Output"
|
output_header: str = "Source: Output"
|
||||||
@@ -124,12 +125,13 @@ def read_excel_data(
|
|||||||
if name in ("Parameter:", "Value", "BlockPath"):
|
if name in ("Parameter:", "Value", "BlockPath"):
|
||||||
column += 1
|
column += 1
|
||||||
continue
|
continue
|
||||||
sig_type = sheet.cell(config.type_row, column).value
|
attributes = __get_signal_attributes(sheet, column, source_row)
|
||||||
|
if attributes is None:
|
||||||
|
break
|
||||||
signals[name] = SignalData(
|
signals[name] = SignalData(
|
||||||
signal_type=sig_type,
|
|
||||||
column=column,
|
column=column,
|
||||||
datalog=get_data_log(sheet, source_row + config.data_start_row_offset, column, config.time_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)}")
|
logger.debug(f"解析信号: {name}, 数据点: {len(signals[name].datalog)}")
|
||||||
else:
|
else:
|
||||||
@@ -157,7 +159,27 @@ def read_excel_data(
|
|||||||
return data
|
return data
|
||||||
|
|
||||||
|
|
||||||
def get_data_log(
|
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,
|
sheet: Worksheet,
|
||||||
row: int,
|
row: int,
|
||||||
column: int,
|
column: int,
|
||||||
|
|||||||
@@ -30,11 +30,11 @@ def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None:
|
|||||||
|
|
||||||
old_wb = old_data.pop('wb')
|
old_wb = old_data.pop('wb')
|
||||||
old_sheet = old_data.pop('sheet')
|
old_sheet = old_data.pop('sheet')
|
||||||
source_row = old_data.pop('source_row')
|
old_source_row = old_data.pop('source_row')
|
||||||
|
|
||||||
new_data.pop('wb')
|
new_data.pop('wb')
|
||||||
new_data.pop('sheet')
|
new_data.pop('sheet')
|
||||||
new_data.pop('source_row')
|
new_source_row = new_data.pop('source_row')
|
||||||
|
|
||||||
if not isinstance(old_wb, Workbook) or not isinstance(old_sheet, Worksheet):
|
if not isinstance(old_wb, Workbook) or not isinstance(old_sheet, Worksheet):
|
||||||
logger.error("数据类型错误,Workbook 或 Worksheet 类型不匹配")
|
logger.error("数据类型错误,Workbook 或 Worksheet 类型不匹配")
|
||||||
@@ -51,13 +51,15 @@ def update_case_excel(filename: str, old_data: dict, new_data: dict) -> None:
|
|||||||
for name, info in add_input.items():
|
for name, info in add_input.items():
|
||||||
old_sheet.insert_cols(info['column'])
|
old_sheet.insert_cols(info['column'])
|
||||||
|
|
||||||
old_sheet.cell(1, info['column']).value = name
|
for index in range(1,old_source_row):
|
||||||
old_sheet.cell(1, info['column']).data_type = "str"
|
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"
|
||||||
|
|
||||||
old_sheet.cell(3, info['column']).value = info['type']
|
for index in range(old_source_row + 1, datalog_len + old_source_row + 1):
|
||||||
old_sheet.cell(3, info['column']).data_type = "str"
|
|
||||||
|
|
||||||
for index in range(source_row + 1, datalog_len + source_row + 1):
|
|
||||||
try:
|
try:
|
||||||
value = int(info['datalog'][0].value)
|
value = int(info['datalog'][0].value)
|
||||||
old_sheet.cell(index, info['column']).value = value
|
old_sheet.cell(index, info['column']).value = value
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import pytest
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from openpyxl import Workbook
|
from openpyxl import Workbook
|
||||||
|
|
||||||
from src.core.mil_read_data_excel import read_excel_data, get_data_log, ExcelReaderConfig
|
from src.core.mil_read_data_excel import read_excel_data, ExcelReaderConfig
|
||||||
from src.core.exceptions import ExcelReadError, ExcelFormatError
|
from src.core.exceptions import ExcelReadError, ExcelFormatError
|
||||||
|
|
||||||
|
|
||||||
@@ -65,12 +65,7 @@ def test_datalog_is_list(sample_excel_path: Path):
|
|||||||
assert hasattr(first_signal["datalog"][0], "value")
|
assert hasattr(first_signal["datalog"][0], "value")
|
||||||
|
|
||||||
|
|
||||||
def test_get_data_log_empty_sheet():
|
|
||||||
"""空 sheet 返回空列表"""
|
|
||||||
wb = Workbook()
|
|
||||||
sheet = wb.active
|
|
||||||
datalog = get_data_log(sheet, row=1, column=1)
|
|
||||||
assert len(datalog) == 0
|
|
||||||
|
|
||||||
|
|
||||||
def test_read_excel_data_return_object(sample_excel_path: Path):
|
def test_read_excel_data_return_object(sample_excel_path: Path):
|
||||||
|
|||||||
Reference in New Issue
Block a user