154a11d057
- Implement functionality to read and parse raw_s3.bin and raw_bw.bin files. - Define protocol constants and mappings for various command and response identifiers. - Create data structures for frames, sessions, and diffs to facilitate analysis. - Develop functions for annotating frames, splitting sessions, and generating reports. - Include live mode for continuous monitoring and reporting of protocol frames. - Add command-line interface for user interaction and configuration.
941 lines
42 KiB
Python
941 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
gui_analyzer.py — Tkinter GUI for s3_analyzer.
|
|
|
|
Layout:
|
|
┌─────────────────────────────────────────────────────────┐
|
|
│ [S3 file: ___________ Browse] [BW file: ___ Browse] │
|
|
│ [Analyze] [Live mode toggle] Status: Idle │
|
|
├──────────────────┬──────────────────────────────────────┤
|
|
│ Session list │ Detail panel (tabs) │
|
|
│ ─ Session 0 │ Inventory | Hex Dump | Diff │
|
|
│ └ POLL (BW) │ │
|
|
│ └ POLL_RESP │ (content of selected tab) │
|
|
│ ─ Session 1 │ │
|
|
│ └ ... │ │
|
|
└──────────────────┴──────────────────────────────────────┘
|
|
│ Status bar │
|
|
└─────────────────────────────────────────────────────────┘
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import queue
|
|
import sys
|
|
import threading
|
|
import time
|
|
import tkinter as tk
|
|
from pathlib import Path
|
|
from tkinter import filedialog, font, messagebox, ttk
|
|
from typing import Optional
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from s3_analyzer import ( # noqa: E402
|
|
AnnotatedFrame,
|
|
FrameDiff,
|
|
Session,
|
|
annotate_frames,
|
|
diff_sessions,
|
|
format_hex_dump,
|
|
parse_bw,
|
|
parse_s3,
|
|
render_session_report,
|
|
split_into_sessions,
|
|
write_claude_export,
|
|
)
|
|
from frame_db import FrameDB, DEFAULT_DB_PATH # noqa: E402
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Colour palette (dark-ish terminal feel)
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
BG = "#1e1e1e"
|
|
BG2 = "#252526"
|
|
BG3 = "#2d2d30"
|
|
FG = "#d4d4d4"
|
|
FG_DIM = "#6a6a6a"
|
|
ACCENT = "#569cd6"
|
|
ACCENT2 = "#4ec9b0"
|
|
RED = "#f44747"
|
|
YELLOW = "#dcdcaa"
|
|
GREEN = "#4caf50"
|
|
ORANGE = "#ce9178"
|
|
|
|
COL_BW = "#9cdcfe" # BW frames
|
|
COL_S3 = "#4ec9b0" # S3 frames
|
|
COL_DIFF = "#f44747" # Changed bytes
|
|
COL_KNOW = "#4caf50" # Known-field annotations
|
|
COL_HEAD = "#569cd6" # Section headers
|
|
|
|
MONO = ("Consolas", 9)
|
|
MONO_SM = ("Consolas", 8)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# State container
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
class AnalyzerState:
|
|
def __init__(self) -> None:
|
|
self.sessions: list[Session] = []
|
|
self.diffs: list[Optional[list[FrameDiff]]] = [] # diffs[i] = diff of session i vs i-1
|
|
self.s3_path: Optional[Path] = None
|
|
self.bw_path: Optional[Path] = None
|
|
self.last_capture_id: Optional[int] = None
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Main GUI
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
class AnalyzerGUI(tk.Tk):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.title("S3 Protocol Analyzer")
|
|
self.configure(bg=BG)
|
|
self.minsize(1050, 600)
|
|
|
|
self.state = AnalyzerState()
|
|
self._live_thread: Optional[threading.Thread] = None
|
|
self._live_stop = threading.Event()
|
|
self._live_q: queue.Queue[str] = queue.Queue()
|
|
self._db = FrameDB()
|
|
|
|
self._build_widgets()
|
|
self._poll_live_queue()
|
|
|
|
# ── widget construction ────────────────────────────────────────────────
|
|
|
|
def _build_widgets(self) -> None:
|
|
self._build_toolbar()
|
|
self._build_panes()
|
|
self._build_statusbar()
|
|
|
|
def _build_toolbar(self) -> None:
|
|
bar = tk.Frame(self, bg=BG2, pady=4)
|
|
bar.pack(side=tk.TOP, fill=tk.X)
|
|
|
|
pad = {"padx": 5, "pady": 2}
|
|
|
|
# S3 file
|
|
tk.Label(bar, text="S3 raw:", bg=BG2, fg=FG, font=MONO).pack(side=tk.LEFT, **pad)
|
|
self.s3_var = tk.StringVar()
|
|
tk.Entry(bar, textvariable=self.s3_var, width=28, bg=BG3, fg=FG,
|
|
insertbackground=FG, relief="flat", font=MONO).pack(side=tk.LEFT, **pad)
|
|
tk.Button(bar, text="Browse", bg=BG3, fg=FG, relief="flat",
|
|
activebackground=ACCENT, cursor="hand2",
|
|
command=lambda: self._browse_file(self.s3_var, "raw_s3.bin")
|
|
).pack(side=tk.LEFT, **pad)
|
|
|
|
tk.Label(bar, text=" BW raw:", bg=BG2, fg=FG, font=MONO).pack(side=tk.LEFT, **pad)
|
|
self.bw_var = tk.StringVar()
|
|
tk.Entry(bar, textvariable=self.bw_var, width=28, bg=BG3, fg=FG,
|
|
insertbackground=FG, relief="flat", font=MONO).pack(side=tk.LEFT, **pad)
|
|
tk.Button(bar, text="Browse", bg=BG3, fg=FG, relief="flat",
|
|
activebackground=ACCENT, cursor="hand2",
|
|
command=lambda: self._browse_file(self.bw_var, "raw_bw.bin")
|
|
).pack(side=tk.LEFT, **pad)
|
|
|
|
# Buttons
|
|
tk.Frame(bar, bg=BG2, width=10).pack(side=tk.LEFT)
|
|
self.analyze_btn = tk.Button(bar, text="Analyze", bg=ACCENT, fg="#ffffff",
|
|
relief="flat", padx=10, cursor="hand2",
|
|
font=("Consolas", 9, "bold"),
|
|
command=self._run_analyze)
|
|
self.analyze_btn.pack(side=tk.LEFT, **pad)
|
|
|
|
self.live_btn = tk.Button(bar, text="Live: OFF", bg=BG3, fg=FG,
|
|
relief="flat", padx=10, cursor="hand2",
|
|
font=MONO, command=self._toggle_live)
|
|
self.live_btn.pack(side=tk.LEFT, **pad)
|
|
|
|
self.export_btn = tk.Button(bar, text="Export for Claude", bg=ORANGE, fg="#000000",
|
|
relief="flat", padx=10, cursor="hand2",
|
|
font=("Consolas", 9, "bold"),
|
|
command=self._run_export, state="disabled")
|
|
self.export_btn.pack(side=tk.LEFT, **pad)
|
|
|
|
self.status_var = tk.StringVar(value="Idle")
|
|
tk.Label(bar, textvariable=self.status_var, bg=BG2, fg=FG_DIM,
|
|
font=MONO, anchor="w").pack(side=tk.LEFT, padx=10)
|
|
|
|
def _build_panes(self) -> None:
|
|
pane = tk.PanedWindow(self, orient=tk.HORIZONTAL, bg=BG,
|
|
sashwidth=4, sashrelief="flat")
|
|
pane.pack(fill=tk.BOTH, expand=True, padx=0, pady=0)
|
|
|
|
# ── Left: session/frame tree ──────────────────────────────────────
|
|
left = tk.Frame(pane, bg=BG2, width=260)
|
|
pane.add(left, minsize=200)
|
|
|
|
tk.Label(left, text="Sessions", bg=BG2, fg=ACCENT,
|
|
font=("Consolas", 9, "bold"), anchor="w", padx=6).pack(fill=tk.X)
|
|
|
|
tree_frame = tk.Frame(left, bg=BG2)
|
|
tree_frame.pack(fill=tk.BOTH, expand=True)
|
|
|
|
style = ttk.Style()
|
|
style.theme_use("clam")
|
|
style.configure("Treeview",
|
|
background=BG2, foreground=FG, fieldbackground=BG2,
|
|
font=MONO_SM, rowheight=18, borderwidth=0)
|
|
style.configure("Treeview.Heading",
|
|
background=BG3, foreground=ACCENT, font=MONO_SM)
|
|
style.map("Treeview", background=[("selected", BG3)],
|
|
foreground=[("selected", "#ffffff")])
|
|
|
|
self.tree = ttk.Treeview(tree_frame, columns=("info",), show="tree headings",
|
|
selectmode="browse")
|
|
self.tree.heading("#0", text="Frame")
|
|
self.tree.heading("info", text="Info")
|
|
self.tree.column("#0", width=160, stretch=True)
|
|
self.tree.column("info", width=80, stretch=False)
|
|
|
|
vsb = ttk.Scrollbar(tree_frame, orient="vertical", command=self.tree.yview)
|
|
self.tree.configure(yscrollcommand=vsb.set)
|
|
vsb.pack(side=tk.RIGHT, fill=tk.Y)
|
|
self.tree.pack(fill=tk.BOTH, expand=True)
|
|
|
|
self.tree.tag_configure("session", foreground=ACCENT, font=("Consolas", 9, "bold"))
|
|
self.tree.tag_configure("bw_frame", foreground=COL_BW)
|
|
self.tree.tag_configure("s3_frame", foreground=COL_S3)
|
|
self.tree.tag_configure("bad_chk", foreground=RED)
|
|
self.tree.tag_configure("malformed", foreground=RED)
|
|
|
|
self.tree.bind("<<TreeviewSelect>>", self._on_tree_select)
|
|
|
|
# ── Right: detail notebook ────────────────────────────────────────
|
|
right = tk.Frame(pane, bg=BG)
|
|
pane.add(right, minsize=600)
|
|
|
|
style.configure("TNotebook", background=BG2, borderwidth=0)
|
|
style.configure("TNotebook.Tab", background=BG3, foreground=FG,
|
|
font=MONO, padding=[8, 2])
|
|
style.map("TNotebook.Tab", background=[("selected", BG)],
|
|
foreground=[("selected", ACCENT)])
|
|
|
|
self.nb = ttk.Notebook(right)
|
|
self.nb.pack(fill=tk.BOTH, expand=True)
|
|
|
|
# Tab: Inventory
|
|
self.inv_text = self._make_text_tab("Inventory")
|
|
# Tab: Hex Dump
|
|
self.hex_text = self._make_text_tab("Hex Dump")
|
|
# Tab: Diff
|
|
self.diff_text = self._make_text_tab("Diff")
|
|
# Tab: Full Report (raw text)
|
|
self.report_text = self._make_text_tab("Full Report")
|
|
# Tab: Query (DB)
|
|
self._build_query_tab()
|
|
|
|
# Tag colours for rich text in all tabs
|
|
for w in (self.inv_text, self.hex_text, self.diff_text, self.report_text):
|
|
w.tag_configure("head", foreground=COL_HEAD, font=("Consolas", 9, "bold"))
|
|
w.tag_configure("bw", foreground=COL_BW)
|
|
w.tag_configure("s3", foreground=COL_S3)
|
|
w.tag_configure("changed", foreground=COL_DIFF)
|
|
w.tag_configure("known", foreground=COL_KNOW)
|
|
w.tag_configure("dim", foreground=FG_DIM)
|
|
w.tag_configure("normal", foreground=FG)
|
|
w.tag_configure("warn", foreground=YELLOW)
|
|
w.tag_configure("addr", foreground=ORANGE)
|
|
|
|
def _make_text_tab(self, title: str) -> tk.Text:
|
|
frame = tk.Frame(self.nb, bg=BG)
|
|
self.nb.add(frame, text=title)
|
|
w = tk.Text(frame, bg=BG, fg=FG, font=MONO, state="disabled",
|
|
relief="flat", wrap="none", insertbackground=FG,
|
|
selectbackground=BG3, selectforeground="#ffffff")
|
|
vsb = ttk.Scrollbar(frame, orient="vertical", command=w.yview)
|
|
hsb = ttk.Scrollbar(frame, orient="horizontal", command=w.xview)
|
|
w.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
|
|
vsb.pack(side=tk.RIGHT, fill=tk.Y)
|
|
hsb.pack(side=tk.BOTTOM, fill=tk.X)
|
|
w.pack(fill=tk.BOTH, expand=True)
|
|
return w
|
|
|
|
def _build_query_tab(self) -> None:
|
|
"""Build the Query tab: filter controls + results table + interpretation panel."""
|
|
frame = tk.Frame(self.nb, bg=BG)
|
|
self.nb.add(frame, text="Query DB")
|
|
|
|
# ── Filter row ────────────────────────────────────────────────────
|
|
filt = tk.Frame(frame, bg=BG2, pady=4)
|
|
filt.pack(side=tk.TOP, fill=tk.X)
|
|
|
|
pad = {"padx": 4, "pady": 2}
|
|
|
|
# Capture filter
|
|
tk.Label(filt, text="Capture:", bg=BG2, fg=FG, font=MONO_SM).grid(row=0, column=0, sticky="e", **pad)
|
|
self._q_capture_var = tk.StringVar(value="All")
|
|
self._q_capture_cb = ttk.Combobox(filt, textvariable=self._q_capture_var,
|
|
width=18, font=MONO_SM, state="readonly")
|
|
self._q_capture_cb.grid(row=0, column=1, sticky="w", **pad)
|
|
|
|
# Direction filter
|
|
tk.Label(filt, text="Dir:", bg=BG2, fg=FG, font=MONO_SM).grid(row=0, column=2, sticky="e", **pad)
|
|
self._q_dir_var = tk.StringVar(value="All")
|
|
self._q_dir_cb = ttk.Combobox(filt, textvariable=self._q_dir_var,
|
|
values=["All", "BW", "S3"],
|
|
width=6, font=MONO_SM, state="readonly")
|
|
self._q_dir_cb.grid(row=0, column=3, sticky="w", **pad)
|
|
|
|
# SUB filter
|
|
tk.Label(filt, text="SUB:", bg=BG2, fg=FG, font=MONO_SM).grid(row=0, column=4, sticky="e", **pad)
|
|
self._q_sub_var = tk.StringVar(value="All")
|
|
self._q_sub_cb = ttk.Combobox(filt, textvariable=self._q_sub_var,
|
|
width=12, font=MONO_SM, state="readonly")
|
|
self._q_sub_cb.grid(row=0, column=5, sticky="w", **pad)
|
|
|
|
# Byte offset filter
|
|
tk.Label(filt, text="Offset:", bg=BG2, fg=FG, font=MONO_SM).grid(row=0, column=6, sticky="e", **pad)
|
|
self._q_offset_var = tk.StringVar(value="")
|
|
tk.Entry(filt, textvariable=self._q_offset_var, width=8, bg=BG3, fg=FG,
|
|
font=MONO_SM, insertbackground=FG, relief="flat").grid(row=0, column=7, sticky="w", **pad)
|
|
|
|
# Value filter
|
|
tk.Label(filt, text="Value:", bg=BG2, fg=FG, font=MONO_SM).grid(row=0, column=8, sticky="e", **pad)
|
|
self._q_value_var = tk.StringVar(value="")
|
|
tk.Entry(filt, textvariable=self._q_value_var, width=8, bg=BG3, fg=FG,
|
|
font=MONO_SM, insertbackground=FG, relief="flat").grid(row=0, column=9, sticky="w", **pad)
|
|
|
|
# Run / Refresh buttons
|
|
tk.Button(filt, text="Run Query", bg=ACCENT, fg="#ffffff", relief="flat",
|
|
padx=8, cursor="hand2", font=("Consolas", 8, "bold"),
|
|
command=self._run_db_query).grid(row=0, column=10, padx=8)
|
|
tk.Button(filt, text="Refresh dropdowns", bg=BG3, fg=FG, relief="flat",
|
|
padx=6, cursor="hand2", font=MONO_SM,
|
|
command=self._refresh_query_dropdowns).grid(row=0, column=11, padx=4)
|
|
|
|
# DB stats label
|
|
self._q_stats_var = tk.StringVar(value="DB: —")
|
|
tk.Label(filt, textvariable=self._q_stats_var, bg=BG2, fg=FG_DIM,
|
|
font=MONO_SM).grid(row=0, column=12, padx=12, sticky="w")
|
|
|
|
# ── Results table ─────────────────────────────────────────────────
|
|
res_frame = tk.Frame(frame, bg=BG)
|
|
res_frame.pack(side=tk.TOP, fill=tk.BOTH, expand=True)
|
|
|
|
# Results treeview
|
|
cols = ("cap", "sess", "dir", "sub", "sub_name", "page", "len", "chk")
|
|
self._q_tree = ttk.Treeview(res_frame, columns=cols,
|
|
show="headings", selectmode="browse")
|
|
col_cfg = [
|
|
("cap", "Cap", 40),
|
|
("sess", "Sess", 40),
|
|
("dir", "Dir", 40),
|
|
("sub", "SUB", 50),
|
|
("sub_name", "Name", 160),
|
|
("page", "Page", 60),
|
|
("len", "Len", 50),
|
|
("chk", "Chk", 50),
|
|
]
|
|
for cid, heading, width in col_cfg:
|
|
self._q_tree.heading(cid, text=heading, anchor="w")
|
|
self._q_tree.column(cid, width=width, stretch=(cid == "sub_name"))
|
|
|
|
q_vsb = ttk.Scrollbar(res_frame, orient="vertical", command=self._q_tree.yview)
|
|
q_hsb = ttk.Scrollbar(res_frame, orient="horizontal", command=self._q_tree.xview)
|
|
self._q_tree.configure(yscrollcommand=q_vsb.set, xscrollcommand=q_hsb.set)
|
|
q_vsb.pack(side=tk.RIGHT, fill=tk.Y)
|
|
q_hsb.pack(side=tk.BOTTOM, fill=tk.X)
|
|
self._q_tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
|
|
|
self._q_tree.tag_configure("bw_row", foreground=COL_BW)
|
|
self._q_tree.tag_configure("s3_row", foreground=COL_S3)
|
|
self._q_tree.tag_configure("bad_row", foreground=RED)
|
|
|
|
# ── Interpretation panel (below results) ──────────────────────────
|
|
interp_frame = tk.Frame(frame, bg=BG2, height=120)
|
|
interp_frame.pack(side=tk.BOTTOM, fill=tk.X)
|
|
interp_frame.pack_propagate(False)
|
|
|
|
tk.Label(interp_frame, text="Byte interpretation (click a row, enter offset):",
|
|
bg=BG2, fg=ACCENT, font=MONO_SM, anchor="w", padx=6).pack(fill=tk.X)
|
|
|
|
interp_inner = tk.Frame(interp_frame, bg=BG2)
|
|
interp_inner.pack(fill=tk.X, padx=6, pady=2)
|
|
|
|
tk.Label(interp_inner, text="Offset:", bg=BG2, fg=FG, font=MONO_SM).pack(side=tk.LEFT)
|
|
self._interp_offset_var = tk.StringVar(value="5")
|
|
tk.Entry(interp_inner, textvariable=self._interp_offset_var,
|
|
width=6, bg=BG3, fg=FG, font=MONO_SM,
|
|
insertbackground=FG, relief="flat").pack(side=tk.LEFT, padx=4)
|
|
tk.Button(interp_inner, text="Interpret", bg=BG3, fg=FG, relief="flat",
|
|
cursor="hand2", font=MONO_SM,
|
|
command=self._run_interpret).pack(side=tk.LEFT, padx=4)
|
|
|
|
self._interp_text = tk.Text(interp_frame, bg=BG2, fg=FG, font=MONO_SM,
|
|
height=4, relief="flat", state="disabled",
|
|
insertbackground=FG)
|
|
self._interp_text.pack(fill=tk.X, padx=6, pady=2)
|
|
self._interp_text.tag_configure("label", foreground=FG_DIM)
|
|
self._interp_text.tag_configure("value", foreground=YELLOW)
|
|
|
|
# Store frame rows by tree iid -> db row
|
|
self._q_rows: dict[str, object] = {}
|
|
self._q_capture_rows: list = [None]
|
|
self._q_sub_values: list = [None]
|
|
self._q_tree.bind("<<TreeviewSelect>>", self._on_q_select)
|
|
|
|
# Init dropdowns
|
|
self._refresh_query_dropdowns()
|
|
|
|
def _refresh_query_dropdowns(self) -> None:
|
|
"""Reload capture and SUB dropdowns from the DB."""
|
|
try:
|
|
captures = self._db.list_captures()
|
|
cap_labels = ["All"] + [
|
|
f"#{r['id']} {r['timestamp'][:16]} ({r['frame_count']} frames)"
|
|
for r in captures
|
|
]
|
|
self._q_capture_cb["values"] = cap_labels
|
|
self._q_capture_rows = [None] + [r["id"] for r in captures]
|
|
|
|
subs = self._db.get_distinct_subs()
|
|
sub_labels = ["All"] + [f"0x{s:02X}" for s in subs]
|
|
self._q_sub_cb["values"] = sub_labels
|
|
self._q_sub_values = [None] + subs
|
|
|
|
stats = self._db.get_stats()
|
|
self._q_stats_var.set(
|
|
f"DB: {stats['captures']} captures | {stats['frames']} frames"
|
|
)
|
|
except Exception as exc:
|
|
self._q_stats_var.set(f"DB error: {exc}")
|
|
|
|
def _parse_hex_or_int(self, s: str) -> Optional[int]:
|
|
"""Parse '0x1F', '31', or '' into int or None."""
|
|
s = s.strip()
|
|
if not s:
|
|
return None
|
|
try:
|
|
return int(s, 0)
|
|
except ValueError:
|
|
return None
|
|
|
|
def _run_db_query(self) -> None:
|
|
"""Execute query with current filter values and populate results tree."""
|
|
# Resolve capture_id
|
|
cap_idx = self._q_capture_cb.current()
|
|
cap_id = self._q_capture_rows[cap_idx] if cap_idx > 0 else None
|
|
|
|
# Direction
|
|
dir_val = self._q_dir_var.get()
|
|
direction = dir_val if dir_val != "All" else None
|
|
|
|
# SUB
|
|
sub_idx = self._q_sub_cb.current()
|
|
sub = self._q_sub_values[sub_idx] if sub_idx > 0 else None
|
|
|
|
# Offset / value
|
|
offset = self._parse_hex_or_int(self._q_offset_var.get())
|
|
value = self._parse_hex_or_int(self._q_value_var.get())
|
|
|
|
try:
|
|
if offset is not None:
|
|
rows = self._db.query_by_byte(
|
|
offset=offset, value=value,
|
|
capture_id=cap_id, direction=direction, sub=sub
|
|
)
|
|
else:
|
|
rows = self._db.query_frames(
|
|
capture_id=cap_id, direction=direction, sub=sub
|
|
)
|
|
except Exception as exc:
|
|
messagebox.showerror("Query error", str(exc))
|
|
return
|
|
|
|
# Populate tree
|
|
self._q_tree.delete(*self._q_tree.get_children())
|
|
self._q_rows.clear()
|
|
|
|
for row in rows:
|
|
sub_hex = f"0x{row['sub']:02X}" if row["sub"] is not None else "—"
|
|
page_hex = f"0x{row['page_key']:04X}" if row["page_key"] is not None else "—"
|
|
chk_str = {1: "OK", 0: "BAD", None: "—"}.get(row["checksum_ok"], "—")
|
|
tag = "bw_row" if row["direction"] == "BW" else "s3_row"
|
|
if row["checksum_ok"] == 0:
|
|
tag = "bad_row"
|
|
|
|
iid = str(row["id"])
|
|
self._q_tree.insert("", tk.END, iid=iid, tags=(tag,), values=(
|
|
row["capture_id"],
|
|
row["session_idx"],
|
|
row["direction"],
|
|
sub_hex,
|
|
row["sub_name"] or "",
|
|
page_hex,
|
|
row["payload_len"],
|
|
chk_str,
|
|
))
|
|
self._q_rows[iid] = row
|
|
|
|
self.sb_var.set(f"Query returned {len(rows)} rows")
|
|
|
|
def _on_q_select(self, _event: tk.Event) -> None:
|
|
"""When a DB result row is selected, auto-run interpret at current offset."""
|
|
self._run_interpret()
|
|
|
|
def _run_interpret(self) -> None:
|
|
"""Show multi-format byte interpretation for the selected row + offset."""
|
|
sel = self._q_tree.selection()
|
|
if not sel:
|
|
return
|
|
iid = sel[0]
|
|
row = self._q_rows.get(iid)
|
|
if row is None:
|
|
return
|
|
|
|
offset = self._parse_hex_or_int(self._interp_offset_var.get())
|
|
if offset is None:
|
|
return
|
|
|
|
payload = bytes(row["payload"])
|
|
interp = self._db.interpret_offset(payload, offset)
|
|
|
|
w = self._interp_text
|
|
w.configure(state="normal")
|
|
w.delete("1.0", tk.END)
|
|
|
|
sub_hex = f"0x{row['sub']:02X}" if row["sub"] is not None else "??"
|
|
w.insert(tk.END, f"Frame #{row['id']} [{row['direction']}] SUB={sub_hex} "
|
|
f"offset={offset} (0x{offset:04X})\n", "label")
|
|
|
|
label_order = [
|
|
("uint8", "uint8 "),
|
|
("int8", "int8 "),
|
|
("uint16_be", "uint16 BE "),
|
|
("uint16_le", "uint16 LE "),
|
|
("uint32_be", "uint32 BE "),
|
|
("uint32_le", "uint32 LE "),
|
|
("float32_be", "float32 BE "),
|
|
("float32_le", "float32 LE "),
|
|
]
|
|
line = ""
|
|
for key, label in label_order:
|
|
if key in interp:
|
|
val = interp[key]
|
|
if isinstance(val, float):
|
|
val_str = f"{val:.6g}"
|
|
else:
|
|
val_str = str(val)
|
|
if key.startswith("uint") or key.startswith("int"):
|
|
val_str += f" (0x{int(val) & 0xFFFFFFFF:X})"
|
|
chunk = f"{label}: {val_str}"
|
|
line += f" {chunk:<30}"
|
|
if len(line) > 80:
|
|
w.insert(tk.END, line + "\n", "value")
|
|
line = ""
|
|
if line:
|
|
w.insert(tk.END, line + "\n", "value")
|
|
|
|
w.configure(state="disabled")
|
|
|
|
def _build_statusbar(self) -> None:
|
|
bar = tk.Frame(self, bg=BG3, height=20)
|
|
bar.pack(side=tk.BOTTOM, fill=tk.X)
|
|
self.sb_var = tk.StringVar(value="Ready")
|
|
tk.Label(bar, textvariable=self.sb_var, bg=BG3, fg=FG_DIM,
|
|
font=MONO_SM, anchor="w", padx=6).pack(fill=tk.X)
|
|
|
|
# ── file picking ───────────────────────────────────────────────────────
|
|
|
|
def _browse_file(self, var: tk.StringVar, default_name: str) -> None:
|
|
path = filedialog.askopenfilename(
|
|
title=f"Select {default_name}",
|
|
filetypes=[("Binary files", "*.bin"), ("All files", "*.*")],
|
|
initialfile=default_name,
|
|
)
|
|
if path:
|
|
var.set(path)
|
|
|
|
# ── analysis ──────────────────────────────────────────────────────────
|
|
|
|
def _run_analyze(self) -> None:
|
|
s3_path = Path(self.s3_var.get().strip()) if self.s3_var.get().strip() else None
|
|
bw_path = Path(self.bw_var.get().strip()) if self.bw_var.get().strip() else None
|
|
|
|
if not s3_path or not bw_path:
|
|
messagebox.showerror("Missing files", "Please select both S3 and BW raw files.")
|
|
return
|
|
if not s3_path.exists():
|
|
messagebox.showerror("File not found", f"S3 file not found:\n{s3_path}")
|
|
return
|
|
if not bw_path.exists():
|
|
messagebox.showerror("File not found", f"BW file not found:\n{bw_path}")
|
|
return
|
|
|
|
self.state.s3_path = s3_path
|
|
self.state.bw_path = bw_path
|
|
self._do_analyze(s3_path, bw_path)
|
|
|
|
def _run_export(self) -> None:
|
|
if not self.state.sessions:
|
|
messagebox.showinfo("Export", "Run Analyze first.")
|
|
return
|
|
|
|
outdir = self.state.s3_path.parent if self.state.s3_path else Path(".")
|
|
out_path = write_claude_export(
|
|
self.state.sessions,
|
|
self.state.diffs,
|
|
outdir,
|
|
self.state.s3_path,
|
|
self.state.bw_path,
|
|
)
|
|
|
|
self.sb_var.set(f"Exported: {out_path.name}")
|
|
if messagebox.askyesno(
|
|
"Export complete",
|
|
f"Saved to:\n{out_path}\n\nOpen the folder?",
|
|
):
|
|
import subprocess
|
|
subprocess.Popen(["explorer", str(out_path.parent)])
|
|
|
|
def _do_analyze(self, s3_path: Path, bw_path: Path) -> None:
|
|
self.status_var.set("Parsing...")
|
|
self.update_idletasks()
|
|
|
|
s3_blob = s3_path.read_bytes()
|
|
bw_blob = bw_path.read_bytes()
|
|
|
|
s3_frames = annotate_frames(parse_s3(s3_blob, trailer_len=0), "S3")
|
|
bw_frames = annotate_frames(parse_bw(bw_blob, trailer_len=0, validate_checksum=True), "BW")
|
|
|
|
sessions = split_into_sessions(bw_frames, s3_frames)
|
|
|
|
diffs: list[Optional[list[FrameDiff]]] = [None]
|
|
for i in range(1, len(sessions)):
|
|
diffs.append(diff_sessions(sessions[i - 1], sessions[i]))
|
|
|
|
self.state.sessions = sessions
|
|
self.state.diffs = diffs
|
|
|
|
n_s3 = sum(len(s.s3_frames) for s in sessions)
|
|
n_bw = sum(len(s.bw_frames) for s in sessions)
|
|
self.status_var.set(
|
|
f"{len(sessions)} sessions | BW: {n_bw} frames S3: {n_s3} frames"
|
|
)
|
|
self.sb_var.set(f"Loaded: {s3_path.name} + {bw_path.name}")
|
|
|
|
self.export_btn.configure(state="normal")
|
|
self._rebuild_tree()
|
|
|
|
# Auto-ingest into DB (deduped by SHA256 — fast no-op on re-analyze)
|
|
try:
|
|
cap_id = self._db.ingest(sessions, s3_path, bw_path)
|
|
if cap_id is not None:
|
|
self.state.last_capture_id = cap_id
|
|
self._refresh_query_dropdowns()
|
|
# Pre-select this capture in the Query tab
|
|
cap_labels = list(self._q_capture_cb["values"])
|
|
# Find label that starts with #<cap_id>
|
|
for i, lbl in enumerate(cap_labels):
|
|
if lbl.startswith(f"#{cap_id} "):
|
|
self._q_capture_cb.current(i)
|
|
break
|
|
# else: already ingested — no change to dropdown selection
|
|
except Exception as exc:
|
|
self.sb_var.set(f"DB ingest error: {exc}")
|
|
|
|
# ── tree building ──────────────────────────────────────────────────────
|
|
|
|
def _rebuild_tree(self) -> None:
|
|
self.tree.delete(*self.tree.get_children())
|
|
|
|
for sess in self.state.sessions:
|
|
is_complete = any(
|
|
af.header is not None and af.header.sub == 0x74
|
|
for af in sess.bw_frames
|
|
)
|
|
label = f"Session {sess.index}"
|
|
if not is_complete:
|
|
label += " [partial]"
|
|
n_diff = len(self.state.diffs[sess.index] or [])
|
|
diff_info = f"{n_diff} changes" if n_diff > 0 else ""
|
|
sess_id = self.tree.insert("", tk.END, text=label,
|
|
values=(diff_info,), tags=("session",))
|
|
|
|
for af in sess.all_frames:
|
|
src_tag = "bw_frame" if af.source == "BW" else "s3_frame"
|
|
sub_hex = f"{af.header.sub:02X}" if af.header else "??"
|
|
label_text = f"[{af.source}] {sub_hex} {af.sub_name}"
|
|
extra = ""
|
|
tags = (src_tag,)
|
|
if af.frame.checksum_valid is False:
|
|
extra = "BAD CHK"
|
|
tags = ("bad_chk",)
|
|
elif af.header is None:
|
|
tags = ("malformed",)
|
|
label_text = f"[{af.source}] MALFORMED"
|
|
self.tree.insert(sess_id, tk.END, text=label_text,
|
|
values=(extra,), tags=tags,
|
|
iid=f"frame_{sess.index}_{af.frame.index}_{af.source}")
|
|
|
|
# Expand all sessions
|
|
for item in self.tree.get_children():
|
|
self.tree.item(item, open=True)
|
|
|
|
# ── tree selection → detail panel ─────────────────────────────────────
|
|
|
|
def _on_tree_select(self, _event: tk.Event) -> None:
|
|
sel = self.tree.selection()
|
|
if not sel:
|
|
return
|
|
iid = sel[0]
|
|
|
|
# Determine if it's a session node or a frame node
|
|
if iid.startswith("frame_"):
|
|
# frame_<sessidx>_<frameidx>_<source>
|
|
parts = iid.split("_")
|
|
sess_idx = int(parts[1])
|
|
frame_idx = int(parts[2])
|
|
source = parts[3]
|
|
self._show_frame_detail(sess_idx, frame_idx, source)
|
|
else:
|
|
# Session node — show session summary
|
|
# Find session index from text
|
|
text = self.tree.item(iid, "text")
|
|
try:
|
|
idx = int(text.split()[1])
|
|
self._show_session_detail(idx)
|
|
except (IndexError, ValueError):
|
|
pass
|
|
|
|
def _find_frame(self, sess_idx: int, frame_idx: int, source: str) -> Optional[AnnotatedFrame]:
|
|
if sess_idx >= len(self.state.sessions):
|
|
return None
|
|
sess = self.state.sessions[sess_idx]
|
|
pool = sess.bw_frames if source == "BW" else sess.s3_frames
|
|
for af in pool:
|
|
if af.frame.index == frame_idx:
|
|
return af
|
|
return None
|
|
|
|
# ── detail renderers ──────────────────────────────────────────────────
|
|
|
|
def _clear_all_tabs(self) -> None:
|
|
for w in (self.inv_text, self.hex_text, self.diff_text, self.report_text):
|
|
self._text_clear(w)
|
|
|
|
def _show_session_detail(self, sess_idx: int) -> None:
|
|
if sess_idx >= len(self.state.sessions):
|
|
return
|
|
sess = self.state.sessions[sess_idx]
|
|
diffs = self.state.diffs[sess_idx]
|
|
|
|
self._clear_all_tabs()
|
|
|
|
# ── Inventory tab ────────────────────────────────────────────────
|
|
w = self.inv_text
|
|
self._text_clear(w)
|
|
self._tw(w, f"SESSION {sess.index}", "head"); self._tn(w)
|
|
n_bw, n_s3 = len(sess.bw_frames), len(sess.s3_frames)
|
|
self._tw(w, f"Frames: {n_bw + n_s3} (BW: {n_bw}, S3: {n_s3})\n", "normal")
|
|
if n_bw != n_s3:
|
|
self._tw(w, " WARNING: BW/S3 count mismatch\n", "warn")
|
|
self._tn(w)
|
|
|
|
for seq_i, af in enumerate(sess.all_frames):
|
|
src_tag = "bw" if af.source == "BW" else "s3"
|
|
sub_hex = f"{af.header.sub:02X}" if af.header else "??"
|
|
page_str = f" (page {af.header.page_key:04X})" if af.header and af.header.page_key != 0 else ""
|
|
chk = ""
|
|
if af.frame.checksum_valid is False:
|
|
chk = " [BAD CHECKSUM]"
|
|
elif af.frame.checksum_valid is True:
|
|
chk = f" [{af.frame.checksum_type}]"
|
|
self._tw(w, f" [{af.source}] #{seq_i:<3} ", src_tag)
|
|
self._tw(w, f"SUB={sub_hex} ", "addr")
|
|
self._tw(w, f"{af.sub_name:<30}", src_tag)
|
|
self._tw(w, f"{page_str} len={len(af.frame.payload)}", "dim")
|
|
if chk:
|
|
self._tw(w, chk, "warn" if af.frame.checksum_valid is False else "dim")
|
|
self._tn(w)
|
|
|
|
# ── Diff tab ─────────────────────────────────────────────────────
|
|
w = self.diff_text
|
|
self._text_clear(w)
|
|
if diffs is None:
|
|
self._tw(w, "(No previous session to diff against)\n", "dim")
|
|
elif not diffs:
|
|
self._tw(w, f"DIFF vs SESSION {sess_idx - 1}\n", "head"); self._tn(w)
|
|
self._tw(w, " No changes detected.\n", "dim")
|
|
else:
|
|
self._tw(w, f"DIFF vs SESSION {sess_idx - 1}\n", "head"); self._tn(w)
|
|
for fd in diffs:
|
|
page_str = f" (page {fd.page_key:04X})" if fd.page_key != 0 else ""
|
|
self._tw(w, f"\n SUB {fd.sub:02X} ({fd.sub_name}){page_str}:\n", "addr")
|
|
for bd in fd.diffs:
|
|
before_s = f"{bd.before:02x}" if bd.before >= 0 else "--"
|
|
after_s = f"{bd.after:02x}" if bd.after >= 0 else "--"
|
|
self._tw(w, f" [{bd.payload_offset:3d}] 0x{bd.payload_offset:04X}: ", "dim")
|
|
self._tw(w, f"{before_s} -> {after_s}", "changed")
|
|
if bd.field_name:
|
|
self._tw(w, f" [{bd.field_name}]", "known")
|
|
self._tn(w)
|
|
|
|
# ── Full Report tab ───────────────────────────────────────────────
|
|
report_text = render_session_report(sess, diffs, sess_idx - 1 if sess_idx > 0 else None)
|
|
w = self.report_text
|
|
self._text_clear(w)
|
|
self._tw(w, report_text, "normal")
|
|
|
|
# Switch to Inventory tab
|
|
self.nb.select(0)
|
|
|
|
def _show_frame_detail(self, sess_idx: int, frame_idx: int, source: str) -> None:
|
|
af = self._find_frame(sess_idx, frame_idx, source)
|
|
if af is None:
|
|
return
|
|
|
|
self._clear_all_tabs()
|
|
src_tag = "bw" if source == "BW" else "s3"
|
|
sub_hex = f"{af.header.sub:02X}" if af.header else "??"
|
|
|
|
# ── Inventory tab — single frame summary ─────────────────────────
|
|
w = self.inv_text
|
|
self._tw(w, f"[{af.source}] Frame #{af.frame.index}\n", src_tag)
|
|
self._tw(w, f"Session {sess_idx} | ", "dim")
|
|
self._tw(w, f"SUB={sub_hex} {af.sub_name}\n", "addr")
|
|
if af.header:
|
|
self._tw(w, f" OFFSET: {af.header.page_key:04X} ", "dim")
|
|
self._tw(w, f"CMD={af.header.cmd:02X} FLAGS={af.header.flags:02X}\n", "dim")
|
|
self._tn(w)
|
|
self._tw(w, f"Payload bytes: {len(af.frame.payload)}\n", "dim")
|
|
if af.frame.checksum_valid is False:
|
|
self._tw(w, " BAD CHECKSUM\n", "warn")
|
|
elif af.frame.checksum_valid is True:
|
|
self._tw(w, f" Checksum: {af.frame.checksum_type} {af.frame.checksum_hex}\n", "dim")
|
|
self._tn(w)
|
|
|
|
# Protocol header breakdown
|
|
p = af.frame.payload
|
|
if len(p) >= 5:
|
|
self._tw(w, "Header breakdown:\n", "head")
|
|
self._tw(w, f" [0] CMD = {p[0]:02x}\n", "dim")
|
|
self._tw(w, f" [1] ? = {p[1]:02x}\n", "dim")
|
|
self._tw(w, f" [2] SUB = {p[2]:02x} ({af.sub_name})\n", src_tag)
|
|
self._tw(w, f" [3] OFFSET_HI = {p[3]:02x}\n", "dim")
|
|
self._tw(w, f" [4] OFFSET_LO = {p[4]:02x}\n", "dim")
|
|
if len(p) > 5:
|
|
self._tw(w, f" [5..] data = {len(p) - 5} bytes\n", "dim")
|
|
|
|
# ── Hex Dump tab ─────────────────────────────────────────────────
|
|
w = self.hex_text
|
|
self._tw(w, f"[{af.source}] SUB={sub_hex} {af.sub_name}\n", src_tag)
|
|
self._tw(w, f"Payload ({len(af.frame.payload)} bytes):\n", "dim")
|
|
self._tn(w)
|
|
dump_lines = format_hex_dump(af.frame.payload, indent=" ")
|
|
self._tw(w, "\n".join(dump_lines) + "\n", "normal")
|
|
|
|
# Annotate known field offsets within this frame
|
|
diffs_for_sess = self.state.diffs[sess_idx] if sess_idx < len(self.state.diffs) else None
|
|
if diffs_for_sess and af.header:
|
|
page_key = af.header.page_key
|
|
matching = [fd for fd in diffs_for_sess
|
|
if fd.sub == af.header.sub and fd.page_key == page_key]
|
|
if matching:
|
|
self._tn(w)
|
|
self._tw(w, "Changed bytes in this frame (vs prev session):\n", "head")
|
|
for bd in matching[0].diffs:
|
|
before_s = f"{bd.before:02x}" if bd.before >= 0 else "--"
|
|
after_s = f"{bd.after:02x}" if bd.after >= 0 else "--"
|
|
self._tw(w, f" [{bd.payload_offset:3d}] 0x{bd.payload_offset:04X}: ", "dim")
|
|
self._tw(w, f"{before_s} -> {after_s}", "changed")
|
|
if bd.field_name:
|
|
self._tw(w, f" [{bd.field_name}]", "known")
|
|
self._tn(w)
|
|
|
|
# Switch to Hex Dump tab for frame selection
|
|
self.nb.select(1)
|
|
|
|
# ── live mode ─────────────────────────────────────────────────────────
|
|
|
|
def _toggle_live(self) -> None:
|
|
if self._live_thread and self._live_thread.is_alive():
|
|
self._live_stop.set()
|
|
self.live_btn.configure(text="Live: OFF", bg=BG3, fg=FG)
|
|
self.status_var.set("Live stopped")
|
|
else:
|
|
s3_path = Path(self.s3_var.get().strip()) if self.s3_var.get().strip() else None
|
|
bw_path = Path(self.bw_var.get().strip()) if self.bw_var.get().strip() else None
|
|
if not s3_path or not bw_path:
|
|
messagebox.showerror("Missing files", "Select both raw files before starting live mode.")
|
|
return
|
|
self.state.s3_path = s3_path
|
|
self.state.bw_path = bw_path
|
|
self._live_stop.clear()
|
|
self._live_thread = threading.Thread(
|
|
target=self._live_worker, args=(s3_path, bw_path), daemon=True)
|
|
self._live_thread.start()
|
|
self.live_btn.configure(text="Live: ON", bg=GREEN, fg="#000000")
|
|
self.status_var.set("Live mode running...")
|
|
|
|
def _live_worker(self, s3_path: Path, bw_path: Path) -> None:
|
|
s3_buf = bytearray()
|
|
bw_buf = bytearray()
|
|
s3_pos = bw_pos = 0
|
|
|
|
while not self._live_stop.is_set():
|
|
changed = False
|
|
if s3_path.exists():
|
|
with s3_path.open("rb") as fh:
|
|
fh.seek(s3_pos)
|
|
nb = fh.read()
|
|
if nb:
|
|
s3_buf.extend(nb); s3_pos += len(nb); changed = True
|
|
if bw_path.exists():
|
|
with bw_path.open("rb") as fh:
|
|
fh.seek(bw_pos)
|
|
nb = fh.read()
|
|
if nb:
|
|
bw_buf.extend(nb); bw_pos += len(nb); changed = True
|
|
|
|
if changed:
|
|
self._live_q.put("refresh")
|
|
|
|
time.sleep(0.1)
|
|
|
|
def _poll_live_queue(self) -> None:
|
|
try:
|
|
while True:
|
|
msg = self._live_q.get_nowait()
|
|
if msg == "refresh" and self.state.s3_path and self.state.bw_path:
|
|
self._do_analyze(self.state.s3_path, self.state.bw_path)
|
|
except queue.Empty:
|
|
pass
|
|
finally:
|
|
self.after(150, self._poll_live_queue)
|
|
|
|
# ── text helpers ──────────────────────────────────────────────────────
|
|
|
|
def _text_clear(self, w: tk.Text) -> None:
|
|
w.configure(state="normal")
|
|
w.delete("1.0", tk.END)
|
|
# leave enabled for further inserts
|
|
|
|
def _tw(self, w: tk.Text, text: str, tag: str = "normal") -> None:
|
|
"""Insert text with a colour tag."""
|
|
w.configure(state="normal")
|
|
w.insert(tk.END, text, tag)
|
|
|
|
def _tn(self, w: tk.Text) -> None:
|
|
"""Insert newline."""
|
|
w.configure(state="normal")
|
|
w.insert(tk.END, "\n")
|
|
w.configure(state="disabled")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Entry point
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
app = AnalyzerGUI()
|
|
app.mainloop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|