added reader script
This commit is contained in:
David H
@@ -0,0 +1,278 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
RP2040 Serial Log Reader
|
||||
Connects to RP2040 controller via serial and manages logs (read/clear)
|
||||
"""
|
||||
|
||||
import serial
|
||||
import serial.tools.list_ports
|
||||
import time
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
class RP2040LogReader:
|
||||
def __init__(self):
|
||||
self.ser = None
|
||||
self.port = None
|
||||
self.baudrate = 115200 # Common baudrate for RP2040
|
||||
|
||||
def clear_screen(self):
|
||||
"""Clear the terminal screen"""
|
||||
os.system('clear' if os.name == 'posix' else 'cls')
|
||||
|
||||
def wait_for_continue(self):
|
||||
"""Wait for user to press Enter to continue"""
|
||||
input("\nPress Enter to return to menu...")
|
||||
self.clear_screen()
|
||||
|
||||
def find_ports(self):
|
||||
"""List available serial ports"""
|
||||
ports = serial.tools.list_ports.comports()
|
||||
return ports
|
||||
|
||||
def find_rp2040(self):
|
||||
"""Search for RP2040 device in available ports"""
|
||||
ports = self.find_ports()
|
||||
for port_info in ports:
|
||||
# Check for common RP2040 identifiers
|
||||
if any(identifier in port_info.description.lower() for identifier in ['rp2040', 'pico', 'feather', 'tinyusb']):
|
||||
return port_info.device
|
||||
return None
|
||||
|
||||
def connect(self, port=None, baudrate=115200):
|
||||
"""Connect to the serial port"""
|
||||
try:
|
||||
if port is None:
|
||||
# Try to auto-detect RP2040
|
||||
print("Looking for RP2040 device...")
|
||||
|
||||
rp2040_port = self.find_rp2040()
|
||||
|
||||
if rp2040_port:
|
||||
port = rp2040_port
|
||||
print(f"Found RP2040 at {port}")
|
||||
else:
|
||||
print("RP2040 not found. Waiting for device to connect...")
|
||||
print(" (Please plug in your RP2040 device)")
|
||||
|
||||
# Loop and wait for device
|
||||
while not rp2040_port:
|
||||
time.sleep(2)
|
||||
rp2040_port = self.find_rp2040()
|
||||
if rp2040_port:
|
||||
port = rp2040_port
|
||||
print(f"RP2040 detected at {port}")
|
||||
break
|
||||
print("Still waiting for RP2040...")
|
||||
|
||||
self.ser = serial.Serial(port, baudrate, timeout=2)
|
||||
self.port = port
|
||||
time.sleep(1) # Wait for connection to stabilize
|
||||
print(f"Connected to {port} at {baudrate} baud")
|
||||
return True
|
||||
|
||||
except serial.SerialException as e:
|
||||
print(f"Failed to connect: {e}")
|
||||
return False
|
||||
|
||||
def send_command(self, command):
|
||||
"""Send command to RP2040"""
|
||||
try:
|
||||
self.ser.write(f"{command}\n".encode())
|
||||
time.sleep(0.5)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"Error sending command: {e}")
|
||||
return False
|
||||
|
||||
def read_response(self):
|
||||
"""Read response from serial port"""
|
||||
try:
|
||||
response = ""
|
||||
while self.ser.in_waiting:
|
||||
response += self.ser.read(self.ser.in_waiting).decode(errors='ignore')
|
||||
time.sleep(0.1)
|
||||
return response
|
||||
except Exception as e:
|
||||
print(f"Error reading response: {e}")
|
||||
return ""
|
||||
|
||||
def extract_log_content(self, raw_response):
|
||||
"""Extract content between markers"""
|
||||
start_marker = "=== Key Log Start ==="
|
||||
end_marker = "=== End of Log ==="
|
||||
|
||||
start_idx = raw_response.find(start_marker)
|
||||
end_idx = raw_response.find(end_marker)
|
||||
|
||||
if start_idx == -1 or end_idx == -1:
|
||||
return None
|
||||
|
||||
# Extract content between markers
|
||||
content = raw_response[start_idx + len(start_marker):end_idx].strip()
|
||||
return content
|
||||
|
||||
def cleanup_log(self, raw_log):
|
||||
"""
|
||||
Clean up log by processing special character representations:
|
||||
- [BACKSPACE]: Remove previous character
|
||||
- [TAB]: Remove the tab (skip it)
|
||||
- [SPACE]: Convert to actual space character
|
||||
- [ENTER]: Remove the enter (skip it)
|
||||
"""
|
||||
cleaned = ""
|
||||
i = 0
|
||||
|
||||
while i < len(raw_log):
|
||||
# Check for special key patterns
|
||||
if raw_log[i:i+11] == "[BACKSPACE]":
|
||||
# Delete previous character if it exists
|
||||
if cleaned:
|
||||
cleaned = cleaned[:-1]
|
||||
i += 11
|
||||
elif raw_log[i:i+5] == "[TAB]":
|
||||
# Skip tabs
|
||||
i += 5
|
||||
elif raw_log[i:i+7] == "[SPACE]":
|
||||
# Convert to actual space
|
||||
cleaned += " "
|
||||
i += 7
|
||||
elif raw_log[i:i+7] == "[ENTER]":
|
||||
# Skip enters
|
||||
i += 7
|
||||
else:
|
||||
# Regular character
|
||||
cleaned += raw_log[i]
|
||||
i += 1
|
||||
|
||||
return cleaned.strip()
|
||||
|
||||
def read_log_raw(self):
|
||||
"""Option 1: Read and print raw log"""
|
||||
print("\nReading RAW Log...")
|
||||
print("-" * 50)
|
||||
|
||||
if not self.send_command("DUMP"):
|
||||
return
|
||||
|
||||
response = self.read_response()
|
||||
|
||||
if response:
|
||||
print(response)
|
||||
else:
|
||||
print("No response received")
|
||||
|
||||
print("-" * 50)
|
||||
self.wait_for_continue()
|
||||
|
||||
def read_log_clean(self):
|
||||
"""Option 2: Read and print cleaned log"""
|
||||
print("\nReading Cleaned Log...")
|
||||
print("-" * 50)
|
||||
|
||||
if not self.send_command("DUMP"):
|
||||
return
|
||||
|
||||
response = self.read_response()
|
||||
|
||||
if response:
|
||||
# Extract log content
|
||||
log_content = self.extract_log_content(response)
|
||||
|
||||
if log_content:
|
||||
# Clean up the log
|
||||
cleaned = self.cleanup_log(log_content)
|
||||
print("Cleaned Log:")
|
||||
print(cleaned)
|
||||
else:
|
||||
print("Could not find log markers in response")
|
||||
print("Raw response:")
|
||||
print(response)
|
||||
else:
|
||||
print("No response received")
|
||||
|
||||
print("-" * 50)
|
||||
self.wait_for_continue()
|
||||
|
||||
def clear_log(self):
|
||||
"""Option 3: Clear the log"""
|
||||
print("\nClearing Log...")
|
||||
print("-" * 50)
|
||||
|
||||
confirm = input("Are you sure you want to clear the log? (yes/no): ").strip().lower()
|
||||
|
||||
if confirm not in ['yes', 'y']:
|
||||
print("Clear cancelled")
|
||||
self.wait_for_continue()
|
||||
return
|
||||
|
||||
if not self.send_command("CLEAR"):
|
||||
return
|
||||
|
||||
response = self.read_response()
|
||||
|
||||
if response:
|
||||
print("Response from controller:")
|
||||
print(response)
|
||||
else:
|
||||
print("Clear command sent (no response)")
|
||||
|
||||
print("-" * 50)
|
||||
self.wait_for_continue()
|
||||
|
||||
def disconnect(self):
|
||||
"""Close serial connection"""
|
||||
if self.ser and self.ser.is_open:
|
||||
self.ser.close()
|
||||
print("Disconnected")
|
||||
|
||||
def show_menu(self):
|
||||
"""Display main menu"""
|
||||
print("\n" + "="*50)
|
||||
print(" RP2040 Log Reader")
|
||||
print("="*50)
|
||||
print("1. Read Log (RAW)")
|
||||
print("2. Read Log (Cleaned)")
|
||||
print("3. Clear Log")
|
||||
print("4. Exit")
|
||||
print("="*50)
|
||||
|
||||
def run(self):
|
||||
"""Main program loop"""
|
||||
# Connect to serial port
|
||||
if not self.connect():
|
||||
return
|
||||
|
||||
try:
|
||||
while True:
|
||||
self.clear_screen()
|
||||
self.show_menu()
|
||||
choice = input("Select option (1-4): ").strip()
|
||||
|
||||
# Immediately clear screen after selection
|
||||
self.clear_screen()
|
||||
|
||||
if choice == "1":
|
||||
self.read_log_raw()
|
||||
elif choice == "2":
|
||||
self.read_log_clean()
|
||||
elif choice == "3":
|
||||
self.clear_log()
|
||||
elif choice == "4":
|
||||
print("\nGoodbye!")
|
||||
break
|
||||
else:
|
||||
print("Invalid choice, please try again")
|
||||
time.sleep(1)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n\nProgram interrupted")
|
||||
|
||||
finally:
|
||||
self.disconnect()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
reader = RP2040LogReader()
|
||||
reader.run()
|
||||
Reference in New Issue
Block a user