feat(gui): add FT601Connection class, USB interface selection in V65/V7
- Add FT601Connection in radar_protocol.py using ftd3xx library with proper setChipConfiguration re-enumeration handling (close, wait 2s, re-open) and 4-byte write alignment - Add USB Interface dropdown to V65 Tk GUI (FT2232H default, FT601 option) - Add USB Interface combo to V7 PyQt dashboard with Live/File mode toggle - Fix mock frame_start bit 7 in both FT2232H and FT601 connections - Use FPGA range data from USB packets instead of recomputing in Python - Export FT601Connection from v7/hardware.py and v7/__init__.py - Add 7 FT601Connection tests (91 total in test_GUI_V65_Tk.py)
This commit is contained in:
Jason
@@ -59,7 +59,7 @@ except (ModuleNotFoundError, ImportError):
|
||||
|
||||
# Import protocol layer (no GUI deps)
|
||||
from radar_protocol import (
|
||||
RadarProtocol, FT2232HConnection,
|
||||
RadarProtocol, FT2232HConnection, FT601Connection,
|
||||
DataRecorder, RadarAcquisition,
|
||||
RadarFrame, StatusResponse,
|
||||
NUM_RANGE_BINS, NUM_DOPPLER_BINS, WATERFALL_DEPTH,
|
||||
@@ -388,10 +388,11 @@ class RadarDashboard:
|
||||
BANDWIDTH = 500e6 # Hz — chirp bandwidth
|
||||
C = 3e8 # m/s — speed of light
|
||||
|
||||
def __init__(self, root: tk.Tk, connection: FT2232HConnection,
|
||||
def __init__(self, root: tk.Tk, mock: bool,
|
||||
recorder: DataRecorder, device_index: int = 0):
|
||||
self.root = root
|
||||
self.conn = connection
|
||||
self._mock = mock
|
||||
self.conn: FT2232HConnection | FT601Connection | None = None
|
||||
self.recorder = recorder
|
||||
self.device_index = device_index
|
||||
|
||||
@@ -485,6 +486,16 @@ class RadarDashboard:
|
||||
style="Accent.TButton")
|
||||
self.btn_connect.pack(side="right", padx=4)
|
||||
|
||||
# USB Interface selector (production FT2232H / premium FT601)
|
||||
self._usb_iface_var = tk.StringVar(value="FT2232H (Production)")
|
||||
self.cmb_usb_iface = ttk.Combobox(
|
||||
top, textvariable=self._usb_iface_var,
|
||||
values=["FT2232H (Production)", "FT601 (Premium)"],
|
||||
state="readonly", width=20,
|
||||
)
|
||||
self.cmb_usb_iface.pack(side="right", padx=4)
|
||||
ttk.Label(top, text="USB:", font=("Menlo", 10)).pack(side="right")
|
||||
|
||||
self.btn_record = ttk.Button(top, text="Record", command=self._on_record)
|
||||
self.btn_record.pack(side="right", padx=4)
|
||||
|
||||
@@ -1018,15 +1029,17 @@ class RadarDashboard:
|
||||
|
||||
# ------------------------------------------------------------ Actions
|
||||
def _on_connect(self):
|
||||
if self.conn.is_open:
|
||||
if self.conn is not None and self.conn.is_open:
|
||||
# Disconnect
|
||||
if self._acq_thread is not None:
|
||||
self._acq_thread.stop()
|
||||
self._acq_thread.join(timeout=2)
|
||||
self._acq_thread = None
|
||||
self.conn.close()
|
||||
self.conn = None
|
||||
self.lbl_status.config(text="DISCONNECTED", foreground=RED)
|
||||
self.btn_connect.config(text="Connect")
|
||||
self.cmb_usb_iface.config(state="readonly")
|
||||
log.info("Disconnected")
|
||||
return
|
||||
|
||||
@@ -1036,6 +1049,16 @@ class RadarDashboard:
|
||||
if self._replay_active:
|
||||
self._replay_stop()
|
||||
|
||||
# Create connection based on USB Interface selector
|
||||
iface = self._usb_iface_var.get()
|
||||
if "FT601" in iface:
|
||||
self.conn = FT601Connection(mock=self._mock)
|
||||
else:
|
||||
self.conn = FT2232HConnection(mock=self._mock)
|
||||
|
||||
# Disable interface selector while connecting/connected
|
||||
self.cmb_usb_iface.config(state="disabled")
|
||||
|
||||
# Open connection in a background thread to avoid blocking the GUI
|
||||
self.lbl_status.config(text="CONNECTING...", foreground=YELLOW)
|
||||
self.btn_connect.config(state="disabled")
|
||||
@@ -1062,6 +1085,8 @@ class RadarDashboard:
|
||||
else:
|
||||
self.lbl_status.config(text="CONNECT FAILED", foreground=RED)
|
||||
self.btn_connect.config(text="Connect")
|
||||
self.cmb_usb_iface.config(state="readonly")
|
||||
self.conn = None
|
||||
|
||||
def _on_record(self):
|
||||
if self.recorder.recording:
|
||||
@@ -1110,6 +1135,9 @@ class RadarDashboard:
|
||||
f"Opcode 0x{opcode:02X} is hardware-only (ignored in replay)"))
|
||||
return
|
||||
cmd = RadarProtocol.build_command(opcode, value)
|
||||
if self.conn is None:
|
||||
log.warning("No connection — command not sent")
|
||||
return
|
||||
ok = self.conn.write(cmd)
|
||||
log.info(f"CMD 0x{opcode:02X} val={value} ({'OK' if ok else 'FAIL'})")
|
||||
|
||||
@@ -1148,7 +1176,7 @@ class RadarDashboard:
|
||||
if self._replay_active or self._replay_ctrl is not None:
|
||||
self._replay_stop()
|
||||
if self._acq_thread is not None:
|
||||
if self.conn.is_open:
|
||||
if self.conn is not None and self.conn.is_open:
|
||||
self._on_connect() # disconnect
|
||||
else:
|
||||
# Connection dropped unexpectedly — just clean up the thread
|
||||
@@ -1547,17 +1575,17 @@ def main():
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.live:
|
||||
conn = FT2232HConnection(mock=False)
|
||||
mock = False
|
||||
mode_str = "LIVE"
|
||||
else:
|
||||
conn = FT2232HConnection(mock=True)
|
||||
mock = True
|
||||
mode_str = "MOCK"
|
||||
|
||||
recorder = DataRecorder()
|
||||
|
||||
root = tk.Tk()
|
||||
|
||||
dashboard = RadarDashboard(root, conn, recorder, device_index=args.device)
|
||||
dashboard = RadarDashboard(root, mock, recorder, device_index=args.device)
|
||||
|
||||
if args.record:
|
||||
filepath = os.path.join(
|
||||
@@ -1582,8 +1610,8 @@ def main():
|
||||
if dashboard._acq_thread is not None:
|
||||
dashboard._acq_thread.stop()
|
||||
dashboard._acq_thread.join(timeout=2)
|
||||
if conn.is_open:
|
||||
conn.close()
|
||||
if dashboard.conn is not None and dashboard.conn.is_open:
|
||||
dashboard.conn.close()
|
||||
if recorder.recording:
|
||||
recorder.stop()
|
||||
root.destroy()
|
||||
|
||||
@@ -6,6 +6,7 @@ Pure-logic module for USB packet parsing and command building.
|
||||
No GUI dependencies — safe to import from tests and headless scripts.
|
||||
|
||||
USB Interface: FT2232H USB 2.0 (8-bit, 50T production board) via pyftdi
|
||||
FT601 USB 3.0 (32-bit, 200T premium board) via ftd3xx
|
||||
|
||||
USB Packet Protocol (11-byte):
|
||||
TX (FPGA→Host):
|
||||
@@ -22,7 +23,7 @@ import queue
|
||||
import logging
|
||||
import contextlib
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
from typing import Any, ClassVar
|
||||
from enum import IntEnum
|
||||
|
||||
|
||||
@@ -200,7 +201,9 @@ class RadarProtocol:
|
||||
range_i = _to_signed16(struct.unpack_from(">H", raw, 3)[0])
|
||||
doppler_i = _to_signed16(struct.unpack_from(">H", raw, 5)[0])
|
||||
doppler_q = _to_signed16(struct.unpack_from(">H", raw, 7)[0])
|
||||
detection = raw[9] & 0x01
|
||||
det_byte = raw[9]
|
||||
detection = det_byte & 0x01
|
||||
frame_start = (det_byte >> 7) & 0x01
|
||||
|
||||
return {
|
||||
"range_i": range_i,
|
||||
@@ -208,6 +211,7 @@ class RadarProtocol:
|
||||
"doppler_i": doppler_i,
|
||||
"doppler_q": doppler_q,
|
||||
"detection": detection,
|
||||
"frame_start": frame_start,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
@@ -433,7 +437,191 @@ class FT2232HConnection:
|
||||
pkt += struct.pack(">h", np.clip(range_i, -32768, 32767))
|
||||
pkt += struct.pack(">h", np.clip(dop_i, -32768, 32767))
|
||||
pkt += struct.pack(">h", np.clip(dop_q, -32768, 32767))
|
||||
pkt.append(detection & 0x01)
|
||||
# Bit 7 = frame_start (sample_counter == 0), bit 0 = detection
|
||||
det_byte = (detection & 0x01) | (0x80 if idx == 0 else 0x00)
|
||||
pkt.append(det_byte)
|
||||
pkt.append(FOOTER_BYTE)
|
||||
|
||||
buf += pkt
|
||||
|
||||
self._mock_seq_idx = (start_idx + num_packets) % NUM_CELLS
|
||||
return bytes(buf)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# FT601 USB 3.0 Connection (premium board only)
|
||||
# ============================================================================
|
||||
|
||||
# Optional ftd3xx import (FTDI's proprietary driver for FT60x USB 3.0 chips).
|
||||
# pyftdi does NOT support FT601 — it only handles USB 2.0 chips (FT232H, etc.)
|
||||
try:
|
||||
import ftd3xx # type: ignore[import-untyped]
|
||||
FTD3XX_AVAILABLE = True
|
||||
_Ftd3xxError: type = ftd3xx.FTD3XXError # type: ignore[attr-defined]
|
||||
except ImportError:
|
||||
FTD3XX_AVAILABLE = False
|
||||
_Ftd3xxError = OSError # fallback for type-checking; never raised
|
||||
|
||||
|
||||
class FT601Connection:
|
||||
"""
|
||||
FT601 USB 3.0 SuperSpeed FIFO bridge — premium board only.
|
||||
|
||||
The FT601 has a 32-bit data bus and runs at 100 MHz.
|
||||
VID:PID = 0x0403:0x6030 or 0x6031 (FTDI FT60x).
|
||||
|
||||
Requires the ``ftd3xx`` library (``pip install ftd3xx`` on Windows,
|
||||
or ``libft60x`` on Linux). This is FTDI's proprietary USB 3.0 driver;
|
||||
``pyftdi`` only supports USB 2.0 and will NOT work with FT601.
|
||||
|
||||
Public contract matches FT2232HConnection so callers can swap freely.
|
||||
"""
|
||||
|
||||
VID = 0x0403
|
||||
PID_LIST: ClassVar[list[int]] = [0x6030, 0x6031]
|
||||
|
||||
def __init__(self, mock: bool = True):
|
||||
self._mock = mock
|
||||
self._dev = None
|
||||
self._lock = threading.Lock()
|
||||
self.is_open = False
|
||||
# Mock state (reuses same synthetic data pattern)
|
||||
self._mock_frame_num = 0
|
||||
self._mock_rng = np.random.RandomState(42)
|
||||
|
||||
def open(self, device_index: int = 0) -> bool:
|
||||
if self._mock:
|
||||
self.is_open = True
|
||||
log.info("FT601 mock device opened (no hardware)")
|
||||
return True
|
||||
|
||||
if not FTD3XX_AVAILABLE:
|
||||
log.error(
|
||||
"ftd3xx library required for FT601 hardware — "
|
||||
"install with: pip install ftd3xx"
|
||||
)
|
||||
return False
|
||||
|
||||
try:
|
||||
self._dev = ftd3xx.create(device_index, ftd3xx.OPEN_BY_INDEX)
|
||||
if self._dev is None:
|
||||
log.error("No FT601 device found at index %d", device_index)
|
||||
return False
|
||||
# Verify chip configuration — only reconfigure if needed.
|
||||
# setChipConfiguration triggers USB re-enumeration, which
|
||||
# invalidates the device handle and requires a re-open cycle.
|
||||
cfg = self._dev.getChipConfiguration()
|
||||
needs_reconfig = (
|
||||
cfg.FIFOMode != 0 # 245 FIFO mode
|
||||
or cfg.ChannelConfig != 0 # 1 channel, 32-bit
|
||||
or cfg.OptionalFeatureSupport != 0
|
||||
)
|
||||
if needs_reconfig:
|
||||
cfg.FIFOMode = 0
|
||||
cfg.ChannelConfig = 0
|
||||
cfg.OptionalFeatureSupport = 0
|
||||
self._dev.setChipConfiguration(cfg)
|
||||
# Device re-enumerates — close stale handle, wait, re-open
|
||||
self._dev.close()
|
||||
self._dev = None
|
||||
import time
|
||||
time.sleep(2.0) # wait for USB re-enumeration
|
||||
self._dev = ftd3xx.create(device_index, ftd3xx.OPEN_BY_INDEX)
|
||||
if self._dev is None:
|
||||
log.error("FT601 not found after reconfiguration")
|
||||
return False
|
||||
log.info("FT601 reconfigured and re-opened (index %d)", device_index)
|
||||
self.is_open = True
|
||||
log.info("FT601 device opened (index %d)", device_index)
|
||||
return True
|
||||
except (OSError, _Ftd3xxError) as e:
|
||||
log.error("FT601 open failed: %s", e)
|
||||
self._dev = None
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
if self._dev is not None:
|
||||
with contextlib.suppress(Exception):
|
||||
self._dev.close()
|
||||
self._dev = None
|
||||
self.is_open = False
|
||||
|
||||
def read(self, size: int = 4096) -> bytes | None:
|
||||
"""Read raw bytes from FT601. Returns None on error/timeout."""
|
||||
if not self.is_open:
|
||||
return None
|
||||
|
||||
if self._mock:
|
||||
return self._mock_read(size)
|
||||
|
||||
with self._lock:
|
||||
try:
|
||||
data = self._dev.readPipe(0x82, size, raw=True)
|
||||
return bytes(data) if data else None
|
||||
except (OSError, _Ftd3xxError) as e:
|
||||
log.error("FT601 read error: %s", e)
|
||||
return None
|
||||
|
||||
def write(self, data: bytes) -> bool:
|
||||
"""Write raw bytes to FT601. Data must be 4-byte aligned for 32-bit bus."""
|
||||
if not self.is_open:
|
||||
return False
|
||||
|
||||
if self._mock:
|
||||
log.info(f"FT601 mock write: {data.hex()}")
|
||||
return True
|
||||
|
||||
# Pad to 4-byte alignment (FT601 32-bit bus requirement).
|
||||
# NOTE: Radar commands are already 4 bytes, so this should be a no-op.
|
||||
remainder = len(data) % 4
|
||||
if remainder:
|
||||
data = data + b"\x00" * (4 - remainder)
|
||||
|
||||
with self._lock:
|
||||
try:
|
||||
written = self._dev.writePipe(0x02, data, raw=True)
|
||||
return written == len(data)
|
||||
except (OSError, _Ftd3xxError) as e:
|
||||
log.error("FT601 write error: %s", e)
|
||||
return False
|
||||
|
||||
def _mock_read(self, size: int) -> bytes:
|
||||
"""Generate synthetic radar packets (same pattern as FT2232H mock)."""
|
||||
time.sleep(0.05)
|
||||
self._mock_frame_num += 1
|
||||
|
||||
buf = bytearray()
|
||||
num_packets = min(NUM_CELLS, size // DATA_PACKET_SIZE)
|
||||
start_idx = getattr(self, "_mock_seq_idx", 0)
|
||||
|
||||
for n in range(num_packets):
|
||||
idx = (start_idx + n) % NUM_CELLS
|
||||
rbin = idx // NUM_DOPPLER_BINS
|
||||
dbin = idx % NUM_DOPPLER_BINS
|
||||
|
||||
range_i = int(self._mock_rng.normal(0, 100))
|
||||
range_q = int(self._mock_rng.normal(0, 100))
|
||||
if abs(rbin - 20) < 3:
|
||||
range_i += 5000
|
||||
range_q += 3000
|
||||
|
||||
dop_i = int(self._mock_rng.normal(0, 50))
|
||||
dop_q = int(self._mock_rng.normal(0, 50))
|
||||
if abs(rbin - 20) < 3 and abs(dbin - 8) < 2:
|
||||
dop_i += 8000
|
||||
dop_q += 4000
|
||||
|
||||
detection = 1 if (abs(rbin - 20) < 2 and abs(dbin - 8) < 2) else 0
|
||||
|
||||
pkt = bytearray()
|
||||
pkt.append(HEADER_BYTE)
|
||||
pkt += struct.pack(">h", np.clip(range_q, -32768, 32767))
|
||||
pkt += struct.pack(">h", np.clip(range_i, -32768, 32767))
|
||||
pkt += struct.pack(">h", np.clip(dop_i, -32768, 32767))
|
||||
pkt += struct.pack(">h", np.clip(dop_q, -32768, 32767))
|
||||
# Bit 7 = frame_start (sample_counter == 0), bit 0 = detection
|
||||
det_byte = (detection & 0x01) | (0x80 if idx == 0 else 0x00)
|
||||
pkt.append(det_byte)
|
||||
pkt.append(FOOTER_BYTE)
|
||||
|
||||
buf += pkt
|
||||
@@ -600,6 +788,12 @@ class RadarAcquisition(threading.Thread):
|
||||
if sample.get("detection", 0):
|
||||
self._frame.detections[rbin, dbin] = 1
|
||||
self._frame.detection_count += 1
|
||||
# Accumulate FPGA range profile data (matched-filter output)
|
||||
# Each sample carries the range_i/range_q for this range bin.
|
||||
# Accumulate magnitude across Doppler bins for the range profile.
|
||||
ri = int(sample.get("range_i", 0))
|
||||
rq = int(sample.get("range_q", 0))
|
||||
self._frame.range_profile[rbin] += abs(ri) + abs(rq)
|
||||
|
||||
self._sample_idx += 1
|
||||
|
||||
@@ -607,11 +801,11 @@ class RadarAcquisition(threading.Thread):
|
||||
self._finalize_frame()
|
||||
|
||||
def _finalize_frame(self):
|
||||
"""Complete frame: compute range profile, push to queue, record."""
|
||||
"""Complete frame: push to queue, record."""
|
||||
self._frame.timestamp = time.time()
|
||||
self._frame.frame_number = self._frame_num
|
||||
# Range profile = sum of magnitude across Doppler bins
|
||||
self._frame.range_profile = np.sum(self._frame.magnitude, axis=1)
|
||||
# range_profile is already accumulated from FPGA range_i/range_q
|
||||
# data in _ingest_sample(). No need to synthesize from doppler magnitude.
|
||||
|
||||
# Push to display queue (drop old if backed up)
|
||||
try:
|
||||
|
||||
@@ -16,7 +16,7 @@ import unittest
|
||||
import numpy as np
|
||||
|
||||
from radar_protocol import (
|
||||
RadarProtocol, FT2232HConnection, DataRecorder, RadarAcquisition,
|
||||
RadarProtocol, FT2232HConnection, FT601Connection, DataRecorder, RadarAcquisition,
|
||||
RadarFrame, StatusResponse, Opcode,
|
||||
HEADER_BYTE, FOOTER_BYTE, STATUS_HEADER_BYTE,
|
||||
NUM_RANGE_BINS, NUM_DOPPLER_BINS,
|
||||
@@ -312,6 +312,61 @@ class TestFT2232HConnection(unittest.TestCase):
|
||||
self.assertFalse(conn.write(b"\x00\x00\x00\x00"))
|
||||
|
||||
|
||||
class TestFT601Connection(unittest.TestCase):
|
||||
"""Test mock FT601 connection (mirrors FT2232H tests)."""
|
||||
|
||||
def test_mock_open_close(self):
|
||||
conn = FT601Connection(mock=True)
|
||||
self.assertTrue(conn.open())
|
||||
self.assertTrue(conn.is_open)
|
||||
conn.close()
|
||||
self.assertFalse(conn.is_open)
|
||||
|
||||
def test_mock_read_returns_data(self):
|
||||
conn = FT601Connection(mock=True)
|
||||
conn.open()
|
||||
data = conn.read(4096)
|
||||
self.assertIsNotNone(data)
|
||||
self.assertGreater(len(data), 0)
|
||||
conn.close()
|
||||
|
||||
def test_mock_read_contains_valid_packets(self):
|
||||
"""Mock data should contain parseable data packets."""
|
||||
conn = FT601Connection(mock=True)
|
||||
conn.open()
|
||||
raw = conn.read(4096)
|
||||
packets = RadarProtocol.find_packet_boundaries(raw)
|
||||
self.assertGreater(len(packets), 0)
|
||||
for start, end, ptype in packets:
|
||||
if ptype == "data":
|
||||
result = RadarProtocol.parse_data_packet(raw[start:end])
|
||||
self.assertIsNotNone(result)
|
||||
conn.close()
|
||||
|
||||
def test_mock_write(self):
|
||||
conn = FT601Connection(mock=True)
|
||||
conn.open()
|
||||
cmd = RadarProtocol.build_command(0x01, 1)
|
||||
self.assertTrue(conn.write(cmd))
|
||||
conn.close()
|
||||
|
||||
def test_write_pads_to_4_bytes(self):
|
||||
"""FT601 write() should pad data to 4-byte alignment."""
|
||||
conn = FT601Connection(mock=True)
|
||||
conn.open()
|
||||
# 3-byte payload should be padded internally (no error)
|
||||
self.assertTrue(conn.write(b"\x01\x02\x03"))
|
||||
conn.close()
|
||||
|
||||
def test_read_when_closed(self):
|
||||
conn = FT601Connection(mock=True)
|
||||
self.assertIsNone(conn.read())
|
||||
|
||||
def test_write_when_closed(self):
|
||||
conn = FT601Connection(mock=True)
|
||||
self.assertFalse(conn.write(b"\x00\x00\x00\x00"))
|
||||
|
||||
|
||||
class TestDataRecorder(unittest.TestCase):
|
||||
"""Test HDF5 recording (skipped if h5py not available)."""
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ from .models import (
|
||||
# Hardware interfaces — production protocol via radar_protocol.py
|
||||
from .hardware import (
|
||||
FT2232HConnection,
|
||||
FT601Connection,
|
||||
RadarProtocol,
|
||||
Opcode,
|
||||
RadarAcquisition,
|
||||
@@ -89,7 +90,7 @@ __all__ = [ # noqa: RUF022
|
||||
"USB_AVAILABLE", "FTDI_AVAILABLE", "SCIPY_AVAILABLE",
|
||||
"SKLEARN_AVAILABLE", "FILTERPY_AVAILABLE",
|
||||
# hardware — production FPGA protocol
|
||||
"FT2232HConnection", "RadarProtocol", "Opcode",
|
||||
"FT2232HConnection", "FT601Connection", "RadarProtocol", "Opcode",
|
||||
"RadarAcquisition", "RadarFrame", "StatusResponse", "DataRecorder",
|
||||
"STM32USBInterface",
|
||||
# processing
|
||||
|
||||
@@ -13,13 +13,14 @@ RadarDashboard is a QMainWindow with six tabs:
|
||||
6. Settings — Host-side DSP parameters + About section
|
||||
|
||||
Uses production radar_protocol.py for all FPGA communication:
|
||||
- FT2232HConnection for real hardware
|
||||
- FT2232HConnection for production board (FT2232H USB 2.0)
|
||||
- FT601Connection for premium board (FT601 USB 3.0) — selectable from GUI
|
||||
- Unified replay via SoftwareFPGA + ReplayEngine + ReplayWorker
|
||||
- Mock mode (FT2232HConnection(mock=True)) for development
|
||||
|
||||
The old STM32 magic-packet start flow has been removed. FPGA registers
|
||||
are controlled directly via 4-byte {opcode, addr, value_hi, value_lo}
|
||||
commands sent over FT2232H.
|
||||
commands sent over FT2232H or FT601.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -55,6 +56,7 @@ from .models import (
|
||||
)
|
||||
from .hardware import (
|
||||
FT2232HConnection,
|
||||
FT601Connection,
|
||||
RadarProtocol,
|
||||
RadarFrame,
|
||||
StatusResponse,
|
||||
@@ -142,7 +144,7 @@ class RadarDashboard(QMainWindow):
|
||||
)
|
||||
|
||||
# Hardware interfaces — production protocol
|
||||
self._connection: FT2232HConnection | None = None
|
||||
self._connection: FT2232HConnection | FT601Connection | None = None
|
||||
self._stm32 = STM32USBInterface()
|
||||
self._recorder = DataRecorder()
|
||||
|
||||
@@ -364,7 +366,7 @@ class RadarDashboard(QMainWindow):
|
||||
# Row 0: connection mode + device combos + buttons
|
||||
ctrl_layout.addWidget(QLabel("Mode:"), 0, 0)
|
||||
self._mode_combo = QComboBox()
|
||||
self._mode_combo.addItems(["Mock", "Live FT2232H", "Replay"])
|
||||
self._mode_combo.addItems(["Mock", "Live", "Replay"])
|
||||
self._mode_combo.setCurrentIndex(0)
|
||||
ctrl_layout.addWidget(self._mode_combo, 0, 1)
|
||||
|
||||
@@ -377,6 +379,13 @@ class RadarDashboard(QMainWindow):
|
||||
refresh_btn.clicked.connect(self._refresh_devices)
|
||||
ctrl_layout.addWidget(refresh_btn, 0, 4)
|
||||
|
||||
# USB Interface selector (production FT2232H / premium FT601)
|
||||
ctrl_layout.addWidget(QLabel("USB Interface:"), 0, 5)
|
||||
self._usb_iface_combo = QComboBox()
|
||||
self._usb_iface_combo.addItems(["FT2232H (Production)", "FT601 (Premium)"])
|
||||
self._usb_iface_combo.setCurrentIndex(0)
|
||||
ctrl_layout.addWidget(self._usb_iface_combo, 0, 6)
|
||||
|
||||
self._start_btn = QPushButton("Start Radar")
|
||||
self._start_btn.setStyleSheet(
|
||||
f"QPushButton {{ background-color: {DARK_SUCCESS}; color: white; font-weight: bold; }}"
|
||||
@@ -1001,7 +1010,8 @@ class RadarDashboard(QMainWindow):
|
||||
self._conn_ft2232h = self._make_status_label("FT2232H")
|
||||
self._conn_stm32 = self._make_status_label("STM32 USB")
|
||||
|
||||
conn_layout.addWidget(QLabel("FT2232H:"), 0, 0)
|
||||
self._conn_usb_label = QLabel("USB Data:")
|
||||
conn_layout.addWidget(self._conn_usb_label, 0, 0)
|
||||
conn_layout.addWidget(self._conn_ft2232h, 0, 1)
|
||||
conn_layout.addWidget(QLabel("STM32 USB:"), 1, 0)
|
||||
conn_layout.addWidget(self._conn_stm32, 1, 1)
|
||||
@@ -1167,7 +1177,7 @@ class RadarDashboard(QMainWindow):
|
||||
about_lbl = QLabel(
|
||||
"<b>AERIS-10 Radar System V7</b><br>"
|
||||
"PyQt6 Edition with Embedded Leaflet Map<br><br>"
|
||||
"<b>Data Interface:</b> FT2232H USB 2.0 (production protocol)<br>"
|
||||
"<b>Data Interface:</b> FT2232H USB 2.0 (production) / FT601 USB 3.0 (premium)<br>"
|
||||
"<b>FPGA Protocol:</b> 4-byte register commands, 0xAA/0xBB packets<br>"
|
||||
"<b>Map:</b> OpenStreetMap + Leaflet.js<br>"
|
||||
"<b>Framework:</b> PyQt6 + QWebEngine<br>"
|
||||
@@ -1224,7 +1234,7 @@ class RadarDashboard(QMainWindow):
|
||||
# =====================================================================
|
||||
|
||||
def _send_fpga_cmd(self, opcode: int, value: int):
|
||||
"""Send a 4-byte register command to the FPGA via FT2232H."""
|
||||
"""Send a 4-byte register command to the FPGA via USB (FT2232H or FT601)."""
|
||||
if self._connection is None or not self._connection.is_open:
|
||||
logger.warning(f"Cannot send 0x{opcode:02X}={value}: no connection")
|
||||
return
|
||||
@@ -1287,16 +1297,26 @@ class RadarDashboard(QMainWindow):
|
||||
|
||||
if "Mock" in mode:
|
||||
self._replay_mode = False
|
||||
self._connection = FT2232HConnection(mock=True)
|
||||
iface = self._usb_iface_combo.currentText()
|
||||
if "FT601" in iface:
|
||||
self._connection = FT601Connection(mock=True)
|
||||
else:
|
||||
self._connection = FT2232HConnection(mock=True)
|
||||
if not self._connection.open():
|
||||
QMessageBox.critical(self, "Error", "Failed to open mock connection.")
|
||||
return
|
||||
elif "Live" in mode:
|
||||
self._replay_mode = False
|
||||
self._connection = FT2232HConnection(mock=False)
|
||||
iface = self._usb_iface_combo.currentText()
|
||||
if "FT601" in iface:
|
||||
self._connection = FT601Connection(mock=False)
|
||||
iface_name = "FT601"
|
||||
else:
|
||||
self._connection = FT2232HConnection(mock=False)
|
||||
iface_name = "FT2232H"
|
||||
if not self._connection.open():
|
||||
QMessageBox.critical(self, "Error",
|
||||
"Failed to open FT2232H. Check USB connection.")
|
||||
f"Failed to open {iface_name}. Check USB connection.")
|
||||
return
|
||||
elif "Replay" in mode:
|
||||
self._replay_mode = True
|
||||
@@ -1368,6 +1388,7 @@ class RadarDashboard(QMainWindow):
|
||||
self._start_btn.setEnabled(False)
|
||||
self._stop_btn.setEnabled(True)
|
||||
self._mode_combo.setEnabled(False)
|
||||
self._usb_iface_combo.setEnabled(False)
|
||||
self._demo_btn_main.setEnabled(False)
|
||||
self._demo_btn_map.setEnabled(False)
|
||||
n_frames = self._replay_engine.total_frames
|
||||
@@ -1417,6 +1438,7 @@ class RadarDashboard(QMainWindow):
|
||||
self._start_btn.setEnabled(False)
|
||||
self._stop_btn.setEnabled(True)
|
||||
self._mode_combo.setEnabled(False)
|
||||
self._usb_iface_combo.setEnabled(False)
|
||||
self._demo_btn_main.setEnabled(False)
|
||||
self._demo_btn_map.setEnabled(False)
|
||||
self._status_label_main.setText(f"Status: Running ({mode})")
|
||||
@@ -1462,6 +1484,7 @@ class RadarDashboard(QMainWindow):
|
||||
self._start_btn.setEnabled(True)
|
||||
self._stop_btn.setEnabled(False)
|
||||
self._mode_combo.setEnabled(True)
|
||||
self._usb_iface_combo.setEnabled(True)
|
||||
self._demo_btn_main.setEnabled(True)
|
||||
self._demo_btn_map.setEnabled(True)
|
||||
self._status_label_main.setText("Status: Radar stopped")
|
||||
@@ -1954,6 +1977,12 @@ class RadarDashboard(QMainWindow):
|
||||
self._set_conn_indicator(self._conn_ft2232h, conn_open)
|
||||
self._set_conn_indicator(self._conn_stm32, self._stm32.is_open)
|
||||
|
||||
# Update USB label to reflect which interface is active
|
||||
if isinstance(self._connection, FT601Connection):
|
||||
self._conn_usb_label.setText("FT601:")
|
||||
else:
|
||||
self._conn_usb_label.setText("FT2232H:")
|
||||
|
||||
gps_count = self._gps_packet_count
|
||||
if self._gps_worker:
|
||||
gps_count = self._gps_worker.gps_count
|
||||
|
||||
@@ -25,6 +25,7 @@ if USB_AVAILABLE:
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
from radar_protocol import ( # noqa: F401 — re-exported for v7 package
|
||||
FT2232HConnection,
|
||||
FT601Connection,
|
||||
RadarProtocol,
|
||||
Opcode,
|
||||
RadarAcquisition,
|
||||
|
||||
Reference in New Issue
Block a user