|
@@ -1,8 +1,9 @@
|
|
|
"""故障 → 3D 部件对照表(data/fault_part_map.csv 的读写)。
|
|
"""故障 → 3D 部件对照表(data/fault_part_map.csv 的读写)。
|
|
|
|
|
|
|
|
-列:model, alarm_type, device_point, device_part, part
|
|
|
|
|
|
|
+列:model, alarm_type, device_point, device_part, part, camera_x/y/z, target_x/y/z
|
|
|
- device_point / device_part 支持多值,用 ``|`` 分隔(未来一个故障多个点)。
|
|
- device_point / device_part 支持多值,用 ``|`` 分隔(未来一个故障多个点)。
|
|
|
- part 为前端部件标签(createUniqueName 约定:如 ``三维立体_1687``);一个部件一行。
|
|
- part 为前端部件标签(createUniqueName 约定:如 ``三维立体_1687``);一个部件一行。
|
|
|
|
|
+- camera_/target_ 为该故障的“场景位置”(参考系坐标,空=未记录)。
|
|
|
"""
|
|
"""
|
|
|
|
|
|
|
|
import csv
|
|
import csv
|
|
@@ -11,9 +12,12 @@ from typing import Any
|
|
|
|
|
|
|
|
from config import DATA_DIR, FAULT_MAP_CSV
|
|
from config import DATA_DIR, FAULT_MAP_CSV
|
|
|
|
|
|
|
|
-FIELDS = ["model", "alarm_type", "device_point", "device_part", "part"]
|
|
|
|
|
|
|
+FIELDS = [
|
|
|
|
|
+ "model", "alarm_type", "device_point", "device_part", "part",
|
|
|
|
|
+ "camera_x", "camera_y", "camera_z", "target_x", "target_y", "target_z",
|
|
|
|
|
+]
|
|
|
|
|
|
|
|
-# 初始种子(一个部件一行)。
|
|
|
|
|
|
|
+# 初始种子(一个部件一行,场景位置留空)。
|
|
|
SEED: list[list[str]] = [
|
|
SEED: list[list[str]] = [
|
|
|
["压缩机2326-3000-one.glb", "润滑油压力故障", "YSJ_5", "润滑油压力", "三维立体_1687"],
|
|
["压缩机2326-3000-one.glb", "润滑油压力故障", "YSJ_5", "润滑油压力", "三维立体_1687"],
|
|
|
["压缩机2326-3000-one.glb", "振动故障", "YSJ_11", "链轮端振动", "三维立体_3815"],
|
|
["压缩机2326-3000-one.glb", "振动故障", "YSJ_11", "链轮端振动", "三维立体_3815"],
|
|
@@ -36,20 +40,80 @@ def _ensure_file() -> None:
|
|
|
with FAULT_MAP_CSV.open("w", encoding="utf-8", newline="") as handle:
|
|
with FAULT_MAP_CSV.open("w", encoding="utf-8", newline="") as handle:
|
|
|
writer = csv.writer(handle)
|
|
writer = csv.writer(handle)
|
|
|
writer.writerow(FIELDS)
|
|
writer.writerow(FIELDS)
|
|
|
- writer.writerows(SEED)
|
|
|
|
|
|
|
+ for row in SEED:
|
|
|
|
|
+ writer.writerow(row + [""] * 6)
|
|
|
|
|
|
|
|
|
|
|
|
|
-def list_rows() -> list[dict[str, Any]]:
|
|
|
|
|
|
|
+def _read() -> list[dict[str, str]]:
|
|
|
_ensure_file()
|
|
_ensure_file()
|
|
|
with FAULT_MAP_CSV.open("r", encoding="utf-8", newline="") as handle:
|
|
with FAULT_MAP_CSV.open("r", encoding="utf-8", newline="") as handle:
|
|
|
return [
|
|
return [
|
|
|
- {
|
|
|
|
|
- "model": row["model"].strip(),
|
|
|
|
|
- "alarm_type": row["alarm_type"].strip(),
|
|
|
|
|
- "device_point": row["device_point"].strip(),
|
|
|
|
|
- "device_part": row["device_part"].strip(),
|
|
|
|
|
- "part": row["part"].strip(),
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ {key: (row.get(key) or "").strip() for key in FIELDS}
|
|
|
for row in csv.DictReader(handle)
|
|
for row in csv.DictReader(handle)
|
|
|
- if row.get("part")
|
|
|
|
|
]
|
|
]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _write(rows: list[dict[str, str]]) -> None:
|
|
|
|
|
+ DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
+ tmp = FAULT_MAP_CSV.parent / (FAULT_MAP_CSV.name + ".tmp")
|
|
|
|
|
+ with tmp.open("w", encoding="utf-8", newline="") as handle:
|
|
|
|
|
+ writer = csv.writer(handle)
|
|
|
|
|
+ writer.writerow(FIELDS)
|
|
|
|
|
+ for row in rows:
|
|
|
|
|
+ writer.writerow([row.get(key, "") for key in FIELDS])
|
|
|
|
|
+ os.replace(tmp, FAULT_MAP_CSV)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _split(value: str) -> list[str]:
|
|
|
|
|
+ return [item.strip() for item in str(value or "").split("|") if item.strip()]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _vector(row: dict[str, str], x: str, y: str, z: str) -> list[float] | None:
|
|
|
|
|
+ try:
|
|
|
|
|
+ return [float(row[x]), float(row[y]), float(row[z])]
|
|
|
|
|
+ except (ValueError, KeyError):
|
|
|
|
|
+ return None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def _to_api(row: dict[str, str]) -> dict[str, Any]:
|
|
|
|
|
+ return {
|
|
|
|
|
+ "model": row["model"],
|
|
|
|
|
+ "alarm_type": row["alarm_type"],
|
|
|
|
|
+ "device_point": row["device_point"],
|
|
|
|
|
+ "device_part": row["device_part"],
|
|
|
|
|
+ "part": row["part"],
|
|
|
|
|
+ "camera": _vector(row, "camera_x", "camera_y", "camera_z"),
|
|
|
|
|
+ "target": _vector(row, "target_x", "target_y", "target_z"),
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def list_rows() -> list[dict[str, Any]]:
|
|
|
|
|
+ return [_to_api(row) for row in _read() if row.get("part")]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def set_scene_position(
|
|
|
|
|
+ model: str,
|
|
|
|
|
+ alarm_type: str,
|
|
|
|
|
+ device_point: str,
|
|
|
|
|
+ camera: list[float],
|
|
|
|
|
+ target: list[float],
|
|
|
|
|
+) -> int:
|
|
|
|
|
+ rows = _read()
|
|
|
|
|
+ wanted = set(_split(device_point))
|
|
|
|
|
+ changed = 0
|
|
|
|
|
+ for row in rows:
|
|
|
|
|
+ if row["model"] != model or row["alarm_type"] != alarm_type:
|
|
|
|
|
+ continue
|
|
|
|
|
+ row_points = set(_split(row["device_point"]))
|
|
|
|
|
+ if wanted and not (wanted & row_points):
|
|
|
|
|
+ continue
|
|
|
|
|
+ row["camera_x"], row["camera_y"], row["camera_z"] = (
|
|
|
|
|
+ str(camera[0]), str(camera[1]), str(camera[2]),
|
|
|
|
|
+ )
|
|
|
|
|
+ row["target_x"], row["target_y"], row["target_z"] = (
|
|
|
|
|
+ str(target[0]), str(target[1]), str(target[2]),
|
|
|
|
|
+ )
|
|
|
|
|
+ changed += 1
|
|
|
|
|
+ if changed:
|
|
|
|
|
+ _write(rows)
|
|
|
|
|
+ return changed
|