fixL s3_analyzer noise clean up.
-_extract_a4_inner_frames(payload) — splits the A4 container payload into inner sub-frames using the ACK DLE STX delimiter pattern, returning (sub, page_key, data) tuples -_diff_a4_payloads(payload_a, payload_b) — matches inner frames by (sub, page_key), diffs data byte-by-byte (with existing noise masking), and reports added/removed inner frames as synthetic entries
This commit is contained in:
@@ -455,6 +455,140 @@ def lookup_field_name(sub: int, page_key: int, payload_offset: int) -> Optional[
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _extract_a4_inner_frames(payload: bytes) -> list[tuple[int, int, bytes]]:
|
||||||
|
"""
|
||||||
|
Parse the inner sub-frame stream packed inside an A4 (POLL_RESPONSE) payload.
|
||||||
|
|
||||||
|
The payload is a sequence of inner frames, each starting with DLE STX (10 02)
|
||||||
|
and delimited by ACK (41) before the next DLE STX. The inner frame body
|
||||||
|
(after the 10 02 preamble) has the same 5-byte header layout as outer frames:
|
||||||
|
[0] 00
|
||||||
|
[1] 10
|
||||||
|
[2] SUB
|
||||||
|
[3] OFFSET_HI (page_key high byte)
|
||||||
|
[4] OFFSET_LO (page_key low byte)
|
||||||
|
[5+] data
|
||||||
|
|
||||||
|
Returns a list of (sub, page_key, data_bytes) — one entry per inner frame,
|
||||||
|
keeping ALL occurrences (not deduped), so the caller can decide how to match.
|
||||||
|
"""
|
||||||
|
DLE, STX, ACK = 0x10, 0x02, 0x41
|
||||||
|
results: list[tuple[int, int, bytes]] = []
|
||||||
|
|
||||||
|
# Collect start positions of each inner frame (offset of the DLE STX)
|
||||||
|
starts: list[int] = []
|
||||||
|
i = 0
|
||||||
|
# First frame may begin at offset 0 with DLE STX directly
|
||||||
|
if len(payload) >= 2 and payload[0] == DLE and payload[1] == STX:
|
||||||
|
starts.append(0)
|
||||||
|
i = 2
|
||||||
|
while i < len(payload) - 2:
|
||||||
|
if payload[i] == ACK and payload[i + 1] == DLE and payload[i + 2] == STX:
|
||||||
|
starts.append(i + 1) # point at the DLE
|
||||||
|
i += 3
|
||||||
|
else:
|
||||||
|
i += 1
|
||||||
|
|
||||||
|
for k, s in enumerate(starts):
|
||||||
|
# Body starts after DLE STX (2 bytes)
|
||||||
|
body_start = s + 2
|
||||||
|
body_end = starts[k + 1] - 1 if k + 1 < len(starts) else len(payload)
|
||||||
|
body = payload[body_start:body_end]
|
||||||
|
if len(body) < 5:
|
||||||
|
continue
|
||||||
|
# body[0]=0x00, body[1]=0x10, body[2]=SUB, body[3]=OFFSET_HI, body[4]=OFFSET_LO
|
||||||
|
sub = body[2]
|
||||||
|
page_key = (body[3] << 8) | body[4]
|
||||||
|
data = body[5:]
|
||||||
|
results.append((sub, page_key, data))
|
||||||
|
|
||||||
|
return results
|
||||||
|
|
||||||
|
|
||||||
|
def _diff_a4_payloads(payload_a: bytes, payload_b: bytes) -> list[ByteDiff]:
|
||||||
|
"""
|
||||||
|
Diff two A4 container payloads at the inner sub-frame level.
|
||||||
|
|
||||||
|
Inner frames are matched by (sub, page_key). For each pair of matching
|
||||||
|
inner frames whose data differs, the changed bytes are reported with
|
||||||
|
payload_offset encoded as: (inner_frame_index << 16) | byte_offset_in_data.
|
||||||
|
|
||||||
|
Inner frames present in one payload but not the other are reported as a
|
||||||
|
single synthetic ByteDiff entry with before/after = -1 / -2 respectively,
|
||||||
|
and field_name describing the missing inner SUB.
|
||||||
|
|
||||||
|
The high-16 / low-16 split in payload_offset lets the GUI render these
|
||||||
|
differently if desired, but they degrade gracefully in the existing renderer.
|
||||||
|
"""
|
||||||
|
frames_a = _extract_a4_inner_frames(payload_a)
|
||||||
|
frames_b = _extract_a4_inner_frames(payload_b)
|
||||||
|
|
||||||
|
# Build multimap: (sub, page_key) → list of data blobs, preserving order
|
||||||
|
def index(frames):
|
||||||
|
idx: dict[tuple[int, int], list[bytes]] = {}
|
||||||
|
for sub, pk, data in frames:
|
||||||
|
idx.setdefault((sub, pk), []).append(data)
|
||||||
|
return idx
|
||||||
|
|
||||||
|
idx_a = index(frames_a)
|
||||||
|
idx_b = index(frames_b)
|
||||||
|
|
||||||
|
all_keys = sorted(set(idx_a) | set(idx_b))
|
||||||
|
diffs: list[ByteDiff] = []
|
||||||
|
|
||||||
|
for sub, pk in all_keys:
|
||||||
|
list_a = idx_a.get((sub, pk), [])
|
||||||
|
list_b = idx_b.get((sub, pk), [])
|
||||||
|
|
||||||
|
# Pair up by position; extras are treated as added/removed
|
||||||
|
n = max(len(list_a), len(list_b))
|
||||||
|
for pos in range(n):
|
||||||
|
da = list_a[pos] if pos < len(list_a) else None
|
||||||
|
db = list_b[pos] if pos < len(list_b) else None
|
||||||
|
|
||||||
|
if da is None:
|
||||||
|
# Inner frame added in B
|
||||||
|
entry = SUB_TABLE.get(sub)
|
||||||
|
name = entry[0] if entry else f"UNKNOWN_{sub:02X}"
|
||||||
|
diffs.append(ByteDiff(
|
||||||
|
payload_offset=(sub << 16) | (pk & 0xFFFF),
|
||||||
|
before=-1,
|
||||||
|
after=-2,
|
||||||
|
field_name=f"[A4 inner] SUB {sub:02X} ({name}) pk={pk:04X} added",
|
||||||
|
))
|
||||||
|
continue
|
||||||
|
if db is None:
|
||||||
|
entry = SUB_TABLE.get(sub)
|
||||||
|
name = entry[0] if entry else f"UNKNOWN_{sub:02X}"
|
||||||
|
diffs.append(ByteDiff(
|
||||||
|
payload_offset=(sub << 16) | (pk & 0xFFFF),
|
||||||
|
before=-2,
|
||||||
|
after=-1,
|
||||||
|
field_name=f"[A4 inner] SUB {sub:02X} ({name}) pk={pk:04X} removed",
|
||||||
|
))
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Both present — byte diff the data sections
|
||||||
|
da_m = _mask_noisy(sub, da)
|
||||||
|
db_m = _mask_noisy(sub, db)
|
||||||
|
if da_m == db_m:
|
||||||
|
continue
|
||||||
|
max_len = max(len(da_m), len(db_m))
|
||||||
|
for off in range(max_len):
|
||||||
|
ba = da_m[off] if off < len(da_m) else None
|
||||||
|
bb = db_m[off] if off < len(db_m) else None
|
||||||
|
if ba != bb:
|
||||||
|
field = lookup_field_name(sub, pk, off + HEADER_LEN)
|
||||||
|
diffs.append(ByteDiff(
|
||||||
|
payload_offset=(sub << 16) | (off & 0xFFFF),
|
||||||
|
before=ba if ba is not None else -1,
|
||||||
|
after=bb if bb is not None else -1,
|
||||||
|
field_name=field or f"[A4:{sub:02X} pk={pk:04X}] off={off}",
|
||||||
|
))
|
||||||
|
|
||||||
|
return diffs
|
||||||
|
|
||||||
|
|
||||||
def diff_sessions(sess_a: Session, sess_b: Session) -> list[FrameDiff]:
|
def diff_sessions(sess_a: Session, sess_b: Session) -> list[FrameDiff]:
|
||||||
"""
|
"""
|
||||||
Compare two sessions frame-by-frame, matched by (sub, page_key).
|
Compare two sessions frame-by-frame, matched by (sub, page_key).
|
||||||
@@ -484,6 +618,16 @@ def diff_sessions(sess_a: Session, sess_b: Session) -> list[FrameDiff]:
|
|||||||
af_a = idx_a[key]
|
af_a = idx_a[key]
|
||||||
af_b = idx_b[key]
|
af_b = idx_b[key]
|
||||||
|
|
||||||
|
# A4 is a container frame — diff at the inner sub-frame level to avoid
|
||||||
|
# phase-shift noise when the number of embedded records differs.
|
||||||
|
if sub == 0xA4:
|
||||||
|
diffs = _diff_a4_payloads(af_a.frame.payload, af_b.frame.payload)
|
||||||
|
if diffs:
|
||||||
|
entry = SUB_TABLE.get(sub)
|
||||||
|
sub_name = entry[0] if entry else f"UNKNOWN_{sub:02X}"
|
||||||
|
results.append(FrameDiff(sub=sub, page_key=page_key, sub_name=sub_name, diffs=diffs))
|
||||||
|
continue
|
||||||
|
|
||||||
data_a = _mask_noisy(sub, _get_data_section(af_a))
|
data_a = _mask_noisy(sub, _get_data_section(af_a))
|
||||||
data_b = _mask_noisy(sub, _get_data_section(af_b))
|
data_b = _mask_noisy(sub, _get_data_section(af_b))
|
||||||
|
|
||||||
|
|||||||
@@ -825,8 +825,17 @@ class AnalyzerPanel(tk.Frame):
|
|||||||
w.tag_bind(link_tag, "<Leave>", lambda e: w.configure(cursor=""))
|
w.tag_bind(link_tag, "<Leave>", lambda e: w.configure(cursor=""))
|
||||||
w.configure(state="disabled")
|
w.configure(state="disabled")
|
||||||
for bd in fd.diffs:
|
for bd in fd.diffs:
|
||||||
b = f"{bd.before:02x}" if bd.before >= 0 else "--"
|
def _fmt(v: int) -> str:
|
||||||
a = f"{bd.after:02x}" if bd.after >= 0 else "--"
|
if v == -2: return "ADD"
|
||||||
|
if v == -1: return "--"
|
||||||
|
return f"{v:02x}"
|
||||||
|
b, a = _fmt(bd.before), _fmt(bd.after)
|
||||||
|
# A4 inner-frame add/remove: field_name carries the full description
|
||||||
|
if bd.field_name and (bd.before == -2 or bd.after == -2 or bd.before == -1 or bd.after == -1) and bd.field_name.startswith("[A4"):
|
||||||
|
self._tw(w, f" {b} -> {a} ", "changed")
|
||||||
|
self._tw(w, bd.field_name, "known")
|
||||||
|
self._tn(w)
|
||||||
|
else:
|
||||||
self._tw(w, f" [{bd.payload_offset:3d}] 0x{bd.payload_offset:04X}: ", "dim")
|
self._tw(w, f" [{bd.payload_offset:3d}] 0x{bd.payload_offset:04X}: ", "dim")
|
||||||
self._tw(w, f"{b} -> {a}", "changed")
|
self._tw(w, f"{b} -> {a}", "changed")
|
||||||
if bd.field_name: self._tw(w, f" [{bd.field_name}]", "known")
|
if bd.field_name: self._tw(w, f" [{bd.field_name}]", "known")
|
||||||
|
|||||||
Reference in New Issue
Block a user