reconnect issue

gistfile1.txt

#!/usr/bin/env python3
"""
BTECH UV-PRO Bluetooth Connection Test Script
==============================================

This script demonstrates a connection issue with the BTECH UV-PRO radio
where RFCOMM channel 1 is rejected until the radio is power cycled.

Issue Summary:
- RFCOMM channel 1 returns "DM (Disconnect Mode)" on first connection attempt
- After power cycling the radio (30 seconds off), channel 1 works correctly
- Channel 3 accepts connections but does NOT trigger RF transmission

This script implements automatic power cycle with retry logic.

Usage:
    sudo python3 btech_connection_test.py

Requirements:
    - Linux with BlueZ bluetooth stack
    - pyserial: pip install pyserial
    - Radio must be paired (use bluetoothctl to pair first)
"""

import subprocess
import time
import serial
from datetime import datetime

# CONFIGURATION
MAC_ADDRESS = "38:D2:00:01:06:6F"  # BTECH UV-PRO MAC address
CHANNEL = 1  # RFCOMM channel (must be 1 for RF transmission)
DEVICE_NAME = "BTECH UV-PRO"
SERIAL_PORT_UUID = "00001101-0000-1000-8000-00805f9b34fb"
DEBUG_LOG = "btech_connection_debug.log"


def debug_log(msg):
    """Write debug message to log file"""
    try:
        with open(DEBUG_LOG, 'a') as f:
            timestamp = datetime.now().isoformat()
            f.write(f"[{timestamp}] {msg}\n")
            print(f"[DEBUG] {msg}")
    except:
        print(f"[DEBUG] {msg}")


def cleanup_bluetooth(mac_address):
    """Best-effort: drop any stale BT connection before we reconnect"""
    print("Cleaning up existing Bluetooth connections...")
    try:
        result = subprocess.run(['bluetoothctl', 'disconnect', mac_address],
                      capture_output=True, timeout=5, check=False, text=True)
        debug_log(f"cleanup disconnect: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
        
        result = subprocess.run(['bluetoothctl', 'connect', mac_address, SERIAL_PORT_UUID],
                      capture_output=True, timeout=5, check=False, text=True)
        debug_log(f"cleanup connect: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
    except Exception as e:
        debug_log(f"cleanup_bluetooth exception: {e}")
        pass


def connect_to_radio():
    """Connect to BTECH UV-PRO radio using Linux rfcomm
    
    Returns:
        serial.Serial object if successful, None otherwise
    """
    print(f"\n{'='*60}")
    print(f"BTECH UV-PRO Connection Test")
    print(f"{'='*60}")
    print(f"Device: {DEVICE_NAME}")
    print(f"MAC Address: {MAC_ADDRESS}")
    print(f"RFCOMM Channel: {CHANNEL}")
    print(f"{'='*60}\n")
    
    debug_log(f"=== Starting connection attempt ===")
    
    # Try connection twice: first attempt, then after power cycle if needed
    for attempt in range(2):
        try:
            print(f"\n--- Attempt {attempt + 1}/2 ---\n")
            
            # Cleanup any old connection
            result = subprocess.run(['bluetoothctl', 'disconnect', MAC_ADDRESS],
                          capture_output=True, timeout=5, check=False, text=True)
            debug_log(f"bluetoothctl disconnect: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
            
            result = subprocess.run(['bluetoothctl', 'connect', MAC_ADDRESS, SERIAL_PORT_UUID],
                          capture_output=True, timeout=10, check=False, text=True)
            debug_log(f"bluetoothctl connect: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
            
            # Release old rfcomm
            print("Releasing any existing RFCOMM bindings...")
            result = subprocess.run(["sudo", "rfcomm", "release", "0"], check=False,
                          capture_output=True, text=True)
            debug_log(f"rfcomm release: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
            time.sleep(0.5)
            
            # Trust device
            print("Trusting device...")
            result = subprocess.run(["bluetoothctl", "trust", MAC_ADDRESS],
                          capture_output=True, timeout=2, check=False, text=True)
            debug_log(f"bluetoothctl trust: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
            
            # Bind rfcomm to channel 1
            print(f"Binding RFCOMM channel {CHANNEL}...")
            result = subprocess.run(["sudo", "rfcomm", "bind", "0", MAC_ADDRESS, str(CHANNEL)],
                          capture_output=True, text=True)
            debug_log(f"rfcomm bind channel {CHANNEL}: rc={result.returncode}, stdout={result.stdout}, stderr={result.stderr}")
            
            if result.returncode != 0:
                print(f"❌ RFCOMM bind failed with return code {result.returncode}")
                if result.stderr:
                    print(f"   Error: {result.stderr.strip()}")
                
                if attempt == 0:
                    # First attempt failed - initiate power cycle
                    print("\n" + "="*60)
                    print("⚠️  POWER CYCLE REQUIRED")
                    print("="*60)
                    print("\n🔴 Please TURN OFF the radio NOW\n")
                    
                    # 30-second countdown
                    for remaining in range(30, 0, -1):
                        print(f"⏱️  Waiting {remaining} seconds for radio to reset...", end='\r')
                        time.sleep(1)
                    
                    print("\n\n🟢 Please TURN ON the radio NOW\n")
                    print("Waiting for radio to boot (15 seconds)...")
                    time.sleep(15)
                    print("\nRetrying connection...\n")
                    continue  # Retry
                else:
                    # Second attempt also failed
                    print("\n❌ Connection failed after power cycle")
                    print("⚠️  Please check radio and try again")
                    return None
            
            # Wait for connection
            print("Waiting for Bluetooth connection to stabilize...")
            time.sleep(1.5)
            
            # Open serial port
            print("Opening serial port /dev/rfcomm0...")
            ser = serial.Serial("/dev/rfcomm0", 115200, timeout=0.1, write_timeout=2)
            ser.reset_input_buffer()
            ser.reset_output_buffer()
            time.sleep(0.5)
            
            # Verify connection with test frame
            print("Verifying connection with test frame...")
            test_frame = bytes([0xC0, 0x00, 0xC0])  # KISS frame
            ser.write(test_frame)
            ser.flush()
            time.sleep(0.1)
            
            if ser.in_waiting > 0:
                response = ser.read(ser.in_waiting)
                print(f"✓ Got response: {len(response)} bytes")
                debug_log(f"Test frame response: {response.hex()}")
                ser.reset_input_buffer()
            else:
                print("✓ No response (normal for BTECH radios)")
            
            print("\n" + "="*60)
            print("✅ CONNECTION SUCCESSFUL")
            print("="*60)
            print(f"Serial port: /dev/rfcomm0")
            print(f"Baud rate: 115200")
            print(f"Timeout: 0.1s")
            print("="*60 + "\n")
            
            debug_log(f"✓ Connection successful on attempt {attempt + 1}")
            return ser
            
        except Exception as e:
            debug_log(f"Connection attempt {attempt + 1} failed: {e}")
            print(f"❌ Exception during connection: {e}")
            
            # Clean up on error
            try:
                if 'ser' in locals() and ser:
                    ser.close()
            except:
                pass
            subprocess.run(["sudo", "rfcomm", "release", "0"],
                          check=False, capture_output=True)
            
            if attempt == 0:
                # First attempt failed - initiate power cycle
                print("\n" + "="*60)
                print("⚠️  POWER CYCLE REQUIRED")
                print("="*60)
                print("\n🔴 Please TURN OFF the radio NOW\n")
                
                # 30-second countdown
                for remaining in range(30, 0, -1):
                    print(f"⏱️  Waiting {remaining} seconds for radio to reset...", end='\r')
                    time.sleep(1)
                
                print("\n\n🟢 Please TURN ON the radio NOW\n")
                print("Waiting for radio to boot (15 seconds)...")
                time.sleep(15)
                print("\nRetrying connection...\n")
                continue  # Retry
            else:
                # Second attempt also failed
                print("\n❌ Connection failed after power cycle")
                print("⚠️  Please check radio and try again")
                return None
    
    return None


def test_connection(ser):
    """Test the connection by sending a simple KISS frame"""
    print("\nTesting connection...")
    print("Sending test KISS frame...")
    
    try:
        # Send empty KISS frame
        test_frame = bytes([0xC0, 0x00, 0xC0])
        ser.write(test_frame)
        ser.flush()
        print("✓ Test frame sent successfully")
        
        # Wait for potential response
        time.sleep(0.5)
        if ser.in_waiting > 0:
            response = ser.read(ser.in_waiting)
            print(f"✓ Received {len(response)} bytes: {response.hex()}")
        else:
            print("✓ No response (expected for empty frame)")
        
        return True
    except Exception as e:
        print(f"❌ Test failed: {e}")
        return False


def disconnect(ser):
    """Disconnect and clean up"""
    print("\nDisconnecting...")
    try:
        if ser and ser.is_open:
            # Proper cleanup sequence
            try:
                ser.flush()
                ser.reset_output_buffer()
                ser.reset_input_buffer()
                ser.dtr = False
                ser.rts = False
                time.sleep(0.2)
            except:
                pass
            ser.close()
            print("✓ Serial port closed")
    finally:
        # Release RFCOMM binding
        subprocess.run(["sudo", "rfcomm", "release", "0"],
                      check=False, capture_output=True)
        print("✓ RFCOMM binding released")


def main():
    """Main function"""
    print("\nBTECH UV-PRO Bluetooth Connection Test")
    print("="*60)
    print("\nThis script demonstrates the connection issue where:")
    print("1. RFCOMM channel 1 is rejected on first connection attempt")
    print("2. After power cycling (30 seconds off), channel 1 works")
    print("3. The script will automatically guide you through power cycle")
    print("\nPress Ctrl+C to exit at any time")
    print("="*60 + "\n")
    
    # Clear debug log
    try:
        with open(DEBUG_LOG, 'w') as f:
            f.write(f"BTECH UV-PRO Connection Test - {datetime.now().isoformat()}\n")
            f.write("="*60 + "\n")
    except:
        pass
    
    ser = None
    try:
        # Connect to radio
        ser = connect_to_radio()
        
        if ser:
            # Test the connection
            test_connection(ser)
            
            # Keep connection open
            print("\nConnection is active. Press Ctrl+C to disconnect.")
            try:
                while True:
                    time.sleep(1)
            except KeyboardInterrupt:
                print("\n\nInterrupted by user")
        else:
            print("\n❌ Failed to connect to radio")
            print("\nTroubleshooting:")
            print("1. Ensure radio is paired (use: bluetoothctl pair " + MAC_ADDRESS + ")")
            print("2. Check MAC address is correct")
            print("3. Try power cycling the radio manually")
            print(f"4. Check debug log: {DEBUG_LOG}")
            return 1
    
    except KeyboardInterrupt:
        print("\n\nInterrupted by user")
    except Exception as e:
        print(f"\n❌ Unexpected error: {e}")
        import traceback
        traceback.print_exc()
        return 1
    finally:
        if ser:
            disconnect(ser)
    
    return 0


if __name__ == "__main__":
    import sys
    sys.exit(main())