Compressor/viewer/backend/services/fault_map_service.py

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

列：model, alarm_type, device_point, device_part, part, camera_x/y/z, target_x/y/z
- device_point / device_part 支持多值，用 ``|`` 分隔（未来一个故障多个点）。
- part 为前端部件标签（createUniqueName 约定：如 ``三维立体_1687``）；一个部件一行。
- camera_/target_ 为该故障的“场景位置”（参考系坐标，空=未记录）。
"""

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",
    "camera_x", "camera_y", "camera_z", "target_x", "target_y", "target_z",
]

# 初始种子（一个部件一行，场景位置留空）。
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)
        for row in SEED:
            writer.writerow(row + [""] * 6)


def _read() -> list[dict[str, str]]:
    _ensure_file()
    with FAULT_MAP_CSV.open("r", encoding="utf-8", newline="") as handle:
        return [
            {key: (row.get(key) or "").strip() for key in FIELDS}
            for row in csv.DictReader(handle)
        ]


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 remap_parts(model: str, mapping: dict[str, str]) -> int:
    """按 旧label→新label 映射更新指定模型的 part 列，返回改动行数。"""
    if not mapping:
        return 0
    rows = _read()
    changed = 0
    for row in rows:
        if row["model"] != model:
            continue
        part = row.get("part", "")
        if part in mapping:
            row["part"] = mapping[part]
            changed += 1
    if changed:
        _write(rows)
    return changed


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