feat(edc-web): 线圈参数/模拟车辆参数管理 + 工装关联 + 测试环境显示

新增功能： - 线圈参数管理页 (/coil-info): 增删改查，日志记录 - 模拟车辆管理页 (/simulate-car): 增删改查，日志记录 - 工装配置页新增线圈/模拟车辆选择区，保存时关联到 tb_fixture_param - 测试信息查询页新增「测试环境」列，显示当前线圈和模拟车辆信息 - edc_server 写入测试数据时自动从 fixture 获取线圈/车辆关联数据库： - 新增 tb_coil_info、tb_simulate_car 表 - tb_fixture_param 增加 coil_id/simulate_car_id 字段 - tb_state_tst 增加 coil_id/simulate_car_id 字段后端： - models.py 新增线圈/模拟车辆 CRUD - get_fixture_param 改为 LEFT JOIN 返回线圈/车辆详情 - upsert_fixture_param 支持 coil_id/simulate_car_id - 测试数据查询 LEFT JOIN 线圈/车辆信息
2026-06-08 10:42:13 +08:00
parent e538efafb5
commit 431653d033
10 changed files with 1085 additions and 25 deletions
--- a/edc-web/app/models.py
+++ b/edc-web/app/models.py
@@ -192,8 +192,13 @@ def get_test_data(page: int = 1, per_page: int = 20,
            # data
            offset = (page - 1) * per_page
            cur.execute(
-                f"SELECT t.*, d.serial FROM tb_state_tst t "
+                f"SELECT t.*, d.serial, "
+                f"c.coil_num, c.name as coil_name, "
+                f"sc.simulate_num, sc.name as car_name "
+                f"FROM tb_state_tst t "
                f"JOIN dnt_info d ON t.dnt_id = d.id "
+                f"LEFT JOIN tb_coil_info c ON t.coil_id = c.id "
+                f"LEFT JOIN tb_simulate_car sc ON t.simulate_car_id = sc.id "
                f"WHERE {where_clause} "
                f"ORDER BY t.id DESC LIMIT %s OFFSET %s",
                params + [per_page, offset],
@@ -236,8 +241,13 @@ def get_all_test_data_for_export(serial: str = "", date_from: str = "",

            where_clause = " AND ".join(where) if where else "1=1"
            cur.execute(
-                f"SELECT t.*, d.serial FROM tb_state_tst t "
+                f"SELECT t.*, d.serial, "
+                f"c.coil_num, c.name as coil_name, "
+                f"sc.simulate_num, sc.name as car_name "
+                f"FROM tb_state_tst t "
                f"JOIN dnt_info d ON t.dnt_id = d.id "
+                f"LEFT JOIN tb_coil_info c ON t.coil_id = c.id "
+                f"LEFT JOIN tb_simulate_car sc ON t.simulate_car_id = sc.id "
                f"WHERE {where_clause} ORDER BY t.id DESC",
                params,
            )
@@ -446,12 +456,24 @@ def get_logs(page: int = 1, per_page: int = 30,
 # ─── tb_fixture_param ──────────────────────────────────────────────

 def get_fixture_param(dnt_id: int) -> dict | None:
-    """获取设备的工装测试参数"""
+    """获取设备的工装测试参数（含线圈和模拟车辆信息）"""
    conn = get_conn()
    try:
        with conn.cursor() as cur:
            cur.execute(
-                "SELECT * FROM tb_fixture_param WHERE dnt_id=%s", (dnt_id,),
+                "SELECT fp.*, "
+                "c.coil_num, c.name as coil_name, c.shape as coil_shape, "
+                "c.length as coil_length, c.width as coil_width, c.radius as coil_radius, "
+                "c.turns as coil_turns, c.resistance as coil_resistance, "
+                "c.material as coil_material, "
+                "sc.simulate_num, sc.name as car_name, sc.shape as car_shape, "
+                "sc.length as car_length, sc.width as car_width, sc.radius as car_radius, "
+                "sc.material as car_material "
+                "FROM tb_fixture_param fp "
+                "LEFT JOIN tb_coil_info c ON fp.coil_id = c.id "
+                "LEFT JOIN tb_simulate_car sc ON fp.simulate_car_id = sc.id "
+                "WHERE fp.dnt_id=%s",
+                (dnt_id,),
            )
            return cur.fetchone()
    finally:
@@ -467,6 +489,7 @@ def upsert_fixture_param(dnt_id: int, **kwargs):
                "SELECT id FROM tb_fixture_param WHERE dnt_id=%s", (dnt_id,),
            )
            existing = cur.fetchone()
+            # 主线参数字段（不含 coil_id/simulate_car_id，后面单独处理）
            fields = [
                "Addr", "DevType", "TestMode", "RestDis", "MinusDis",
                "SensMin", "SensMax", "FreMin", "FreMax", "PeakMin", "PeakMax",
@@ -488,6 +511,17 @@ def upsert_fixture_param(dnt_id: int, **kwargs):
                    f"VALUES (%s, {placeholders})",
                    [dnt_id] + values,
                )
+            # 单独处理线圈/模拟车辆关联（可选，不覆盖已有值）
+            if "coil_id" in kwargs:
+                cur.execute(
+                    "UPDATE tb_fixture_param SET coil_id=%s WHERE dnt_id=%s",
+                    (kwargs["coil_id"], dnt_id),
+                )
+            if "simulate_car_id" in kwargs:
+                cur.execute(
+                    "UPDATE tb_fixture_param SET simulate_car_id=%s WHERE dnt_id=%s",
+                    (kwargs["simulate_car_id"], dnt_id),
+                )
        conn.commit()
    finally:
        conn.close()
@@ -584,6 +618,172 @@ def delete_vehicle_base_test(test_id: int):
        conn.close()


+# ─── 线圈参数 CRUD ──────────────────────────────────────────────────
+
+def get_coil_info_list(search: str = "") -> list[dict]:
+    """获取线圈参数列表"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            if search:
+                cur.execute(
+                    "SELECT * FROM tb_coil_info WHERE coil_num LIKE %s OR name LIKE %s "
+                    "ORDER BY id DESC",
+                    (f"%{search}%", f"%{search}%"),
+                )
+            else:
+                cur.execute("SELECT * FROM tb_coil_info ORDER BY id DESC")
+            return cur.fetchall()
+    finally:
+        conn.close()
+
+
+def get_coil_info_by_id(coil_id: int) -> dict | None:
+    """获取单个线圈参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT * FROM tb_coil_info WHERE id=%s", (coil_id,))
+            return cur.fetchone()
+    finally:
+        conn.close()
+
+
+def create_coil_info(**kwargs) -> int:
+    """创建线圈参数，返回新 ID"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            fields = [
+                "coil_num", "name", "induct", "shape", "length", "width",
+                "radius", "turns", "resistance", "material", "remark",
+            ]
+            col_names = ", ".join(f"`{f}`" for f in fields)
+            placeholders = ", ".join(["%s"] * len(fields))
+            values = [kwargs.get(f, "" if f in ("coil_num", "name", "shape", "material", "remark") else 0) for f in fields]
+            cur.execute(
+                f"INSERT INTO tb_coil_info ({col_names}) VALUES ({placeholders})",
+                values,
+            )
+        conn.commit()
+        return cur.lastrowid
+    finally:
+        conn.close()
+
+
+def update_coil_info(coil_id: int, **kwargs):
+    """更新线圈参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            fields = [
+                "coil_num", "name", "induct", "shape", "length", "width",
+                "radius", "turns", "resistance", "material", "remark",
+            ]
+            sets = ", ".join(f"`{f}`=%s" for f in fields)
+            values = [kwargs.get(f, "" if f in ("coil_num", "name", "shape", "material", "remark") else 0) for f in fields] + [coil_id]
+            cur.execute(
+                f"UPDATE tb_coil_info SET {sets} WHERE id=%s", values,
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def delete_coil_info(coil_id: int):
+    """删除线圈参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("DELETE FROM tb_coil_info WHERE id=%s", (coil_id,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+# ─── 模拟车辆参数 CRUD ──────────────────────────────────────────────
+
+def get_simulate_car_list(search: str = "") -> list[dict]:
+    """获取模拟车辆参数列表"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            if search:
+                cur.execute(
+                    "SELECT * FROM tb_simulate_car WHERE simulate_num LIKE %s OR name LIKE %s "
+                    "ORDER BY id DESC",
+                    (f"%{search}%", f"%{search}%"),
+                )
+            else:
+                cur.execute("SELECT * FROM tb_simulate_car ORDER BY id DESC")
+            return cur.fetchall()
+    finally:
+        conn.close()
+
+
+def get_simulate_car_by_id(car_id: int) -> dict | None:
+    """获取单个模拟车辆参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT * FROM tb_simulate_car WHERE id=%s", (car_id,))
+            return cur.fetchone()
+    finally:
+        conn.close()
+
+
+def create_simulate_car(**kwargs) -> int:
+    """创建模拟车辆参数，返回新 ID"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            fields = [
+                "simulate_num", "name", "shape", "length", "width",
+                "radius", "material", "remark",
+            ]
+            col_names = ", ".join(f"`{f}`" for f in fields)
+            placeholders = ", ".join(["%s"] * len(fields))
+            values = [kwargs.get(f, "" if f in ("simulate_num", "name", "shape", "material", "remark") else 0) for f in fields]
+            cur.execute(
+                f"INSERT INTO tb_simulate_car ({col_names}) VALUES ({placeholders})",
+                values,
+            )
+        conn.commit()
+        return cur.lastrowid
+    finally:
+        conn.close()
+
+
+def update_simulate_car(car_id: int, **kwargs):
+    """更新模拟车辆参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            fields = [
+                "simulate_num", "name", "shape", "length", "width",
+                "radius", "material", "remark",
+            ]
+            sets = ", ".join(f"`{f}`=%s" for f in fields)
+            values = [kwargs.get(f, "" if f in ("simulate_num", "name", "shape", "material", "remark") else 0) for f in fields] + [car_id]
+            cur.execute(
+                f"UPDATE tb_simulate_car SET {sets} WHERE id=%s", values,
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def delete_simulate_car(car_id: int):
+    """删除模拟车辆参数"""
+    conn = get_conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("DELETE FROM tb_simulate_car WHERE id=%s", (car_id,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
 # ─── 测试数据删除 ──────────────────────────────────────────────

 def delete_test_data(serial: str = "", date_from: str = "",