wangfq/edc_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: DG430 V2.0.3 — 波动测试模式支持

Browse Source 
- dg430.py: 新增 DG430WaveStatus + parse_b4_wave_status() 0xB4解析
- dg430.py: 0x4C 扩展6字段(向后兼容旧版长度)
- models.py: tb_fixture_param DDL + upsert 新增6个波动参数
- handlers.py: parse_loop 添加0xB4处理; 0x4C传参扩展
- TestMode=1 模拟过车→波动测试 (注释)
This commit is contained in:
wangfq
2026-06-02 18:06:07 +08:00
parent 6ecc653133
commit 2d6c9f03dd
3 changed files with 129 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/dg430.py
View File

@@ -19,7 +19,7 @@ class DG430Status:
"""解析后的 DG430 状态数据"""
addr: int # 地址
dev_model: int # 1 PD132, 2 DLD110
test_mode: int # 0 灵敏度, 1 模拟过车
test_mode: int # 0 灵敏度, 1 波动测试
is_finished: bool # 是否正常完成
finish_code: int # 0 正常, 1 未完成, 2 地感死机
fault: int # bitmask
@@ -150,7 +150,7 @@ def parse_b2_status(data: bytes) -> DG430Status | None:
# 设备型号
dev_model = payload[0] # 1=PD132, 2=DLD110
test_mode = payload[1] # 0=灵敏度, 1=模拟过车
test_mode = payload[1] # 0=灵敏度, 1=波动测试
finish_code = payload[2] # 0=正常, 1=未完成, 2=死机
fault = payload[3] # bitmask
relay_out = payload[4] # bitmask
@@ -191,6 +191,70 @@ def parse_b2_status(data: bytes) -> DG430Status | None:
)
# ─── 0xB4 波动测试上报 ──────────────────────────────────────────
@dataclass
class DG430WaveStatus:
"""波动测试状态上报数据"""
addr: int # 地址
remain_count: int # 剩余波动次数
relay_out: int # 继电器输出 bitmask
work_freq: float # 工作频率 Hz
curr_dist: int # 当前距离 mm
speed: int # 当前速度 dm/s
near_dist: int # 波动最近距离 mm
far_dist: int # 波动最远距离 mm
enter_dist: int # 进入高度 mm
leave_dist: int # 离开高度 mm
def parse_b4_wave_status(data: bytes) -> DG430WaveStatus | None:
"""解析 0xB4 波动测试状态上报包
格式: STX | ADDR | 11 | B4 | DATA(16B) | XOR | SUM
DATA: RemainCount(1) | Relay(1) | WorkFreq(2 LE) | CurrDist(2 LE) |
Speed(2 LE) | NearDist(2 LE) | FarDist(2 LE) |
EnterDist(2 LE) | LeaveDist(2 LE)
"""
if not verify_packet(data):
logger.warning("DG430 B4 数据包校验失败")
return None
cmd = data[3]
if cmd != 0xB4:
logger.debug(f"非 B4 指令: 0x{cmd:02X}")
return None
payload = data[4:3 + data[2]]
if len(payload) < 16:
logger.warning(f"B4 数据长度不足: {len(payload)} < 16")
return None
addr = data[1] & 0x7F
remain_count = payload[0]
relay_out = payload[1]
work_freq = _le16(payload, 2) * 10.0
curr_dist = _le16(payload, 4)
speed = _le16(payload, 6)
near_dist = _le16(payload, 8)
far_dist = _le16(payload, 10)
enter_dist = _le16(payload, 12)
leave_dist = _le16(payload, 14)
return DG430WaveStatus(
addr=addr,
remain_count=remain_count,
relay_out=relay_out,
work_freq=round(work_freq, 1),
curr_dist=curr_dist,
speed=speed,
near_dist=near_dist,
far_dist=far_dist,
enter_dist=enter_dist,
leave_dist=leave_dist,
)
# ─── 故障解码 ───────────────────────────────────────────────────────
FAULT_BITS = {
@@ -290,7 +354,7 @@ class DG430FixtureParams:
flag: int # 0=正常, 1=故障
dev_addr: int # 设备地址
dev_type: int # 设备型号
test_mode: int # 0 灵敏度, 1 模拟过车
test_mode: int # 0 灵敏度, 1 波动测试
reset_dis: int # 复位距离 cm
minus_dis: int # 皮距 cm
sens_min: int # 灵敏度最小值
@@ -299,14 +363,22 @@ class DG430FixtureParams:
fre_max: int # 频率最大值 Hz
peak_min: int # 峰峰值最小值
peak_max: int # 峰峰值最大值
far_tol: int # 最远容差 cm
near_tol: int # 最近容差 cm
step_tol: int # 步进容差 cm
back_forth: int # 来回次数
near_stay: int # 最近停留时间 ms
far_stay: int # 最远停留时间 ms
def parse_4c_params(data: bytes) -> DG430FixtureParams | None:
"""解析 0x4C 查询测试参数响应
"""解析 0x4C 查询测试参数响应 (V2.0.3 扩展)
格式: 7F | ADDR | 13 | 4C | Flag | Addr | DevType | TestMode |
格式: 7F | ADDR | 1B | 4C | Flag | Addr | DevType | TestMode |
ResetDis | MinusDis | SensMin(2) | SensMax(2) |
FreMin(2) | FreMax(2) | PeakMin(2) | PeakMax(2) | XOR | SUM
FreMin(2) | FreMax(2) | PeakMin(2) | PeakMax(2) |
FarTol(1) | NearTol(1) | StepTol(1) | BackForth(1) |
NearStay(2) | FarStay(2) | XOR | SUM
"""
if not verify_packet(data):
return None
@@ -320,6 +392,14 @@ def parse_4c_params(data: bytes) -> DG430FixtureParams | None:
addr = data[1] & 0x7F
# 0x4B/0x4C 多字节字段为小端序
# V2.0.3 新增6个波动参数字段，兼容旧版（长度不足时默认为0）
far_tol = payload[18] if len(payload) >= 19 else 0
near_tol = payload[19] if len(payload) >= 20 else 0
step_tol = payload[20] if len(payload) >= 21 else 0
back_forth = payload[21] if len(payload) >= 22 else 0
near_stay = _le16(payload, 22) if len(payload) >= 24 else 0
far_stay = _le16(payload, 24) if len(payload) >= 26 else 0
return DG430FixtureParams(
addr=addr,
flag=payload[0],
@@ -334,6 +414,12 @@ def parse_4c_params(data: bytes) -> DG430FixtureParams | None:
fre_max=_le16(payload, 12),
peak_min=_le16(payload, 14),
peak_max=_le16(payload, 16),
far_tol=far_tol,
near_tol=near_tol,
step_tol=step_tol,
back_forth=back_forth,
near_stay=near_stay,
far_stay=far_stay,
)

29
src/handlers.py
View File

@@ -34,6 +34,7 @@ from src.dg430 import (
parse_4a_version,
parse_flag_response,
parse_4c_params,
parse_b4_wave_status,
get_packet_cmd,
hex_str_to_bytes,
split_packets,
@@ -322,6 +323,12 @@ async def parse_loop():
FreMax=params.fre_max,
PeakMin=params.peak_min,
PeakMax=params.peak_max,
FarTol=params.far_tol,
NearTol=params.near_tol,
StepTol=params.step_tol,
BackForth=params.back_forth,
NearStay=params.near_stay,
FarStay=params.far_stay,
)
logger.info(
f"0x4C 工装参数已更新 dnt_id={dnt_id} "
@@ -332,6 +339,28 @@ async def parse_loop():
has_valid = True
all_failed = False
# ── 0xB4 波动测试上报 ──
elif cmd == 0xB4:
if not verify_packet(pkt):
logger.debug(f"0xB4 数据包校验失败: {device_id}")
continue
wave = parse_b4_wave_status(pkt)
if wave is None:
logger.warning(f"0xB4 解析失败: {device_id}")
continue
relay_info = decode_relay_info(wave.relay_out)
logger.info(
f"B4波动上报: {device_id} 剩余={wave.remain_count} "
f"当前距离={wave.curr_dist}mm 速度={wave.speed}dm/s "
f"最近={wave.near_dist}mm 最远={wave.far_dist}mm "
f"进入={wave.enter_dist}mm 离开={wave.leave_dist}mm "
f"继电器={relay_info}"
)
has_valid = True
all_failed = False
# ── 测试指令 Flag 响应（不匹配 serialnet，等待 B2 完成）──
elif cmd in (0xB0, 0xB1, 0xBA, 0xBB, 0xBC):
if not verify_packet(pkt):

9
src/models.py
View File

@@ -154,7 +154,7 @@ async def _create_tables(pool: aiomysql.Pool):
`dnt_id` INT NOT NULL COMMENT 'FK → dnt_info.id',
`Addr` TINYINT DEFAULT 1 COMMENT '工装设备地址',
`DevType` TINYINT DEFAULT 0 COMMENT '被检设备型号类型编码',
`TestMode` TINYINT DEFAULT 0 COMMENT '0 灵敏度测试, 1 模拟过车',
`TestMode` TINYINT DEFAULT 0 COMMENT '0 灵敏度测试, 1 波动测试',
`RestDis` INT DEFAULT 0 COMMENT '复位距离 cm',
`MinusDis` INT DEFAULT 0 COMMENT '皮距/开始距离 cm',
`SensMin` INT DEFAULT 0 COMMENT '灵敏度最小值',
@@ -163,6 +163,12 @@ async def _create_tables(pool: aiomysql.Pool):
`FreMax` INT DEFAULT 0 COMMENT '频率最大值 Hz',
`PeakMin` INT DEFAULT 0 COMMENT '峰峰值最小值',
`PeakMax` INT DEFAULT 0 COMMENT '峰峰值最大值',
`FarTol` INT DEFAULT 0 COMMENT '最远容差 cm (V2.0.3)',
`NearTol` INT DEFAULT 0 COMMENT '最近容差 cm (V2.0.3)',
`StepTol` INT DEFAULT 0 COMMENT '步进容差 cm (V2.0.3)',
`BackForth` INT DEFAULT 0 COMMENT '来回次数 (V2.0.3)',
`NearStay` INT DEFAULT 0 COMMENT '最近停留时间 ms (V2.0.3)',
`FarStay` INT DEFAULT 0 COMMENT '最远停留时间 ms (V2.0.3)',
`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP,
`update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE INDEX `idx_dnt_id` (`dnt_id`)
@@ -479,6 +485,7 @@ async def upsert_fixture_param(dnt_id: int, **kwargs):
fields = [
"Addr", "DevType", "TestMode", "RestDis", "MinusDis",
"SensMin", "SensMax", "FreMin", "FreMax", "PeakMin", "PeakMax",
"FarTol", "NearTol", "StepTol", "BackForth", "NearStay", "FarStay",
]
if existing:
sets = ", ".join(f"`{f}`=%s" for f in fields)