247 lines
9.5 KiB
Python
247 lines
9.5 KiB
Python
"""flatness-suite 批量测量结果的模板关联与旧报表解析。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import posixpath
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from zipfile import BadZipFile, ZipFile
|
|
from xml.etree import ElementTree as ET
|
|
|
|
|
|
_SHEET_NS = {"a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
|
|
_REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships"
|
|
|
|
|
|
def _column_index(reference: str) -> int:
|
|
letters = "".join(character for character in reference if character.isalpha())
|
|
if not letters:
|
|
raise ValueError(f"无效的 Excel 单元格引用: {reference}")
|
|
result = 0
|
|
for character in letters.upper():
|
|
result = result * 26 + ord(character) - ord("A") + 1
|
|
return result
|
|
|
|
|
|
def _cell_value(cell: ET.Element, shared_strings: list[str]) -> Any:
|
|
cell_type = cell.attrib.get("t")
|
|
if cell_type == "inlineStr":
|
|
return "".join(text.text or "" for text in cell.findall(".//a:t", _SHEET_NS))
|
|
|
|
value_node = cell.find("a:v", _SHEET_NS)
|
|
if value_node is None or value_node.text is None:
|
|
return ""
|
|
raw = value_node.text
|
|
if cell_type == "s":
|
|
try:
|
|
return shared_strings[int(raw)]
|
|
except (IndexError, ValueError) as exc:
|
|
raise ValueError("Excel 共享字符串索引无效") from exc
|
|
if cell_type in {"str", "e"}:
|
|
return raw
|
|
if cell_type == "b":
|
|
return raw == "1"
|
|
try:
|
|
number = float(raw)
|
|
return int(number) if number.is_integer() else number
|
|
except ValueError:
|
|
return raw
|
|
|
|
|
|
def read_xlsx_rows(path: str | Path) -> tuple[str, list[list[Any]]]:
|
|
"""只用标准库读取 xlsx 的第一个工作表。"""
|
|
workbook_path = Path(path)
|
|
if not workbook_path.is_file():
|
|
raise FileNotFoundError(f"Excel 文件不存在: {workbook_path}")
|
|
|
|
try:
|
|
with ZipFile(workbook_path) as xlsx:
|
|
names = set(xlsx.namelist())
|
|
shared_strings: list[str] = []
|
|
if "xl/sharedStrings.xml" in names:
|
|
shared_root = ET.fromstring(xlsx.read("xl/sharedStrings.xml"))
|
|
shared_strings = [
|
|
"".join(text.text or "" for text in item.findall(".//a:t", _SHEET_NS))
|
|
for item in shared_root.findall("a:si", _SHEET_NS)
|
|
]
|
|
|
|
workbook_root = ET.fromstring(xlsx.read("xl/workbook.xml"))
|
|
first_sheet = workbook_root.find("a:sheets/a:sheet", _SHEET_NS)
|
|
if first_sheet is None:
|
|
raise ValueError(f"Excel 中没有工作表: {workbook_path}")
|
|
sheet_name = first_sheet.attrib.get("name", "Sheet1")
|
|
relation_id = first_sheet.attrib.get(f"{{{_REL_NS}}}id")
|
|
if not relation_id:
|
|
raise ValueError(f"Excel 工作表关系缺失: {workbook_path}")
|
|
|
|
rels_root = ET.fromstring(xlsx.read("xl/_rels/workbook.xml.rels"))
|
|
relation_targets = {
|
|
relation.attrib["Id"]: relation.attrib["Target"] for relation in rels_root
|
|
}
|
|
target = relation_targets.get(relation_id)
|
|
if not target:
|
|
raise ValueError(f"Excel 工作表文件关系缺失: {workbook_path}")
|
|
sheet_path = target.lstrip("/") if target.startswith("/") else posixpath.normpath(posixpath.join("xl", target))
|
|
if sheet_path not in names:
|
|
raise ValueError(f"Excel 工作表文件不存在: {sheet_path}")
|
|
|
|
sheet_root = ET.fromstring(xlsx.read(sheet_path))
|
|
sparse_rows: list[dict[int, Any]] = []
|
|
max_column = 0
|
|
for row in sheet_root.findall("a:sheetData/a:row", _SHEET_NS):
|
|
values: dict[int, Any] = {}
|
|
for cell in row.findall("a:c", _SHEET_NS):
|
|
column = _column_index(cell.attrib.get("r", ""))
|
|
values[column] = _cell_value(cell, shared_strings)
|
|
max_column = max(max_column, column)
|
|
if values:
|
|
sparse_rows.append(values)
|
|
except (BadZipFile, KeyError, ET.ParseError) as exc:
|
|
raise ValueError(f"无法解析 Excel 文件 {workbook_path}: {exc}") from exc
|
|
|
|
rows = [
|
|
[row.get(column, "") for column in range(1, max_column + 1)]
|
|
for row in sparse_rows
|
|
]
|
|
if not rows:
|
|
raise ValueError(f"Excel 工作表为空: {workbook_path}")
|
|
return sheet_name, rows
|
|
|
|
|
|
def _header_index(headers: list[Any], candidates: tuple[str, ...]) -> int:
|
|
normalized = [str(header).strip() for header in headers]
|
|
for candidate in candidates:
|
|
if candidate in normalized:
|
|
return normalized.index(candidate)
|
|
raise ValueError(f"Excel 缺少列: {'/'.join(candidates)}")
|
|
|
|
|
|
def normalize_task_id(value: Any) -> str:
|
|
if isinstance(value, float) and value.is_integer():
|
|
return str(int(value))
|
|
text = str(value).strip()
|
|
return text[:-2] if text.endswith(".0") and text[:-2].isdigit() else text
|
|
|
|
|
|
def _label(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, float) and value.is_integer():
|
|
return str(int(value))
|
|
return str(value).strip()
|
|
|
|
|
|
def read_measurement_template(path: str | Path) -> dict[str, dict[str, str]]:
|
|
"""读取任务号对应的项点名称与原始公差文本。"""
|
|
_, rows = read_xlsx_rows(path)
|
|
headers = rows[0]
|
|
task_column = _header_index(headers, ("任务号", "任务编号", "task_id"))
|
|
name_column = _header_index(headers, ("项点名称", "项目名称", "项目"))
|
|
tolerance_column = _header_index(headers, ("公差要求", "公差"))
|
|
|
|
result: dict[str, dict[str, str]] = {}
|
|
for row in rows[1:]:
|
|
task_id = normalize_task_id(row[task_column] if task_column < len(row) else "")
|
|
if not task_id:
|
|
continue
|
|
item_name = _label(row[name_column] if name_column < len(row) else "")
|
|
if not item_name:
|
|
raise ValueError(f"模板任务 {task_id} 缺少项点名称")
|
|
result[task_id] = {
|
|
"item_name": item_name,
|
|
"tolerance_label": _label(row[tolerance_column] if tolerance_column < len(row) else ""),
|
|
}
|
|
if not result:
|
|
raise ValueError(f"模板中没有测量任务: {path}")
|
|
return result
|
|
|
|
|
|
def read_legacy_batch_rows(path: str | Path) -> list[dict[str, Any]]:
|
|
"""读取旧版技术汇总表中的任务号、平均值和状态。"""
|
|
_, rows = read_xlsx_rows(path)
|
|
headers = rows[0]
|
|
task_column = _header_index(headers, ("任务号", "任务编号", "task_id"))
|
|
mean_column = _header_index(headers, ("平均值", "mean", "mean_dist", "average_distance"))
|
|
status_column = _header_index(headers, ("状态", "测量结论", "结论"))
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for row in rows[1:]:
|
|
task_id = normalize_task_id(row[task_column] if task_column < len(row) else "")
|
|
if not task_id:
|
|
continue
|
|
result.append({
|
|
"task_id": task_id,
|
|
"mean": row[mean_column] if mean_column < len(row) else None,
|
|
"status": row[status_column] if status_column < len(row) else "",
|
|
})
|
|
if not result:
|
|
raise ValueError(f"旧版 Excel 中没有批量结果: {path}")
|
|
return result
|
|
|
|
|
|
def _finite_float(value: Any) -> float | None:
|
|
try:
|
|
number = float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return number if math.isfinite(number) else None
|
|
|
|
|
|
def build_measurements(
|
|
rows: list[dict[str, Any]],
|
|
template_path: str | Path,
|
|
) -> list[dict[str, Any]]:
|
|
template = read_measurement_template(template_path)
|
|
status_fields = {
|
|
"成功": ("pass", "成功"),
|
|
"部分成功": ("warning", "部分成功"),
|
|
"失败": ("fail", "失败"),
|
|
}
|
|
measurements: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
task_id = normalize_task_id(row.get("task_id", ""))
|
|
if not task_id:
|
|
raise ValueError("批量结果包含空任务号")
|
|
template_row = template.get(task_id)
|
|
if template_row is None:
|
|
raise ValueError(f"模板中找不到任务 {task_id}")
|
|
raw_status = str(row.get("status", "")).strip()
|
|
if raw_status not in status_fields:
|
|
raise ValueError(f"任务 {task_id} 的状态无效: {raw_status or '<空>'}")
|
|
status, _ = status_fields[raw_status]
|
|
measurements.append({
|
|
"name": f"任务{task_id} · {template_row['item_name']}",
|
|
"value": _finite_float(row.get("mean")),
|
|
"unit": "mm",
|
|
"toleranceLabel": template_row["tolerance_label"],
|
|
"status": status,
|
|
"passLabel": "成功",
|
|
"warningLabel": "部分成功",
|
|
"failLabel": "失败",
|
|
})
|
|
return measurements
|
|
|
|
|
|
def measurements_from_batch_result(result: dict[str, Any], template_path: str | Path) -> list[dict[str, Any]]:
|
|
rows = result.get("rows")
|
|
if not isinstance(rows, list):
|
|
raise ValueError("run_contour_batch 返回值缺少 rows")
|
|
return build_measurements(rows, template_path)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="从旧版 flatness-suite Excel 构造通用 measurements")
|
|
parser.add_argument("--excel", required=True)
|
|
parser.add_argument("--template", required=True)
|
|
args = parser.parse_args()
|
|
measurements = build_measurements(read_legacy_batch_rows(args.excel), args.template)
|
|
print(json.dumps(measurements, ensure_ascii=False, allow_nan=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|