vd_test_fixture/edc-web/app/routes/fixture.py

"""工装配置 API"""

from flask import Blueprint, jsonify, render_template, request
from flask_login import login_required, current_user
from app.models import (
    get_device_by_id,
    insert_serialnet,
    get_fixture_param,
    upsert_fixture_param,
    get_vehicle_base_tests,
    get_vehicle_base_test_by_id,
    create_vehicle_base_test,
    update_vehicle_base_test,
    delete_vehicle_base_test,
    get_serialnet_by_id,
    insert_log,
    get_coil_info_list,
    get_coil_info_by_id,
    create_coil_info,
    update_coil_info,
    delete_coil_info,
    get_simulate_car_list,
    get_simulate_car_by_id,
    create_simulate_car,
    update_simulate_car,
    delete_simulate_car,
)

bp = Blueprint("fixture", __name__)

# DG430 工装配置指令 (addr=0x01)
FIXTURE_COMMANDS = {
    "4A": "7F81014ACACC",  # 获取设备版本号
    "4C": "7F81014CCCCE",  # 查询设备测试参数
    "4D": "7F81014DCDCF",  # 出厂初始化
    "4E": "7F81014ECED0",  # 设备复位
}

CMD_NAMES = {
    "4A": "获取设备版本号",
    "4B": "配置设备测试参数",
    "4C": "查询设备测试参数",
    "4D": "出厂初始化",
    "4E": "设备复位",
}


def _xor_sum(data: bytes) -> tuple[int, int]:
    """计算异或和校验 (从 ADDR 到 DATA 末)"""
    xor = 0
    s = 0
    for b in data:
        xor ^= b
        s += b
    return xor & 0xFF, s & 0xFF


def _le16(val: int) -> bytes:
    """int → 小端 2 字节"""
    return bytes([val & 0xFF, (val >> 8) & 0xFF])


def _be16(val: int) -> bytes:
    """int → 大端 2 字节"""
    return bytes([(val >> 8) & 0xFF, val & 0xFF])


def build_4b_packet(addr: int, dev_type: int, test_mode: int,
                    reset_dis: int, minus_dis: int,
                    sens_min: int, sens_max: int,
                    fre_min: int, fre_max: int,
                    peak_min: int, peak_max: int,
                    far_tol: int = 0, near_tol: int = 0,
                    step_tol: int = 0, back_forth: int = 0,
                    near_stay: int = 0, far_stay: int = 0) -> str:
    """构造 0x4B 配置指令 hex 字符串 (V2.0.3 扩展)

    格式: 7F | 81 | 17 | 4B | Addr(1) | DevType(1) | TestMode(1) |
          ResetDis(1) | MinusDis(1) |
          SensMin(2 LE) | SensMax(2 LE) | FreMin(2 LE) | FreMax(2 LE) |
          PeakMin(2 LE) | PeakMax(2 LE) |
          FarTol(1) | NearTol(1) | StepTol(1) | BackForth(1) |
          NearStay(2 LE) | FarStay(2 LE) | XOR | SUM
    """
    payload = bytes([
        0x4B,           # CMD
        addr & 0xFF,
        dev_type & 0xFF,
        test_mode & 0xFF,
        reset_dis & 0xFF,
        minus_dis & 0xFF,
    ])
    payload += (_le16(sens_min) + _le16(sens_max) +
                _le16(fre_min) + _le16(fre_max) +
                _le16(peak_min) + _le16(peak_max))
    # V2.0.3 波动参数
    payload += bytes([
        far_tol & 0xFF,
        near_tol & 0xFF,
        step_tol & 0xFF,
        back_forth & 0xFF,
    ])
    payload += _le16(near_stay) + _le16(far_stay)

    pkt = bytes([0x7F, 0x81, len(payload)]) + payload
    xor, total = _xor_sum(pkt[1:])
    pkt += bytes([xor, total])
    return pkt.hex().upper()


# ─── 页面 ───────────────────────────────────────────────────────────

@bp.route("/fixture/<int:dnt_id>")
@login_required
def fixture_page(dnt_id):
    """工装配置页面"""
    if current_user.role != "admin":
        return "无权限：仅管理员可访问工装配置", 403
    device = get_device_by_id(dnt_id)
    if not device:
        return "设备不存在", 404
    return render_template("fixture.html", device=device)


@bp.route("/vehicle-base-test")
@login_required
def vehicle_base_test_page():
    """车检器测试基准参数管理页面"""
    return render_template("vehicle_base_test.html")


# ─── 工装配置指令 API ──────────────────────────────────────────────

@bp.route("/api/fixture/command", methods=["POST"])
@login_required
def api_fixture_command():
    """发送工装配置指令 (0x4A/0x4B/0x4C/0x4D/0x4E)"""
    if current_user.role != "admin":
        return jsonify({"ok": False, "error": "无权限：仅管理员可执行工装指令"}), 403
    data = request.get_json()
    dnt_id = data.get("dnt_id")
    cmd = data.get("cmd", "").upper()

    device = get_device_by_id(dnt_id)
    target = f"{device['serial']}" if device else f"dnt_id={dnt_id}"

    if cmd == "4B":
        # 动态构造 0x4B 指令 (V2.0.3)
        params = data.get("params", {})
        send_pkg = build_4b_packet(
            addr=params.get("addr", 1),
            dev_type=params.get("dev_type", 0),
            test_mode=params.get("test_mode", 0),
            reset_dis=params.get("reset_dis", 0),
            minus_dis=params.get("minus_dis", 0),
            sens_min=params.get("sens_min", 0),
            sens_max=params.get("sens_max", 0),
            fre_min=params.get("fre_min", 0),
            fre_max=params.get("fre_max", 0),
            peak_min=params.get("peak_min", 0),
            peak_max=params.get("peak_max", 0),
            far_tol=params.get("far_tol", 0),
            near_tol=params.get("near_tol", 0),
            step_tol=params.get("step_tol", 0),
            back_forth=params.get("back_forth", 0),
            near_stay=params.get("near_stay", 0),
            far_stay=params.get("far_stay", 0),
        )
    elif cmd in FIXTURE_COMMANDS:
        send_pkg = FIXTURE_COMMANDS[cmd]
    else:
        return jsonify({"ok": False, "error": f"未知指令: {cmd}"}), 400

    cmd_name = CMD_NAMES.get(cmd, cmd)
    try:
        record_id = insert_serialnet(dnt_id, send_pkg)
        insert_log(
            current_user.id, current_user.username, "command",
            target=target,
            detail=f"工装配置: {cmd_name}(0x{cmd}) → {send_pkg}",
            result="ok",
            ip=request.remote_addr or "",
        )
        return jsonify({"ok": True, "record_id": record_id, "send_pkg": send_pkg})
    except Exception as e:
        insert_log(
            current_user.id, current_user.username, "command",
            target=target,
            detail=f"工装配置 {cmd_name}(0x{cmd}) 失败: {e}",
            result="error",
            ip=request.remote_addr or "",
        )
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/fixture/serialnet/<int:record_id>")
@login_required
def api_get_serialnet(record_id):
    """查询 tb_serialnet 记录状态和返回数据"""
    rec = get_serialnet_by_id(record_id)
    if not rec:
        return jsonify({"error": "记录不存在"}), 404
    return jsonify({
        "id": rec["id"],
        "state": rec["state"],
        "send_pkg": rec.get("send_pkg", ""),
        "rcv_pkg": rec.get("rcv_pkg", ""),
        "create_time": str(rec.get("create_time", "")),
        "update_time": str(rec.get("update_time", "")),
    })


# ─── 工装参数 CRUD API ─────────────────────────────────────────────

@bp.route("/api/fixture/param/<int:dnt_id>")
@login_required
def api_get_fixture_param(dnt_id):
    """获取工装测试参数"""
    param = get_fixture_param(dnt_id)
    return jsonify(param or {})


@bp.route("/api/fixture/param/<int:dnt_id>", methods=["POST"])
@login_required
def api_save_fixture_param(dnt_id):
    """保存工装测试参数（仅数据库，不下发设备）"""
    if current_user.role != "admin":
        return jsonify({"ok": False, "error": "无权限：仅管理员可修改工装参数"}), 403
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    upsert_fixture_param(dnt_id, **data)
    device = get_device_by_id(dnt_id)
    target = f"{device['serial']}" if device else f"dnt_id={dnt_id}"
    insert_log(
        current_user.id, current_user.username, "update",
        target=target,
        detail="保存工装配置参数",
        result="ok",
        ip=request.remote_addr or "",
    )
    return jsonify({"ok": True})


# ─── 车检器测试基准参数 CRUD API ──────────────────────────────────

@bp.route("/api/vehicle-base-test")
@login_required
def api_list_vehicle_base_tests():
    """列出车检器测试基准参数"""
    search = request.args.get("search", "")
    tests = get_vehicle_base_tests(search)
    return jsonify(tests)


@bp.route("/api/vehicle-base-test/<int:test_id>")
@login_required
def api_get_vehicle_base_test(test_id):
    """获取单个车检器测试基准"""
    test = get_vehicle_base_test_by_id(test_id)
    if not test:
        return jsonify({"error": "不存在"}), 404
    return jsonify(test)


@bp.route("/api/vehicle-base-test", methods=["POST"])
@login_required
def api_create_vehicle_base_test():
    """创建车检器测试基准"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        test_id = create_vehicle_base_test(**data)
        return jsonify({"ok": True, "id": test_id})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/vehicle-base-test/<int:test_id>", methods=["PUT"])
@login_required
def api_update_vehicle_base_test(test_id):
    """更新车检器测试基准"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        update_vehicle_base_test(test_id, **data)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/vehicle-base-test/<int:test_id>", methods=["DELETE"])
@login_required
def api_delete_vehicle_base_test(test_id):
    """删除车检器测试基准"""
    try:
        delete_vehicle_base_test(test_id)
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ─── 线圈参数页面 ──────────────────────────────────────────────────

@bp.route("/coil-info")
@login_required
def coil_info_page():
    """线圈参数管理页面"""
    return render_template("coil_info.html")


@bp.route("/api/coil-info")
@login_required
def api_list_coil_info():
    """列出线圈参数"""
    search = request.args.get("search", "")
    items = get_coil_info_list(search)
    return jsonify(items)


@bp.route("/api/coil-info/<int:coil_id>")
@login_required
def api_get_coil_info(coil_id):
    """获取单个线圈参数"""
    item = get_coil_info_by_id(coil_id)
    if not item:
        return jsonify({"error": "不存在"}), 404
    return jsonify(item)


@bp.route("/api/coil-info", methods=["POST"])
@login_required
def api_create_coil_info():
    """创建线圈参数"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        coil_id = create_coil_info(**data)
        insert_log(
            current_user.id, current_user.username, "create",
            target="coil_info",
            detail=f"创建线圈: {data.get('coil_num','')} {data.get('name','')}",
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True, "id": coil_id})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/coil-info/<int:coil_id>", methods=["PUT"])
@login_required
def api_update_coil_info(coil_id):
    """更新线圈参数"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        update_coil_info(coil_id, **data)
        insert_log(
            current_user.id, current_user.username, "update",
            target="coil_info",
            detail=f"更新线圈 id={coil_id}: {data.get('coil_num','')} {data.get('name','')}",
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/coil-info/<int:coil_id>", methods=["DELETE"])
@login_required
def api_delete_coil_info(coil_id):
    """删除线圈参数"""
    try:
        item = get_coil_info_by_id(coil_id)
        detail = f"删除线圈 id={coil_id}"
        if item:
            detail += f": {item.get('coil_num','')} {item.get('name','')}"
        delete_coil_info(coil_id)
        insert_log(
            current_user.id, current_user.username, "delete",
            target="coil_info",
            detail=detail,
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


# ─── 模拟车辆参数页面 ──────────────────────────────────────────────

@bp.route("/simulate-car")
@login_required
def simulate_car_page():
    """模拟车辆参数管理页面"""
    return render_template("simulate_car.html")


@bp.route("/api/simulate-car")
@login_required
def api_list_simulate_car():
    """列出模拟车辆参数"""
    search = request.args.get("search", "")
    items = get_simulate_car_list(search)
    return jsonify(items)


@bp.route("/api/simulate-car/<int:car_id>")
@login_required
def api_get_simulate_car(car_id):
    """获取单个模拟车辆参数"""
    item = get_simulate_car_by_id(car_id)
    if not item:
        return jsonify({"error": "不存在"}), 404
    return jsonify(item)


@bp.route("/api/simulate-car", methods=["POST"])
@login_required
def api_create_simulate_car():
    """创建模拟车辆参数"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        car_id = create_simulate_car(**data)
        insert_log(
            current_user.id, current_user.username, "create",
            target="simulate_car",
            detail=f"创建模拟车辆: {data.get('simulate_num','')} {data.get('name','')}",
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True, "id": car_id})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/simulate-car/<int:car_id>", methods=["PUT"])
@login_required
def api_update_simulate_car(car_id):
    """更新模拟车辆参数"""
    data = request.get_json()
    if not data:
        return jsonify({"ok": False, "error": "数据为空"}), 400
    try:
        update_simulate_car(car_id, **data)
        insert_log(
            current_user.id, current_user.username, "update",
            target="simulate_car",
            detail=f"更新模拟车辆 id={car_id}: {data.get('simulate_num','')} {data.get('name','')}",
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500


@bp.route("/api/simulate-car/<int:car_id>", methods=["DELETE"])
@login_required
def api_delete_simulate_car(car_id):
    """删除模拟车辆参数"""
    try:
        item = get_simulate_car_by_id(car_id)
        detail = f"删除模拟车辆 id={car_id}"
        if item:
            detail += f": {item.get('simulate_num','')} {item.get('name','')}"
        delete_simulate_car(car_id)
        insert_log(
            current_user.id, current_user.username, "delete",
            target="simulate_car",
            detail=detail,
            result="ok", ip=request.remote_addr or "",
        )
        return jsonify({"ok": True})
    except Exception as e:
        return jsonify({"ok": False, "error": str(e)}), 500