Skip to content

Commit 77b7b0c

Browse files
widgetiiclaude
andauthored
Add firmware auto-download to TUI (#5)
* Add firmware auto-download to TUI The TUI main screen now shows a download button when a chip with OpenIPC firmware is selected. Shows cached status, download progress, and error handling. Local file input still available below. Flow: 1. Select chip → "Download U-Boot for hi3516ev300" button appears 2. Click → downloads and caches firmware 3. Start button enables automatically 4. Cached firmware shown with size on subsequent launches Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Fix TUI crashes, duplicate messages, and handshake timeouts - Fix BadIdentifier crash: stage ids now replace spaces with hyphens - Remove duplicate log messages (Boot mode entered x2, Recovery complete x3) - Remove artificial handshake timeouts — wait forever for user to power-cycle - Add "power-cycle the device now!" to all handshake messages (all 3 protocols) - Apply same fixes to WebUI JS (infinite loop, consistent messages) - Add regression test for ProgressScreen stage id rendering Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Fix mypy: explicit str() for Input.value Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Dmitry Ilyin <widgetii@users.noreply.github.com> Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 3aa3d43 commit 77b7b0c

9 files changed

Lines changed: 186 additions & 62 deletions

File tree

‎src/defib/protocol/hisilicon_cv6xx.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ async def handshake(
166166
"""Send DEADBEEF handshake until 'uart ddr' or 'uart flash' response."""
167167
_emit(on_progress, ProgressEvent(
168168
stage=Stage.HANDSHAKE, bytes_sent=0, bytes_total=1,
169-
message="Sending CV6xx handshake...",
169+
message="Waiting for bootrom... power-cycle the device now!",
170170
))
171171

172172
# Build handshake frame: magic + baudrate(LE) + serial params + CRC(LE)

‎src/defib/protocol/hisilicon_standard.py‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@
4040
BOOTMODE_MARKER = b"\x20"
4141
BOOTMODE_COUNT = 5
4242
BOOTMODE_ACK = b"\xaa"
43-
MAX_INIT_READS = 30
43+
MAX_INIT_READS = 30 # Only used by tests; interactive mode loops forever
4444

4545
FRAME_SEND_RETRIES_SHORT = 16
4646
FRAME_SEND_RETRIES_LONG = 32
@@ -77,11 +77,11 @@ async def handshake(
7777
"""Wait for bootrom 0x20 pattern and send 0xAA acknowledgment."""
7878
_emit(on_progress, ProgressEvent(
7979
stage=Stage.HANDSHAKE, bytes_sent=0, bytes_total=1,
80-
message="Waiting for bootrom...",
80+
message="Waiting for bootrom... power-cycle the device now!",
8181
))
8282

8383
counter = 0
84-
for i in range(MAX_INIT_READS):
84+
while True:
8585
try:
8686
byte = await transport.read(1, timeout=1.0)
8787
except TransportTimeout:
@@ -105,8 +105,6 @@ async def handshake(
105105
))
106106
return HandshakeResult(success=True, message="Boot mode entered")
107107

108-
return HandshakeResult(success=False, message="Bootrom handshake timeout")
109-
110108
async def _send_frame_with_retry(
111109
self,
112110
transport: Transport,

‎src/defib/protocol/hisilicon_v500.py‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -68,15 +68,14 @@ async def handshake(
6868
"""Send V500 handshake frame until device responds with chip ID."""
6969
_emit(on_progress, ProgressEvent(
7070
stage=Stage.HANDSHAKE, bytes_sent=0, bytes_total=1,
71-
message="Sending V500 handshake...",
71+
message="Waiting for bootrom... power-cycle the device now!",
7272
))
7373

7474
handshake_frame = append_crc(
7575
V500_HANDSHAKE_MAGIC + b"\x00\x00\x00\x00\x00\x00\x00\x00"
7676
)
7777

78-
start_time = time.monotonic()
79-
while time.monotonic() - start_time < HANDSHAKE_TIMEOUT:
78+
while True:
8079
await transport.write(handshake_frame)
8180
try:
8281
response = await transport.read(14, timeout=0.1)
@@ -95,8 +94,6 @@ async def handshake(
9594
except TransportTimeout:
9695
continue
9796

98-
return HandshakeResult(success=False, message="V500 handshake timeout")
99-
10097
async def _send_frame_wait_ack(
10198
self,
10299
transport: Transport,

‎src/defib/recovery/session.py‎

Lines changed: 7 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -101,8 +101,8 @@ async def run(
101101
elapsed_ms=elapsed,
102102
)
103103

104-
if on_log:
105-
on_log(LogEvent(level="info", message=handshake.message))
104+
# Note: handshake success is already reported via on_progress,
105+
# so we don't duplicate it here via on_log.
106106

107107
# Firmware transfer
108108
firmware = self._load_firmware()
@@ -126,13 +126,10 @@ async def run(
126126
except Exception:
127127
pass
128128

129-
if on_log:
130-
if result.success:
131-
on_log(LogEvent(
132-
level="info",
133-
message=f"Recovery complete in {result.elapsed_ms:.0f}ms",
134-
))
135-
else:
136-
on_log(LogEvent(level="error", message=f"Recovery failed: {result.error}"))
129+
# Note: completion/failure is already reported via on_progress
130+
# (Stage.COMPLETE event). We only log errors here that weren't
131+
# already surfaced by the protocol.
132+
if on_log and not result.success:
133+
on_log(LogEvent(level="error", message=f"Recovery failed: {result.error}"))
137134

138135
return result

‎src/defib/tui/screens/main.py‎

Lines changed: 139 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
"""Main setup screen: chip selector, file picker, port selector, start button."""
1+
"""Main setup screen: chip selector, firmware (auto-download or local), port selector."""
22

33
from __future__ import annotations
44

@@ -18,14 +18,14 @@
1818
Checkbox,
1919
)
2020

21+
from defib.firmware import has_firmware, download_firmware, get_cached_path
2122
from defib.profiles.loader import list_all_chips
2223

2324

2425
def _get_serial_ports() -> list[tuple[str, str]]:
2526
"""Get available serial ports as (label, value) tuples."""
2627
try:
2728
from serial.tools.list_ports import comports
28-
# Filter out ghost/placeholder ports (no USB vendor ID = not real)
2929
ports = sorted(
3030
[p for p in comports() if p.vid is not None],
3131
key=lambda p: p.device,
@@ -47,9 +47,9 @@ class MainScreen(Screen[None]):
4747
}
4848
4949
#form-container {
50-
width: 70;
50+
width: 74;
5151
height: auto;
52-
max-height: 30;
52+
max-height: 34;
5353
border: thick $accent;
5454
padding: 1 2;
5555
background: $panel;
@@ -72,6 +72,18 @@ class MainScreen(Screen[None]):
7272
width: 100%;
7373
}
7474
75+
#fw-status {
76+
height: 1;
77+
color: $success;
78+
margin-top: 0;
79+
}
80+
81+
#fw-hint {
82+
height: 1;
83+
color: $text-muted;
84+
text-style: italic;
85+
}
86+
7587
#button-row {
7688
margin-top: 1;
7789
align: center middle;
@@ -82,8 +94,8 @@ class MainScreen(Screen[None]):
8294
min-width: 20;
8395
}
8496
85-
#chip-input {
86-
width: 100%;
97+
#download-btn {
98+
min-width: 30;
8799
}
88100
89101
#file-input {
@@ -110,9 +122,17 @@ def compose(self) -> ComposeResult:
110122
allow_blank=True,
111123
)
112124

113-
yield Label("Firmware File:")
125+
yield Label("Firmware:")
126+
yield Button(
127+
"Select a chip first",
128+
variant="default",
129+
id="download-btn",
130+
disabled=True,
131+
)
132+
yield Static("", id="fw-status")
133+
yield Static("", id="fw-hint")
114134
yield Input(
115-
placeholder="/path/to/u-boot.bin",
135+
placeholder="Or enter path to local firmware file",
116136
id="file-input",
117137
)
118138

@@ -125,34 +145,136 @@ def compose(self) -> ComposeResult:
125145
value=port_options[0][1] if port_options else "",
126146
)
127147

128-
yield Checkbox("Send Ctrl-C after upload (enter U-Boot console)", id="break-check")
148+
yield Checkbox(
149+
"Send Ctrl-C after upload (enter U-Boot console)",
150+
id="break-check",
151+
)
129152

130153
with Horizontal(id="button-row"):
131154
yield Button("Start Recovery", variant="primary", id="start-btn")
132155

133156
yield Footer()
134157

158+
def _get_chip(self) -> str:
159+
sel = self.query_one("#chip-select", Select)
160+
return str(sel.value) if sel.value != Select.BLANK else ""
161+
162+
def on_select_changed(self, event: Select.Changed) -> None:
163+
if event.select.id == "chip-select":
164+
self._on_chip_changed()
165+
166+
def on_input_changed(self, event: Input.Changed) -> None:
167+
if event.input.id == "file-input":
168+
self._update_start_button()
169+
170+
def _on_chip_changed(self) -> None:
171+
chip = self._get_chip()
172+
dl_btn = self.query_one("#download-btn", Button)
173+
hint = self.query_one("#fw-hint", Static)
174+
status = self.query_one("#fw-status", Static)
175+
176+
if not chip:
177+
dl_btn.label = "Select a chip first"
178+
dl_btn.disabled = True
179+
dl_btn.variant = "default"
180+
hint.update("")
181+
status.update("")
182+
elif has_firmware(chip):
183+
cached = get_cached_path(chip)
184+
if cached:
185+
dl_btn.label = f"Re-download U-Boot for {chip}"
186+
dl_btn.disabled = False
187+
dl_btn.variant = "default"
188+
status.update(f"✓ Cached: {cached.name} ({cached.stat().st_size // 1024} KB)")
189+
else:
190+
dl_btn.label = f"Download U-Boot for {chip}"
191+
dl_btn.disabled = False
192+
dl_btn.variant = "success"
193+
status.update("")
194+
hint.update("Or enter a local file path below for custom builds.")
195+
else:
196+
dl_btn.label = "No OpenIPC build available"
197+
dl_btn.disabled = True
198+
dl_btn.variant = "default"
199+
hint.update("Enter a local firmware file path below.")
200+
status.update("")
201+
202+
self._update_start_button()
203+
135204
def on_button_pressed(self, event: Button.Pressed) -> None:
136205
if event.button.id == "start-btn":
137206
self._start_recovery()
207+
elif event.button.id == "download-btn":
208+
self._download_firmware()
209+
210+
def _download_firmware(self) -> None:
211+
chip = self._get_chip()
212+
if not chip or not has_firmware(chip):
213+
return
214+
215+
dl_btn = self.query_one("#download-btn", Button)
216+
status = self.query_one("#fw-status", Static)
217+
218+
dl_btn.label = "Downloading..."
219+
dl_btn.disabled = True
220+
status.update("")
221+
222+
try:
223+
path = download_firmware(chip)
224+
status.update(f"✓ {path.name} ({path.stat().st_size // 1024} KB)")
225+
dl_btn.label = f"Re-download U-Boot for {chip}"
226+
dl_btn.disabled = False
227+
dl_btn.variant = "default"
228+
self.notify(
229+
f"Downloaded {path.name}",
230+
severity="information",
231+
title="Firmware Ready",
232+
)
233+
except (ValueError, ConnectionError) as e:
234+
status.update("")
235+
dl_btn.label = "Download failed — retry?"
236+
dl_btn.disabled = False
237+
dl_btn.variant = "error"
238+
self.notify(str(e), severity="error", title="Download Failed")
239+
240+
self._update_start_button()
241+
242+
def _get_firmware_path(self) -> str:
243+
"""Get firmware path: local file input takes priority, then cached download."""
244+
local = str(self.query_one("#file-input", Input).value).strip()
245+
if local:
246+
return local
247+
248+
chip = self._get_chip()
249+
if chip:
250+
cached = get_cached_path(chip)
251+
if cached:
252+
return str(cached)
253+
254+
return ""
255+
256+
def _update_start_button(self) -> None:
257+
chip = self._get_chip()
258+
firmware = self._get_firmware_path()
259+
port_sel = self.query_one("#port-select", Select)
260+
port = str(port_sel.value) if port_sel.value != Select.BLANK else ""
261+
262+
self.query_one("#start-btn", Button).disabled = not (chip and firmware and port)
138263

139264
def _start_recovery(self) -> None:
140-
chip_select = self.query_one("#chip-select", Select)
141-
file_input = self.query_one("#file-input", Input)
142-
port_select = self.query_one("#port-select", Select)
265+
chip = self._get_chip()
266+
firmware_path = self._get_firmware_path()
267+
port_sel = self.query_one("#port-select", Select)
268+
port = str(port_sel.value) if port_sel.value != Select.BLANK else ""
143269
break_check = self.query_one("#break-check", Checkbox)
144-
145-
chip = str(chip_select.value) if chip_select.value != Select.BLANK else ""
146-
firmware_path = file_input.value.strip()
147-
port = str(port_select.value) if port_select.value != Select.BLANK else ""
148270
send_break = break_check.value
149271

150272
# Validation
151273
errors: list[str] = []
152274
if not chip:
153275
errors.append("Select a chip model")
154276
if not firmware_path:
155-
errors.append("Enter a firmware file path")
277+
errors.append("Download firmware or enter a file path")
156278
elif not Path(firmware_path).is_file():
157279
errors.append(f"File not found: {firmware_path}")
158280
if not port:
@@ -162,7 +284,6 @@ def _start_recovery(self) -> None:
162284
self.notify("\n".join(errors), severity="error", title="Validation Error")
163285
return
164286

165-
# Start recovery via the app
166287
from defib.tui.app import DefibApp
167288
app = self.app
168289
if isinstance(app, DefibApp):

‎src/defib/tui/screens/progress.py‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,8 @@ def compose(self) -> ComposeResult:
139139
with Horizontal(id="stages-row"):
140140
stages = ["Handshake", "DDR Init", "SPL/GSL", "U-Boot"]
141141
for name in stages:
142-
indicator = StageIndicator(name, id=f"stage-{name.lower().replace('/', '-')}")
142+
safe_id = name.lower().replace("/", "-").replace(" ", "-")
143+
indicator = StageIndicator(name, id=f"stage-{safe_id}")
143144
self._stage_widgets[name.lower()] = indicator
144145
yield indicator
145146

@@ -204,7 +205,7 @@ def _on_progress(self, event: ProgressEvent) -> None:
204205
self._stage_widgets[indicator_name].set_active()
205206
self._current_stage = indicator_name
206207

207-
if event.message:
208+
if event.message and event.stage != Stage.COMPLETE:
208209
self._log(event.message)
209210

210211
def _on_log(self, event: LogEvent) -> None:

‎tests/test_protocol_standard.py‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -51,15 +51,8 @@ async def test_handshake_with_noise(self):
5151
result = await protocol.handshake(transport)
5252
assert result.success
5353

54-
@pytest.mark.asyncio
55-
async def test_handshake_timeout(self):
56-
transport = MockTransport()
57-
# Only 3x 0x20, not enough
58-
transport.enqueue_rx(b"\x20\x20\x20")
59-
60-
protocol = HiSiliconStandard()
61-
result = await protocol.handshake(transport)
62-
assert not result.success
54+
# No timeout test — handshake waits forever for user to power-cycle.
55+
# User cancels via Ctrl-C, not a timeout.
6356

6457

6558
class TestStandardFirmwareTransfer:

0 commit comments

Comments
 (0)