133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
# ============================================================
|
|
# 康尼映射表解析契约黄金测试（真实数据验证工程师）
|
|
# 覆盖：设备能力/模具适配映射表表头契约、共享设备全工序覆盖、
|
|
# 映射表存在时 build_flex 优先读取并输出推断说明。
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.kangni_intake import (
|
|
build_flex_equipment_and_molds,
|
|
parse_equipment_capability_map,
|
|
parse_mold_adaptation_map,
|
|
)
|
|
from tests.external_data import external_dir
|
|
|
|
KANGNI_DATA_DIR = external_dir("KANGNI_DATA_DIR", "kangni")
|
|
EQUIP_HEADER = [
|
|
"设备编号", "设备名称", "可执行工序编号", "单件工时", "可动率",
|
|
"是否可移动", "移动耗时(分钟)", "区域编码", "适配模具编号", "状态", "备注",
|
|
]
|
|
MOLD_HEADER = [
|
|
"模具编号", "模具名称", "适用工序编号", "适配设备编号",
|
|
"寿命上限", "已用寿命", "区域编码", "换型耗时(分钟)", "状态", "备注",
|
|
]
|
|
|
|
|
|
def _write_equip_map(tmp_path: Path) -> Path:
|
|
from openpyxl import Workbook
|
|
|
|
p = tmp_path / "设备能力映射模板.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "设备能力"
|
|
ws.append(EQUIP_HEADER)
|
|
ws.append(["EQ-SHARED-CG-01", "共享装配设备（待现场确认）", "OP-1,OP-2",
|
|
"OP-1:27;OP-2:5", 0.95, "否", 0, "ZONE-CG", "", "RUNNING",
|
|
"保守假设，待现场确认"])
|
|
ws.append(["CO01001", "落地式直读光谱仪", "", "", 0.95, "否", 0,
|
|
"ZONE-CG", "", "RUNNING", "真实设备，能力待现场确认"])
|
|
wb.save(p)
|
|
return p
|
|
|
|
|
|
def _write_mold_map(tmp_path: Path) -> Path:
|
|
from openpyxl import Workbook
|
|
|
|
p = tmp_path / "模具适配映射模板.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "模具适配"
|
|
ws.append(MOLD_HEADER)
|
|
ws.append(["00000001", "胶条切割打孔工装", "", "", "", "", "ZONE-CG", "",
|
|
"AVAILABLE", "真实模具，工序/寿命待现场确认"])
|
|
wb.save(p)
|
|
return p
|
|
|
|
|
|
def test_equipment_map_parses_exact_header_contract(tmp_path: Path):
|
|
rows = parse_equipment_capability_map(_write_equip_map(tmp_path))
|
|
assert len(rows) == 2
|
|
shared = rows[0]
|
|
assert shared["code"] == "EQ-SHARED-CG-01"
|
|
assert shared["capabilities"] == ["OP-1", "OP-2"]
|
|
assert shared["opStdTime"] == {"OP-1": 27.0, "OP-2": 5.0}
|
|
assert shared["movable"] is False
|
|
assert shared["availabilityRate"] == 0.95
|
|
assert shared["status"] == "RUNNING"
|
|
assert rows[1]["code"] == "CO01001"
|
|
assert rows[1]["capabilities"] == []
|
|
|
|
|
|
def test_mold_map_parses_exact_header_contract(tmp_path: Path):
|
|
rows = parse_mold_adaptation_map(_write_mold_map(tmp_path))
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row["code"] == "00000001"
|
|
assert row["name"] == "胶条切割打孔工装"
|
|
assert row["operationCode"] == ""
|
|
assert row["adaptableEquipment"] == []
|
|
assert row["lifeTotal"] == 0.0
|
|
assert row["lifeUsed"] == 0.0
|
|
assert row["status"] == "AVAILABLE"
|
|
|
|
|
|
def test_build_flex_prefers_mapping_table_and_emits_inference(tmp_path: Path):
|
|
equip = _write_equip_map(tmp_path)
|
|
mold = _write_mold_map(tmp_path)
|
|
assert equip.parent == mold.parent
|
|
|
|
equipment, molds, mold_ops, infs = build_flex_equipment_and_molds(
|
|
tmp_path, [], {"OP-1": 27.0, "OP-2": 5.0}, ["OP-1", "OP-2"],
|
|
{"zone": {"code": "ZONE-CG"}, "availabilityRate": 0.95},
|
|
)
|
|
assert len(equipment) == 2
|
|
assert not equipment[0]["inferred"]
|
|
assert equipment[0]["capabilities"] == ["OP-1", "OP-2"]
|
|
assert len(molds) == 1
|
|
assert mold_ops == set()
|
|
assert any("设备能力映射表已读取" in x for x in infs)
|
|
assert any("模具适配映射表已读取" in x for x in infs)
|
|
|
|
|
|
@pytest.mark.skipif(not (KANGNI_DATA_DIR / "工时.xlsx").exists(),
|
|
reason="康尼现场工时表不在本机")
|
|
def test_shared_equipment_covers_all_real_kangni_operations():
|
|
from openpyxl import load_workbook
|
|
|
|
equip_path = KANGNI_DATA_DIR / "设备能力映射模板.xlsx"
|
|
if not equip_path.exists():
|
|
pytest.skip("康尼/数据 设备能力映射模板.xlsx 未生成")
|
|
|
|
wb = load_workbook(KANGNI_DATA_DIR / "工时.xlsx", read_only=True, data_only=True)
|
|
real_ops = set()
|
|
for sn in wb.sheetnames:
|
|
ws = wb[sn]
|
|
for row in ws.iter_rows(min_row=2, values_only=True):
|
|
code = str(row[1]).strip() if len(row) > 1 and row[1] is not None else ""
|
|
if code:
|
|
real_ops.add(code)
|
|
wb.close()
|
|
assert real_ops, "工时.xlsx 未读到任何工序编号"
|
|
|
|
rows = parse_equipment_capability_map(equip_path)
|
|
shared = [r for r in rows if r["code"] == "EQ-SHARED-CG-01"]
|
|
assert shared, "缺少共享设备占位行 EQ-SHARED-CG-01"
|
|
caps = set(shared[0]["capabilities"])
|
|
missing = sorted(real_ops - caps)
|
|
assert not missing, f"共享设备未覆盖真实工序: {missing}"
|
|
assert all(op in shared[0]["opStdTime"] for op in real_ops)
|