Add Device, Stream, and Config screens to TUI (F6-F8)

Expand from 5 to 8 mode screens. F6 Device provides firmware
management, EEPROM flash with full safety state machine (C2
validation, auto-backup, 3s countdown, page write, byte verify),
and diagnostics (boot test, I2C scan, register dump). F7 Stream
does live TS capture with PID distribution and PAT/PMT tree.
F8 Config manages LNB power, DiSEqC switching, and modulation/FEC.

Foundation: 12 new SkyWalker1 methods (device info, FX2 RAM,
EEPROM I2C, diagnostics), matching DemoDevice synthetics with
realistic C2 image and TS packets, 20 bridge wrappers (RLock).
This commit is contained in:
Ryan Malloy 2026-02-14 16:08:58 -07:00
parent 5d9dfa7794
commit 567bf4d9e0
13 changed files with 3304 additions and 5 deletions

View file

@ -480,6 +480,123 @@ class SkyWalker1:
index=count, length=count)
return bytes(data)
# -- Device info (extended) --
def get_serial_number(self) -> bytes:
"""Read 8-byte serial number from device."""
return self._vendor_in(CMD_GET_SERIAL_NUMBER, length=8)
def get_usb_speed(self) -> int:
"""Read USB connection speed. 0=unknown, 1=Full, 2=High."""
data = self._vendor_in(0x07, length=1)
return data[0]
def get_vendor_string(self) -> str:
"""Read vendor string descriptor from FX2."""
data = self._vendor_in(0x0C, length=64)
return bytes(data).rstrip(b'\x00').decode('ascii', errors='replace')
def get_product_string(self) -> str:
"""Read product string descriptor from FX2."""
data = self._vendor_in(0x0D, length=64)
return bytes(data).rstrip(b'\x00').decode('ascii', errors='replace')
# -- FX2 RAM access (standard Cypress A0 vendor request) --
def fx2_ram_read(self, addr: int, length: int) -> bytes:
"""Read FX2 internal RAM via A0 vendor request. Non-destructive."""
length = max(1, min(64, length))
data = self.dev.ctrl_transfer(
usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_IN,
0xA0, addr, 0, length, 2000
)
if self.verbose:
raw = bytes(data).hex(' ')
print(f" RAM IN addr=0x{addr:04X} len={length} -> {raw}")
return bytes(data)
def fx2_ram_write(self, addr: int, data: bytes) -> int:
"""Write FX2 internal RAM via A0 vendor request. Reverts on power cycle."""
if self.verbose:
raw = data.hex(' ')
print(f" RAM OUT addr=0x{addr:04X} len={len(data)} data={raw}")
return self.dev.ctrl_transfer(
usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_OUT,
0xA0, addr, 0, data, 2000
)
def fx2_cpu_halt(self) -> None:
"""Halt FX2 CPU by writing 1 to CPUCS register (0xE600)."""
self.fx2_ram_write(0xE600, b'\x01')
def fx2_cpu_start(self) -> None:
"""Release FX2 CPU by writing 0 to CPUCS register (0xE600)."""
self.fx2_ram_write(0xE600, b'\x00')
# -- EEPROM access (via I2C proxy commands) --
EEPROM_SLAVE = 0x51
EEPROM_PAGE_SIZE = 16
EEPROM_WRITE_CYCLE_MS = 10
def eeprom_read(self, offset: int, length: int = 64) -> bytes:
"""Read from boot EEPROM at given offset via I2C."""
return self._vendor_in(CMD_I2C_READ, value=self.EEPROM_SLAVE,
index=offset, length=length)
def eeprom_write_page(self, offset: int, data: bytes) -> int:
"""Write a page (up to 16 bytes) to EEPROM. Caller handles alignment."""
return self._vendor_out(CMD_I2C_WRITE, value=self.EEPROM_SLAVE,
index=offset, data=data)
def eeprom_read_all(self, size: int = 16384) -> bytes:
"""Read entire EEPROM contents up to size bytes."""
chunk_size = 64
result = bytearray()
for offset in range(0, size, chunk_size):
remaining = min(chunk_size, size - offset)
chunk = self.eeprom_read(offset, remaining)
result.extend(chunk)
return bytes(result)
# -- Diagnostics --
def boot_debug(self, mode: int) -> dict:
"""
Run boot diagnostic with specified mode byte.
Modes: 0x80=no-op, 0x81=GPIO init, 0x82=I2C probe,
0x83=BCM4500 reset, 0x84=FW load, 0x85=full boot.
Returns 3-byte status: {stage, result, detail}.
"""
data = self._vendor_in(CMD_BOOT_8PSK, value=mode, length=3)
return {
"stage": data[0],
"result": data[1],
"detail": data[2],
}
def i2c_bus_scan(self) -> list[int]:
"""
Scan I2C bus for responding devices.
Returns list of 7-bit slave addresses that ACK'd.
The firmware returns a 16-byte bitmap (128 bits for addresses 0-127).
"""
data = self._vendor_in(CMD_I2C_BUS_SCAN, length=16)
addresses = []
for byte_idx in range(16):
for bit_idx in range(8):
if data[byte_idx] & (1 << bit_idx):
addresses.append(byte_idx * 8 + bit_idx)
return addresses
def i2c_raw_read(self, slave: int, reg: int) -> int:
"""Read a single register from an I2C device."""
data = self._vendor_in(CMD_I2C_RAW_READ, value=slave,
index=reg, length=1)
return data[0]
# -- High-level sweep helpers --
def sweep_spectrum(self, start_mhz: float, stop_mhz: float,