NawfalMotii79
616 lines
23 KiB
Plaintext
616 lines
23 KiB
Plaintext
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
import threading
|
|
import queue
|
|
import time
|
|
import struct
|
|
import numpy as np
|
|
import matplotlib.pyplot as plt
|
|
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
|
|
from matplotlib.figure import Figure
|
|
import matplotlib.patches as patches
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Dict, List, Tuple, Optional
|
|
from scipy import signal
|
|
from sklearn.cluster import DBSCAN
|
|
from filterpy.kalman import KalmanFilter
|
|
import crcmod
|
|
import math
|
|
import webbrowser
|
|
import tempfile
|
|
import os
|
|
|
|
try:
|
|
import usb.core
|
|
import usb.util
|
|
USB_AVAILABLE = True
|
|
except ImportError:
|
|
USB_AVAILABLE = False
|
|
logging.warning("pyusb not available. USB functionality will be disabled.")
|
|
|
|
try:
|
|
from pyftdi.ftdi import Ftdi
|
|
from pyftdi.usbtools import UsbTools
|
|
from pyftdi.ftdi import FtdiError
|
|
FTDI_AVAILABLE = True
|
|
except ImportError:
|
|
FTDI_AVAILABLE = False
|
|
logging.warning("pyftdi not available. FTDI functionality will be disabled.")
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
# Dark theme colors (same as before)
|
|
DARK_BG = "#2b2b2b"
|
|
DARK_FG = "#e0e0e0"
|
|
DARK_ACCENT = "#3c3f41"
|
|
DARK_HIGHLIGHT = "#4e5254"
|
|
DARK_BORDER = "#555555"
|
|
DARK_TEXT = "#cccccc"
|
|
DARK_BUTTON = "#3c3f41"
|
|
DARK_BUTTON_HOVER = "#4e5254"
|
|
DARK_TREEVIEW = "#3c3f41"
|
|
DARK_TREEVIEW_ALT = "#404040"
|
|
|
|
@dataclass
|
|
class RadarTarget:
|
|
id: int
|
|
range: float
|
|
velocity: float
|
|
azimuth: int
|
|
elevation: int
|
|
latitude: float = 0.0
|
|
longitude: float = 0.0
|
|
snr: float = 0.0
|
|
timestamp: float = 0.0
|
|
track_id: int = -1
|
|
|
|
@dataclass
|
|
class RadarSettings:
|
|
system_frequency: float = 10e9
|
|
chirp_duration_1: float = 30e-6 # Long chirp duration
|
|
chirp_duration_2: float = 0.5e-6 # Short chirp duration
|
|
chirps_per_position: int = 32
|
|
freq_min: float = 10e6
|
|
freq_max: float = 30e6
|
|
prf1: float = 1000
|
|
prf2: float = 2000
|
|
max_distance: float = 50000
|
|
map_size: float = 50000 # Map size in meters
|
|
|
|
@dataclass
|
|
class GPSData:
|
|
latitude: float
|
|
longitude: float
|
|
altitude: float
|
|
pitch: float # Pitch angle in degrees
|
|
timestamp: float
|
|
|
|
class MapGenerator:
|
|
# [MapGenerator class remains the same as before]
|
|
pass
|
|
|
|
class FT601Interface:
|
|
"""
|
|
Interface for FT601 USB 3.0 SuperSpeed controller
|
|
"""
|
|
def __init__(self):
|
|
self.ftdi = None
|
|
self.is_open = False
|
|
self.device = None
|
|
self.ep_in = None
|
|
self.ep_out = None
|
|
|
|
# FT601 specific parameters
|
|
self.channel = 0 # Default channel
|
|
self.fifo_mode = True
|
|
self.buffer_size = 512 # FT601 optimal buffer size
|
|
|
|
def list_devices(self):
|
|
"""List available FT601 devices using pyftdi"""
|
|
if not FTDI_AVAILABLE:
|
|
logging.warning("FTDI not available - please install pyftdi")
|
|
return []
|
|
|
|
try:
|
|
devices = []
|
|
# FT601 vendor/product IDs
|
|
ft601_vid_pids = [
|
|
(0x0403, 0x6030), # FT601
|
|
(0x0403, 0x6031), # FT601Q
|
|
]
|
|
|
|
for vid, pid in ft601_vid_pids:
|
|
found_devices = usb.core.find(find_all=True, idVendor=vid, idProduct=pid)
|
|
for dev in found_devices:
|
|
try:
|
|
product = usb.util.get_string(dev, dev.iProduct) if dev.iProduct else "FT601 USB3.0"
|
|
serial = usb.util.get_string(dev, dev.iSerialNumber) if dev.iSerialNumber else "Unknown"
|
|
|
|
# Create FTDI URL for the device
|
|
url = f"ftdi://{vid:04x}:{pid:04x}:{serial}/1"
|
|
|
|
devices.append({
|
|
'description': f"{product} ({serial})",
|
|
'vendor_id': vid,
|
|
'product_id': pid,
|
|
'url': url,
|
|
'device': dev,
|
|
'serial': serial
|
|
})
|
|
except Exception as e:
|
|
devices.append({
|
|
'description': f"FT601 USB3.0 (VID:{vid:04X}, PID:{pid:04X})",
|
|
'vendor_id': vid,
|
|
'product_id': pid,
|
|
'url': f"ftdi://{vid:04x}:{pid:04x}/1",
|
|
'device': dev
|
|
})
|
|
|
|
return devices
|
|
except Exception as e:
|
|
logging.error(f"Error listing FT601 devices: {e}")
|
|
# Return mock devices for testing
|
|
return [
|
|
{'description': 'FT601 USB3.0 Device A',
|
|
'url': 'ftdi://device/1',
|
|
'vendor_id': 0x0403,
|
|
'product_id': 0x6030}
|
|
]
|
|
|
|
def open_device(self, device_url):
|
|
"""Open FT601 device using pyftdi"""
|
|
if not FTDI_AVAILABLE:
|
|
logging.error("FTDI not available - cannot open device")
|
|
return False
|
|
|
|
try:
|
|
self.ftdi = Ftdi()
|
|
|
|
# Open device with FT601 specific configuration
|
|
self.ftdi.open_from_url(device_url)
|
|
|
|
# Configure for FT601 SuperSpeed mode
|
|
# Set to 245 FIFO mode (similar to FT2232 but with 32-bit bus)
|
|
self.ftdi.set_bitmode(0xFF, Ftdi.BitMode.SYNCFF)
|
|
|
|
# Set high baud rate for USB 3.0 (500MHz / 5 = 100MHz)
|
|
self.ftdi.set_frequency(100e6) # 100 MHz clock
|
|
|
|
# Configure latency timer for optimal performance
|
|
self.ftdi.set_latency_timer(2) # 2ms latency
|
|
|
|
# Set transfer size for large packets
|
|
self.ftdi.write_data_set_chunksize(self.buffer_size)
|
|
|
|
# Purge buffers
|
|
self.ftdi.purge_buffers()
|
|
|
|
self.is_open = True
|
|
logging.info(f"FT601 device opened: {device_url}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error opening FT601 device: {e}")
|
|
return False
|
|
|
|
def open_device_direct(self, device_info):
|
|
"""Open FT601 device directly using USB (alternative method)"""
|
|
if not USB_AVAILABLE:
|
|
logging.error("USB not available - cannot open device")
|
|
return False
|
|
|
|
try:
|
|
self.device = device_info['device']
|
|
|
|
# Detach kernel driver if active
|
|
if self.device.is_kernel_driver_active(0):
|
|
self.device.detach_kernel_driver(0)
|
|
|
|
# Set configuration
|
|
self.device.set_configuration()
|
|
|
|
# Get FT601 endpoints
|
|
cfg = self.device.get_active_configuration()
|
|
intf = cfg[(0,0)]
|
|
|
|
# FT601 typically has:
|
|
# EP1 OUT (host to device)
|
|
# EP1 IN (device to host)
|
|
# EP2 OUT
|
|
# EP2 IN
|
|
|
|
# Find bulk endpoints for high-speed transfer
|
|
self.ep_out = usb.util.find_descriptor(
|
|
intf,
|
|
custom_match=lambda e:
|
|
usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT and
|
|
e.bEndpointAddress & 0xF in [1, 2] # EP1 or EP2
|
|
)
|
|
|
|
self.ep_in = usb.util.find_descriptor(
|
|
intf,
|
|
custom_match=lambda e:
|
|
usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN and
|
|
e.bEndpointAddress & 0xF in [1, 2] # EP1 or EP2
|
|
)
|
|
|
|
if self.ep_out is None or self.ep_in is None:
|
|
logging.error("Could not find FT601 endpoints")
|
|
return False
|
|
|
|
self.is_open = True
|
|
logging.info(f"FT601 device opened: {device_info['description']}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error opening FT601 device: {e}")
|
|
return False
|
|
|
|
def read_data(self, bytes_to_read=None):
|
|
"""Read data from FT601 (32-bit word aligned)"""
|
|
if not self.is_open or (self.ftdi is None and self.device is None):
|
|
return None
|
|
|
|
try:
|
|
if self.ftdi:
|
|
# Using pyftdi
|
|
# FT601 reads are 32-bit aligned
|
|
if bytes_to_read is None:
|
|
bytes_to_read = self.buffer_size
|
|
|
|
# Ensure read size is multiple of 4 bytes
|
|
bytes_to_read = ((bytes_to_read + 3) // 4) * 4
|
|
|
|
data = self.ftdi.read_data(bytes_to_read)
|
|
if data:
|
|
return bytes(data)
|
|
return None
|
|
|
|
elif self.device and self.ep_in:
|
|
# Direct USB access
|
|
if bytes_to_read is None:
|
|
bytes_to_read = 512
|
|
|
|
# FT601 maximum packet size
|
|
max_packet = 512
|
|
|
|
data = bytearray()
|
|
while len(data) < bytes_to_read:
|
|
chunk_size = min(max_packet, bytes_to_read - len(data))
|
|
try:
|
|
chunk = self.ep_in.read(chunk_size, timeout=100)
|
|
data.extend(chunk)
|
|
except usb.core.USBError as e:
|
|
if e.errno == 110: # Timeout
|
|
break
|
|
raise
|
|
|
|
return bytes(data) if data else None
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error reading from FT601: {e}")
|
|
return None
|
|
|
|
def write_data(self, data):
|
|
"""Write data to FT601 (32-bit word aligned)"""
|
|
if not self.is_open or (self.ftdi is None and self.device is None):
|
|
return False
|
|
|
|
try:
|
|
if self.ftdi:
|
|
# Using pyftdi
|
|
# Ensure data length is multiple of 4 for 32-bit alignment
|
|
if len(data) % 4 != 0:
|
|
padding = 4 - (len(data) % 4)
|
|
data += b'\x00' * padding
|
|
|
|
self.ftdi.write_data(data)
|
|
return True
|
|
|
|
elif self.device and self.ep_out:
|
|
# Direct USB access
|
|
# FT601 supports large transfers
|
|
max_packet = 512
|
|
|
|
for i in range(0, len(data), max_packet):
|
|
chunk = data[i:i + max_packet]
|
|
self.ep_out.write(chunk, timeout=100)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error writing to FT601: {e}")
|
|
return False
|
|
|
|
def configure_burst_mode(self, enable=True):
|
|
"""Configure FT601 burst mode for maximum throughput"""
|
|
if self.ftdi:
|
|
try:
|
|
# FT601 specific commands for burst mode
|
|
if enable:
|
|
# Enable burst mode
|
|
self.ftdi.set_bitmode(0xFF, Ftdi.BitMode.SYNCFF)
|
|
self.ftdi.write_data_set_chunksize(4096) # Larger chunks for burst
|
|
logging.info("FT601 burst mode enabled")
|
|
else:
|
|
# Disable burst mode
|
|
self.ftdi.set_bitmode(0xFF, Ftdi.BitMode.RESET)
|
|
logging.info("FT601 burst mode disabled")
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Error configuring burst mode: {e}")
|
|
return False
|
|
return False
|
|
|
|
def close(self):
|
|
"""Close FT601 device"""
|
|
if self.ftdi and self.is_open:
|
|
try:
|
|
self.ftdi.close()
|
|
self.is_open = False
|
|
logging.info("FT601 device closed")
|
|
except Exception as e:
|
|
logging.error(f"Error closing FT601 device: {e}")
|
|
|
|
if self.device and self.is_open:
|
|
try:
|
|
usb.util.dispose_resources(self.device)
|
|
self.is_open = False
|
|
except Exception as e:
|
|
logging.error(f"Error closing FT601 device: {e}")
|
|
|
|
# [RadarProcessor class remains the same]
|
|
class RadarProcessor:
|
|
# ... (same as before)
|
|
pass
|
|
|
|
class USBPacketParser:
|
|
# ... (same as before)
|
|
pass
|
|
|
|
class RadarPacketParser:
|
|
# ... (same as before)
|
|
pass
|
|
|
|
class RadarGUI:
|
|
def __init__(self, root):
|
|
self.root = root
|
|
self.root.title("Advanced Radar System GUI - FT601 USB 3.0")
|
|
self.root.geometry("1400x900")
|
|
|
|
# Apply dark theme
|
|
self.root.configure(bg=DARK_BG)
|
|
|
|
# Configure ttk style
|
|
self.style = ttk.Style()
|
|
self.style.theme_use('clam')
|
|
self.configure_dark_theme()
|
|
|
|
# Initialize interfaces - Replace FTDI with FT601
|
|
self.stm32_usb_interface = STM32USBInterface()
|
|
self.ft601_interface = FT601Interface() # Changed from FTDIInterface
|
|
self.radar_processor = RadarProcessor()
|
|
self.usb_packet_parser = USBPacketParser()
|
|
self.radar_packet_parser = RadarPacketParser()
|
|
self.map_generator = MapGenerator()
|
|
self.settings = RadarSettings()
|
|
|
|
# Data queues
|
|
self.radar_data_queue = queue.Queue()
|
|
self.gps_data_queue = queue.Queue()
|
|
|
|
# Thread control
|
|
self.running = False
|
|
self.radar_thread = None
|
|
self.gps_thread = None
|
|
|
|
# Counters
|
|
self.received_packets = 0
|
|
self.current_gps = GPSData(latitude=41.9028, longitude=12.4964, altitude=0, pitch=0.0, timestamp=0)
|
|
self.corrected_elevations = []
|
|
self.map_file_path = None
|
|
self.google_maps_api_key = "YOUR_GOOGLE_MAPS_API_KEY"
|
|
|
|
self.create_gui()
|
|
self.start_background_threads()
|
|
|
|
def configure_dark_theme(self):
|
|
# ... (same as before)
|
|
pass
|
|
|
|
def create_gui(self):
|
|
# ... (same as before, update device label)
|
|
pass
|
|
|
|
def setup_main_tab(self):
|
|
"""Setup the main radar display tab"""
|
|
# Control frame
|
|
control_frame = ttk.Frame(self.tab_main)
|
|
control_frame.pack(fill='x', padx=10, pady=5)
|
|
|
|
# USB Device selection
|
|
ttk.Label(control_frame, text="STM32 USB Device:").grid(row=0, column=0, padx=5)
|
|
self.stm32_usb_combo = ttk.Combobox(control_frame, state="readonly", width=40)
|
|
self.stm32_usb_combo.grid(row=0, column=1, padx=5)
|
|
|
|
# FT601 Device selection (replaces FTDI)
|
|
ttk.Label(control_frame, text="FT601 USB 3.0 Device:").grid(row=0, column=2, padx=5)
|
|
self.ft601_combo = ttk.Combobox(control_frame, state="readonly", width=40)
|
|
self.ft601_combo.grid(row=0, column=3, padx=5)
|
|
|
|
# Burst mode checkbox (new for FT601)
|
|
self.burst_mode_var = tk.BooleanVar(value=True)
|
|
ttk.Checkbutton(control_frame, text="Burst Mode",
|
|
variable=self.burst_mode_var).grid(row=0, column=4, padx=5)
|
|
|
|
ttk.Button(control_frame, text="Refresh Devices",
|
|
command=self.refresh_devices).grid(row=0, column=5, padx=5)
|
|
|
|
self.start_button = ttk.Button(control_frame, text="Start Radar",
|
|
command=self.start_radar)
|
|
self.start_button.grid(row=0, column=6, padx=5)
|
|
|
|
self.stop_button = ttk.Button(control_frame, text="Stop Radar",
|
|
command=self.stop_radar, state="disabled")
|
|
self.stop_button.grid(row=0, column=7, padx=5)
|
|
|
|
# GPS and Pitch info
|
|
self.gps_label = ttk.Label(control_frame, text="GPS: Waiting for data...")
|
|
self.gps_label.grid(row=1, column=0, columnspan=4, sticky='w', padx=5, pady=2)
|
|
|
|
# Pitch display
|
|
self.pitch_label = ttk.Label(control_frame, text="Pitch: --.--�")
|
|
self.pitch_label.grid(row=1, column=4, columnspan=2, padx=5, pady=2)
|
|
|
|
# Status info with FT601 specific info
|
|
self.status_label = ttk.Label(control_frame,
|
|
text="Status: Ready - FT601 USB 3.0")
|
|
self.status_label.grid(row=1, column=6, columnspan=2, sticky='e', padx=5, pady=2)
|
|
|
|
# [Rest of setup_main_tab remains the same]
|
|
# ...
|
|
|
|
def refresh_devices(self):
|
|
"""Refresh available USB devices"""
|
|
# STM32 USB devices
|
|
stm32_devices = self.stm32_usb_interface.list_devices()
|
|
stm32_names = [dev['description'] for dev in stm32_devices]
|
|
self.stm32_usb_combo['values'] = stm32_names
|
|
|
|
# FT601 devices (replaces FTDI)
|
|
ft601_devices = self.ft601_interface.list_devices()
|
|
ft601_names = [dev['description'] for dev in ft601_devices]
|
|
self.ft601_combo['values'] = ft601_names
|
|
|
|
if stm32_names:
|
|
self.stm32_usb_combo.current(0)
|
|
if ft601_names:
|
|
self.ft601_combo.current(0)
|
|
|
|
def start_radar(self):
|
|
"""Start radar operation with FT601"""
|
|
try:
|
|
# Open STM32 USB device
|
|
stm32_index = self.stm32_usb_combo.current()
|
|
if stm32_index == -1:
|
|
messagebox.showerror("Error", "Please select an STM32 USB device")
|
|
return
|
|
|
|
stm32_devices = self.stm32_usb_interface.list_devices()
|
|
if stm32_index >= len(stm32_devices):
|
|
messagebox.showerror("Error", "Invalid STM32 device selection")
|
|
return
|
|
|
|
if not self.stm32_usb_interface.open_device(stm32_devices[stm32_index]):
|
|
messagebox.showerror("Error", "Failed to open STM32 USB device")
|
|
return
|
|
|
|
# Open FT601 device
|
|
ft601_index = self.ft601_combo.current()
|
|
if ft601_index != -1:
|
|
ft601_devices = self.ft601_interface.list_devices()
|
|
if ft601_index < len(ft601_devices):
|
|
# Try direct USB first, fallback to pyftdi
|
|
if not self.ft601_interface.open_device_direct(ft601_devices[ft601_index]):
|
|
device_url = ft601_devices[ft601_index]['url']
|
|
if not self.ft601_interface.open_device(device_url):
|
|
logging.warning("Failed to open FT601 device, continuing without radar data")
|
|
messagebox.showwarning("Warning", "Failed to open FT601 device")
|
|
else:
|
|
# Configure burst mode if enabled
|
|
if self.burst_mode_var.get():
|
|
self.ft601_interface.configure_burst_mode(True)
|
|
else:
|
|
logging.warning("No FT601 device selected, continuing without radar data")
|
|
else:
|
|
logging.warning("No FT601 device selected, continuing without radar data")
|
|
|
|
# Send start flag to STM32
|
|
if not self.stm32_usb_interface.send_start_flag():
|
|
messagebox.showerror("Error", "Failed to send start flag to STM32")
|
|
return
|
|
|
|
# Send settings to STM32
|
|
self.apply_settings()
|
|
|
|
# Start radar operation
|
|
self.running = True
|
|
self.start_button.config(state="disabled")
|
|
self.stop_button.config(state="normal")
|
|
self.status_label.config(text="Status: Radar running - FT601 USB 3.0 active")
|
|
|
|
logging.info("Radar system started successfully with FT601 USB 3.0")
|
|
|
|
except Exception as e:
|
|
messagebox.showerror("Error", f"Failed to start radar: {e}")
|
|
logging.error(f"Start radar error: {e}")
|
|
|
|
def stop_radar(self):
|
|
"""Stop radar operation"""
|
|
self.running = False
|
|
self.start_button.config(state="normal")
|
|
self.stop_button.config(state="disabled")
|
|
self.status_label.config(text="Status: Radar stopped")
|
|
|
|
self.stm32_usb_interface.close()
|
|
self.ft601_interface.close()
|
|
|
|
logging.info("Radar system stopped")
|
|
|
|
def process_radar_data(self):
|
|
"""Process incoming radar data from FT601"""
|
|
buffer = bytearray()
|
|
while True:
|
|
if self.running and self.ft601_interface.is_open:
|
|
try:
|
|
# Read from FT601 (supports larger transfers)
|
|
data = self.ft601_interface.read_data(4096)
|
|
if data:
|
|
buffer.extend(data)
|
|
|
|
# Process packets (32-bit aligned)
|
|
while len(buffer) >= 8: # Minimum packet size
|
|
# Try to find valid packet
|
|
packet = self.radar_packet_parser.parse_packet(bytes(buffer))
|
|
if packet:
|
|
self.process_radar_packet(packet)
|
|
# Remove processed packet from buffer
|
|
packet_length = self.get_packet_length(packet)
|
|
if packet_length > 0:
|
|
buffer = buffer[packet_length:]
|
|
self.received_packets += 1
|
|
else:
|
|
# No valid packet found, shift buffer
|
|
if len(buffer) > 4:
|
|
buffer = buffer[1:]
|
|
else:
|
|
break
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error processing radar data: {e}")
|
|
time.sleep(0.1)
|
|
else:
|
|
time.sleep(0.1)
|
|
|
|
def get_packet_length(self, packet):
|
|
"""Calculate packet length including header and footer"""
|
|
# This should match your packet structure
|
|
return 64 # Example: 64-byte packets
|
|
|
|
# [Rest of the methods remain the same]
|
|
# ...
|
|
|
|
def main():
|
|
"""Main application entry point"""
|
|
try:
|
|
root = tk.Tk()
|
|
app = RadarGUI(root)
|
|
root.mainloop()
|
|
except Exception as e:
|
|
logging.error(f"Application error: {e}")
|
|
messagebox.showerror("Fatal Error", f"Application failed to start: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |