Compressor/3d/backend/services/fault_map_service.py

"""故障 → 3D 部件对照表（data/fault_part_map.csv 的读写）。

列：model, alarm_type, device_point, device_part, part
- device_point / device_part 支持多值，用 ``|`` 分隔（未来一个故障多个点）。
- part 为前端部件标签（createUniqueName 约定：如 ``三维立体_1687``）；一个部件一行。
"""

import csv
import os
from typing import Any

from config import DATA_DIR, FAULT_MAP_CSV

FIELDS = ["model", "alarm_type", "device_point", "device_part", "part"]

# 初始种子（一个部件一行）。
SEED: list[list[str]] = [
    ["压缩机2326-3000-one.glb", "润滑油压力故障", "YSJ_5", "润滑油压力", "三维立体_1687"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_11", "链轮端振动", "三维立体_3815"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_11", "链轮端振动", "三维立体_3814"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_101"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_97"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_98"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_130"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_135"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_134"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_138"],
    ["压缩机2326-3000-one.glb", "振动故障", "YSJ_10", "联轴器端振动", "三维立体_3813"],
]


def _ensure_file() -> None:
    if FAULT_MAP_CSV.exists():
        return
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    with FAULT_MAP_CSV.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(FIELDS)
        writer.writerows(SEED)


def list_rows() -> list[dict[str, Any]]:
    _ensure_file()
    with FAULT_MAP_CSV.open("r", encoding="utf-8", newline="") as handle:
        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(),
            }
            for row in csv.DictReader(handle)
            if row.get("part")
        ]