feat: 工装配置页面 + 车检器基准参数管理

- 设备页增加「配置」按钮(devices.js)
- 新增工装配置页面(fixture.html+js): 参数表单、5个操作按钮、通信日志区、基准参数表
- 新增车检器基准参数管理页面(vehicle_base_test.html+js): CRUD + 搜索
- 新增 fixture 蓝图(routes/fixture.py): 0x4A~0x4E 指令发送、参数CRUD、serialnet状态查询
- models.py: 新增 get_serialnet_by_id, tb_fixture_param/tb_vechicle_base_test CRUD
- edc_server 子模块更新
This commit is contained in:
wangfq
2026-05-29 17:26:07 +08:00
parent ae816eaffd
commit dbe5d1cefb
10 changed files with 1197 additions and 1 deletions

View File

@@ -108,6 +108,19 @@ def get_serialnet_records(dnt_id: int, limit: int = 50) -> list[dict]:
conn.close()
def get_serialnet_by_id(record_id: int) -> dict | None:
"""根据 ID 获取 tb_serialnet 记录"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT * FROM tb_serialnet WHERE id=%s", (record_id,),
)
return cur.fetchone()
finally:
conn.close()
def clear_serialnet_records(dnt_id: int):
"""清除指定设备的所有透传记录"""
conn = get_conn()
@@ -375,3 +388,143 @@ def get_logs(page: int = 1, per_page: int = 30,
return cur.fetchall(), total
finally:
conn.close()
# ─── 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,),
)
return cur.fetchone()
finally:
conn.close()
def upsert_fixture_param(dnt_id: int, **kwargs):
"""插入或更新工装测试参数"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT id FROM tb_fixture_param WHERE dnt_id=%s", (dnt_id,),
)
existing = cur.fetchone()
fields = [
"Addr", "DevType", "TestMode", "RestDis", "MinusDis",
"SensMin", "SensMax", "FreMin", "FreMax", "PeakMin", "PeakMax",
]
if existing:
sets = ", ".join(f"`{f}`=%s" for f in fields)
values = [kwargs.get(f, 0) for f in fields] + [dnt_id]
cur.execute(
f"UPDATE tb_fixture_param SET {sets} WHERE dnt_id=%s",
values,
)
else:
placeholders = ", ".join(["%s"] * len(fields))
col_names = ", ".join(f"`{f}`" for f in fields)
values = [kwargs.get(f, 0) for f in fields]
cur.execute(
f"INSERT INTO tb_fixture_param (dnt_id, {col_names}) "
f"VALUES (%s, {placeholders})",
[dnt_id] + values,
)
conn.commit()
finally:
conn.close()
# ─── tb_vechicle_base_test ─────────────────────────────────────────
def get_vehicle_base_tests(search: str = "") -> list[dict]:
"""获取车检器测试基准参数列表"""
conn = get_conn()
try:
with conn.cursor() as cur:
if search:
cur.execute(
"SELECT * FROM tb_vechicle_base_test "
"WHERE dev_name LIKE %s OR type_num LIKE %s "
"ORDER BY type_num ASC",
(f"%{search}%", f"%{search}%"),
)
else:
cur.execute(
"SELECT * FROM tb_vechicle_base_test ORDER BY type_num ASC",
)
return cur.fetchall()
finally:
conn.close()
def get_vehicle_base_test_by_id(test_id: int) -> dict | None:
"""根据 ID 获取车检器测试基准"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute(
"SELECT * FROM tb_vechicle_base_test WHERE id=%s", (test_id,),
)
return cur.fetchone()
finally:
conn.close()
def create_vehicle_base_test(**kwargs) -> int:
"""创建车检器测试基准,返回新记录 ID"""
conn = get_conn()
try:
with conn.cursor() as cur:
fields = [
"dev_name", "type_num", "SensMin", "SensMax",
"FreMin", "FreMax", "PeakMin", "PeakMax", "remark",
]
col_names = ", ".join(f"`{f}`" for f in fields)
placeholders = ", ".join(["%s"] * len(fields))
values = [kwargs.get(f, "" if f in ("dev_name", "remark") else 0) for f in fields]
cur.execute(
f"INSERT INTO tb_vechicle_base_test ({col_names}) VALUES ({placeholders})",
values,
)
conn.commit()
return cur.lastrowid
finally:
conn.close()
def update_vehicle_base_test(test_id: int, **kwargs):
"""更新车检器测试基准"""
conn = get_conn()
try:
with conn.cursor() as cur:
fields = [
"dev_name", "type_num", "SensMin", "SensMax",
"FreMin", "FreMax", "PeakMin", "PeakMax", "remark",
]
sets = ", ".join(f"`{f}`=%s" for f in fields)
values = [kwargs.get(f, "" if f in ("dev_name", "remark") else 0) for f in fields] + [test_id]
cur.execute(
f"UPDATE tb_vechicle_base_test SET {sets} WHERE id=%s",
values,
)
conn.commit()
finally:
conn.close()
def delete_vehicle_base_test(test_id: int):
"""删除车检器测试基准"""
conn = get_conn()
try:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM tb_vechicle_base_test WHERE id=%s", (test_id,),
)
conn.commit()
finally:
conn.close()