cli.py

#!/usr/bin/env python3
"""
Interactive CLI for Mi Band 4 connector
"""

import asyncio
import sys
import argparse
import os
from datetime import datetime, timedelta
from mib4 import MiBand4, AlertType, MusicState, find_miband_devices, HeartRateData

try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    print("Note: python-dotenv not installed. Install with 'pip install python-dotenv' to use .env file")

# Global variables for configuration
MAC_ADDRESS = None
AUTH_KEY = None


def load_configuration():
    """Load configuration from .env file, auth_key.txt, or environment variables"""
    global MAC_ADDRESS, AUTH_KEY
    
    # Try to load from .env file first
    mac_from_env = os.getenv('MAC_ADDRESS')
    auth_from_env = os.getenv('AUTH_KEY')
    
    # Fallback to auth_key.txt for backward compatibility
    auth_from_file = None
    try:
        with open("auth_key.txt", "r") as f:
            key = f.read().strip()
            if len(key) == 32:  # Valid auth key length
                auth_from_file = key
            else:
                print(f"Warning: Invalid auth key length in auth_key.txt (got {len(key)}, expected 32)")
    except FileNotFoundError:
        pass
    except Exception as e:
        print(f"Warning: Could not read auth_key.txt: {e}")
    
    # Use environment variable or fallback to file
    AUTH_KEY = auth_from_env or auth_from_file
    MAC_ADDRESS = mac_from_env
    
    return MAC_ADDRESS, AUTH_KEY


def get_mac_address():
    """Get MAC address from user input with validation"""
    while True:
        mac = input("Enter Mi Band 4 MAC address (format: XX:XX:XX:XX:XX:XX): ").strip()
        
        # Validate MAC address format
        if len(mac) == 17 and mac.count(':') == 5:
            # Check if all parts are valid hex
            parts = mac.split(':')
            if all(len(part) == 2 and all(c in '0123456789ABCDEFabcdef' for c in part) for part in parts):
                return mac.upper()
        
        print("Invalid MAC address format. Please use format XX:XX:XX:XX:XX:XX")
        print("Example: A1:B2:C3:D4:E5:F6")


def parse_arguments():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(description='Mi Band 4 Interactive Controller')
    parser.add_argument('-m', '--mac', 
                       help='MAC address of the Mi Band 4 device (format: XX:XX:XX:XX:XX:XX)')
    parser.add_argument('-k', '--key', 
                       help='Auth key for the Mi Band 4 device (32 hex characters)')
    
    args = parser.parse_args()
    return args


def print_menu():
    """Display the main menu"""
    print("\n" + "=" * 60)
    print("Mi Band 4 Interactive Controller")
    print("=" * 60)
    print("1.  Discover Mi Band devices")
    print("2.  Connect and authenticate")
    print("3.  Get device information")
    print("4.  Get battery information")
    print("5.  Get/Set time")
    print("6.  Send custom notification")
    print("7.  Send message notification")
    print("8.  Send call notification")
    print("9.  Send missed call notification")
    print("10. Send mail notification")
    print("11. Set music information")
    
    # Show authentication status for restricted features
    auth_status = "✓" if AUTH_KEY else "🔒"
    print(f"12. Get steps information {auth_status}")
    print(f"13. Set alarm {auth_status}")
    print(f"14. Get heart rate once {auth_status}")
    print(f"15. Start real-time heart rate {auth_status}")
    print(f"16. Set device time {auth_status}")
    print("17. Run all tests")
    print("18. Disconnect")
    print("0.  Exit")
    print("=" * 60)
    if not AUTH_KEY:
        print("🔒 = Requires authentication (auth key needed)")


def get_user_input(prompt, default=None):
    """Get user input with optional default"""
    if default:
        user_input = input(f"{prompt} [{default}]: ").strip()
        return user_input if user_input else default
    return input(f"{prompt}: ").strip()


def get_notification_input():
    """Get notification details from user"""
    title = get_user_input("Enter title/phone number", "Test Title")
    message = get_user_input("Enter message (optional)", "")
    return title, message


def get_music_input():
    """Get music information from user"""
    print("\nMusic Information:")
    artist = get_user_input("Artist", "Unknown Artist")
    album = get_user_input("Album", "Unknown Album")
    track = get_user_input("Track", "Unknown Track")
    
    try:
        volume = int(get_user_input("Volume (0-100)", "75"))
        volume = max(0, min(100, volume))
    except ValueError:
        volume = 75
    
    try:
        position = int(get_user_input("Position in seconds", "0"))
    except ValueError:
        position = 0
        
    try:
        duration = int(get_user_input("Duration in seconds", "240"))
    except ValueError:
        duration = 240
    
    state_input = get_user_input("State (playing/paused)", "playing").lower()
    state = MusicState.PLAYING if state_input.startswith('p') else MusicState.PAUSED
    
    return state, artist, album, track, volume, position, duration


def get_alarm_input():
    """Get alarm details from user"""
    print("\nAlarm Setup:")
    try:
        hour = int(get_user_input("Hour (0-23)", "9"))
        hour = max(0, min(23, hour))
    except ValueError:
        hour = 9
        
    try:
        minute = int(get_user_input("Minute (0-59)", "0"))
        minute = max(0, min(59, minute))
    except ValueError:
        minute = 0
    
    print("\nDays (enter numbers separated by spaces):")
    print("1=Monday, 2=Tuesday, 3=Wednesday, 4=Thursday, 5=Friday, 6=Saturday, 7=Sunday")
    days_input = get_user_input("Days", "1 2 3 4 5")
    
    try:
        days = tuple(int(d.strip()) for d in days_input.split() if d.strip().isdigit())
        days = tuple(d for d in days if 1 <= d <= 7)
    except ValueError:
        days = (1, 2, 3, 4, 5)  # Weekdays
    
    enabled = get_user_input("Enabled (y/n)", "y").lower().startswith('y')
    snooze = get_user_input("Snooze (y/n)", "y").lower().startswith('y')
    
    return hour, minute, days, enabled, snooze


def check_auth_requirement(feature_name):
    """Check if authentication is required and available for a feature"""
    if not AUTH_KEY:
        print(f"\n🔒 {feature_name} requires authentication!")
        print("To use this feature, you need to:")
        print("1. Add AUTH_KEY to your .env file, or")
        print("2. Use --key flag, or") 
        print("3. Put your auth key in auth_key.txt file")
        print("\nAuth key should be 32 hex characters (e.g., 75bff8071a13603b26647d4f33bbeb26)")
        return False
    return True


class InteractiveMiBand:
    def __init__(self):
        self.band = None
        self.connected = False
        self.authenticated = False
        
    async def discover_devices(self):
        """Discover Mi Band devices"""
        print("\nDiscovering Mi Band devices...")
        try:
            devices = await find_miband_devices()
            if devices:
                print(f"Found devices: {', '.join(devices)}")
                print("You can use any of these MAC addresses in your .env file or with -m flag")
            else:
                print("No Mi Band devices found")
        except Exception as e:
            print(f"Discovery failed: {e}")
    
    async def connect_and_authenticate(self):
        """Connect and authenticate to the device"""
        if self.connected:
            print("Already connected!")
            return
            
        if not MAC_ADDRESS:
            print("No MAC address configured!")
            return
            
        print(f"\nConnecting to {MAC_ADDRESS}...")
        try:
            self.band = MiBand4(MAC_ADDRESS, AUTH_KEY, timeout=15.0)
            await self.band.connect()
            self.connected = True
            print("✓ Connected!")
            
            # Set up music callbacks
            self.band.set_music_callbacks(
                play=lambda: print("🎵 Play pressed on band"),
                pause=lambda: print("🎵 Pause pressed on band"),
                forward=lambda: print("🎵 Forward pressed on band"),
                backward=lambda: print("🎵 Back pressed on band"),
                volume_up=lambda: print("🎵 Volume Up pressed on band"),
                volume_down=lambda: print("🎵 Volume Down pressed on band"),
                focus_in=lambda: print("🎵 Music app opened on band"),
                focus_out=lambda: print("🎵 Music app closed on band")
            )
            
            # Try authentication if auth key is available
            if AUTH_KEY:
                print("Authenticating...")
                try:
                    success = await self.band.authenticate()
                    if success:
                        self.authenticated = True
                        print("✓ Authenticated! Full functionality available.")
                    else:
                        print("✗ Authentication failed! Check your auth key.")
                except Exception as e:
                    print(f"Authentication error: {e}")
            else:
                print("⚠ No auth key provided - limited functionality available")
                print("  Add AUTH_KEY to .env file for full features")
                
        except Exception as e:
            print(f"Connection failed: {e}")
    
    async def get_device_info(self):
        """Get device information"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nGetting device information...")
        try:
            info = await self.band.get_device_info()
            print(f"Firmware: {info.firmware_version}")
            print(f"Hardware: {info.hardware_version}")
            print(f"Serial: {info.serial_number}")
        except Exception as e:
            print(f"Failed to get device info: {e}")
    
    async def get_battery_info(self):
        """Get battery information"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nGetting battery information...")
        try:
            battery = await self.band.get_battery_info()
            print(f"Battery: {battery.level}%")
            print(f"Status: {battery.status}")
            if battery.last_charge:
                print(f"Last charge: {battery.last_charge}")
        except Exception as e:
            print(f"Failed to get battery info: {e}")
    
    async def manage_time(self):
        """Get or set device time"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\n1. Get current time")
        print("2. Set current time to system time (requires auth)")
        choice = get_user_input("Choice", "1")
        
        if choice == "1":
            try:
                current_time = await self.band.get_current_time()
                print(f"Device time: {current_time}")
            except Exception as e:
                print(f"Failed to get time: {e}")
        elif choice == "2":
            if not check_auth_requirement("Set device time"):
                return
            if not self.authenticated:
                print("Not authenticated! Please authenticate first.")
                return
            try:
                now = datetime.now()
                await self.band.set_current_time(now)
                print(f"✓ Time set to: {now}")
            except Exception as e:
                print(f"Failed to set time: {e}")
    
    async def send_custom_notification(self):
        """Send a custom notification with user input"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nSelect notification type:")
        print("1. Message")
        print("2. Mail")
        print("3. Call")
        print("4. Missed Call")
        
        choice = get_user_input("Choice", "1")
        type_map = {
            "1": AlertType.MESSAGE,
            "2": AlertType.MAIL,
            "3": AlertType.CALL,
            "4": AlertType.MISSED_CALL
        }
        
        if choice not in type_map:
            print("Invalid choice!")
            return
            
        alert_type = type_map[choice]
        title, message = get_notification_input()
        
        try:
            await self.band.send_notification(alert_type, title, message)
            print(f"✓ Sent {alert_type.name} notification")
        except Exception as e:
            print(f"Failed to send notification: {e}")
    
    async def send_message_notification(self):
        """Send a message notification"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nMessage Notification:")
        title, message = get_notification_input()
        
        try:
            await self.band.send_notification(AlertType.MESSAGE, title, message)
            print("✓ Message notification sent")
        except Exception as e:
            print(f"Failed to send message: {e}")
    
    async def send_call_notification(self):
        """Send a call notification"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nCall Notification:")
        phone_number = get_user_input("Phone number", "+123456789")
        caller_name = get_user_input("Caller name (optional)", "")
        
        try:
            await self.band.send_notification(AlertType.CALL, phone_number, caller_name)
            print("✓ Call notification sent")
        except Exception as e:
            print(f"Failed to send call notification: {e}")
    
    async def send_missed_call_notification(self):
        """Send a missed call notification"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nMissed Call Notification:")
        phone_number = get_user_input("Phone number", "+123456789")
        caller_name = get_user_input("Caller name (optional)", "")
        
        try:
            await self.band.send_notification(AlertType.MISSED_CALL, phone_number, caller_name)
            print("✓ Missed call notification sent")
        except Exception as e:
            print(f"Failed to send missed call notification: {e}")
    
    async def send_mail_notification(self):
        """Send a mail notification"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        print("\nMail Notification:")
        sender = get_user_input("Sender", "sender@example.com")
        subject = get_user_input("Subject", "New Email")
        
        try:
            await self.band.send_notification(AlertType.MAIL, sender, subject)
            print("✓ Mail notification sent")
        except Exception as e:
            print(f"Failed to send mail notification: {e}")
    
    async def set_music_info(self):
        """Set music information"""
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
            
        state, artist, album, track, volume, position, duration = get_music_input()
        
        try:
            await self.band.set_music_info(
                state=state,
                artist=artist,
                album=album,
                track=track,
                volume=volume,
                position=position,
                duration=duration
            )
            print("✓ Music info set. Try using music controls on the band!")
        except Exception as e:
            print(f"Failed to set music info: {e}")
    
    async def get_steps_info(self):
        """Get steps information (requires auth)"""
        if not check_auth_requirement("Steps information"):
            return
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
        if not self.authenticated:
            print("Not authenticated! Please authenticate first.")
            return
            
        print("\nGetting steps information...")
        try:
            steps = await self.band.get_steps()
            print(f"Steps: {steps.steps}")
            print(f"Distance: {steps.meters}m")
            print(f"Calories: {steps.calories}")
            print(f"Fat burned: {steps.fat_burned}")
        except Exception as e:
            print(f"Failed to get steps: {e}")
    
    async def set_alarm(self):
        """Set an alarm (requires auth)"""
        if not check_auth_requirement("Set alarm"):
            return
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
        if not self.authenticated:
            print("Not authenticated! Please authenticate first.")
            return
            
        hour, minute, days, enabled, snooze = get_alarm_input()
        
        try:
            await self.band.set_alarm(
                hour=hour,
                minute=minute,
                days=days,
                enabled=enabled,
                snooze=snooze,
                alarm_id=0
            )
            day_names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
            day_str = ", ".join(day_names[d-1] for d in days) if days else "No days"
            print(f"✓ Alarm set for {hour:02d}:{minute:02d} on {day_str}")
        except Exception as e:
            print(f"Failed to set alarm: {e}")
    
    async def get_heart_rate_once(self):
        """Get heart rate once (requires auth)"""
        if not check_auth_requirement("Heart rate monitoring"):
            return
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
        if not self.authenticated:
            print("Not authenticated! Please authenticate first.")
            return
            
        print("\nGetting heart rate...")
        try:
            heart_rate = await self.band.get_heart_rate_one_time()
            print(f"✓ Heart rate: {heart_rate} BPM")
        except Exception as e:
            print(f"Failed to get heart rate: {e}")
    
    async def start_realtime_heart_rate(self):
        """Start real-time heart rate monitoring (requires auth)"""
        if not check_auth_requirement("Real-time heart rate monitoring"):
            return
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
        if not self.authenticated:
            print("Not authenticated! Please authenticate first.")
            return
            
        print("\nStarting real-time heart rate monitoring...")
        print("Press Enter to stop monitoring")
        
        heart_rates = []
        monitoring = True
        
        def heart_rate_callback(data: HeartRateData):
            heart_rates.append(data.bpm)
            print(f"📈 Heart rate: {data.bpm} BPM at {data.timestamp.strftime('%H:%M:%S')}")
        
        try:
            # Start monitoring in a task
            monitor_task = asyncio.create_task(
                self.band.start_heart_rate_realtime(heart_rate_callback)
            )
            
            # Wait for user input to stop
            def stop_monitoring():
                nonlocal monitoring
                monitoring = False
            
            # Simple way to wait for Enter key
            import threading
            
            def wait_for_input():
                input()  # Wait for Enter
                stop_monitoring()
            
            input_thread = threading.Thread(target=wait_for_input)
            input_thread.daemon = True
            input_thread.start()
            
            # Keep monitoring until stopped
            while monitoring and monitor_task and not monitor_task.done():
                await asyncio.sleep(1)
            
            # Stop monitoring
            await self.band.stop_heart_rate_realtime()
            monitor_task.cancel()
            
            try:
                await monitor_task
            except asyncio.CancelledError:
                pass
            
            if heart_rates:
                avg_hr = sum(heart_rates) / len(heart_rates)
                print(f"\n✓ Monitoring stopped. Collected {len(heart_rates)} readings, average: {avg_hr:.1f} BPM")
            else:
                print("\nNo heart rate data received")
                
        except Exception as e:
            print(f"Failed real-time heart rate monitoring: {e}")
    
    async def set_device_time(self):
        """Set device time (requires auth)"""
        if not check_auth_requirement("Set device time"):
            return
        if not self.connected:
            print("Not connected! Please connect first (option 2).")
            return
        if not self.authenticated:
            print("Not authenticated! Please authenticate first.")
            return
            
        print("\nSetting device time to current system time...")
        try:
            now = datetime.now()
            await self.band.set_current_time(now)
            print(f"✓ Time set to: {now}")
            
            # Verify
            await asyncio.sleep(1)
            device_time = await self.band.get_current_time()
            print(f"Verified device time: {device_time}")
        except Exception as e:
            print(f"Failed to set time: {e}")
    
    async def run_all_tests(self):
        """Run all available tests"""
        print("\nRunning all available tests...")
        
        if not self.connected:
            await self.connect_and_authenticate()
        
        if not self.connected:
            print("Cannot run tests without connection!")
            return
        
        # Basic tests (no auth required)
        await self.get_device_info()
        await self.get_battery_info()
        
        # Send test notifications
        print("\nSending test notifications...")
        test_notifications = [
            (AlertType.MESSAGE, "Test Message", "Hello from Python!"),
            (AlertType.MAIL, "test@example.com", "You have new mail"),
            (AlertType.CALL, "+123456789", "Test Caller"),
            (AlertType.MISSED_CALL, "+987654321", "Missed Call")
        ]
        
        for alert_type, title, message in test_notifications:
            try:
                await self.band.send_notification(alert_type, title, message)
                print(f"✓ Sent {alert_type.name}")
                await asyncio.sleep(2)
            except Exception as e:
                print(f"✗ Failed to send {alert_type.name}: {e}")
        
        # Set test music info
        try:
            await self.band.set_music_info(
                state=MusicState.PLAYING,
                artist="Test Artist",
                album="Test Album",
                track="Test Track",
                volume=75
            )
            print("✓ Music info set")
        except Exception as e:
            print(f"✗ Failed to set music info: {e}")
        
        # Authenticated tests (only if auth is available)
        if AUTH_KEY and self.authenticated:
            print("\nRunning authenticated tests...")
            await self.get_steps_info()
            await self.get_heart_rate_once()
            
            # Set a test alarm for 2 minutes from now
            try:
                alarm_time = datetime.now() + timedelta(minutes=2)
                await self.band.set_alarm(
                    hour=alarm_time.hour,
                    minute=alarm_time.minute,
                    days=(alarm_time.isoweekday(),),
                    enabled=True,
                    snooze=True,
                    alarm_id=0
                )
                print(f"✓ Test alarm set for {alarm_time.strftime('%H:%M')}")
            except Exception as e:
                print(f"✗ Failed to set alarm: {e}")
        elif not AUTH_KEY:
            print("\nSkipping authenticated tests (no auth key configured)")
        elif not self.authenticated:
            print("\nSkipping authenticated tests (authentication failed)")
        
        print("\nAll available tests completed!")
    
    async def disconnect(self):
        """Disconnect from the device"""
        if not self.connected:
            print("Not connected!")
            return
            
        print("\nDisconnecting...")
        try:
            await self.band.disconnect()
            self.connected = False
            self.authenticated = False
            print("✓ Disconnected!")
        except Exception as e:
            print(f"Disconnect error: {e}")


async def main():
    """Main interactive loop"""
    global MAC_ADDRESS, AUTH_KEY
    
    # Parse command line arguments
    args = parse_arguments()
    
    # Load configuration from .env and other sources
    load_configuration()
    
    # Override with command line arguments if provided
    if args.mac:
        # Validate provided MAC address
        mac = args.mac.strip().upper()
        if len(mac) == 17 and mac.count(':') == 5:
            parts = mac.split(':')
            if all(len(part) == 2 and all(c in '0123456789ABCDEF' for c in part) for part in parts):
                MAC_ADDRESS = mac
            else:
                print(f"Error: Invalid MAC address format: {args.mac}")
                print("Please use format XX:XX:XX:XX:XX:XX")
                sys.exit(1)
        else:
            print(f"Error: Invalid MAC address format: {args.mac}")
            print("Please use format XX:XX:XX:XX:XX:XX")
            sys.exit(1)
    
    if args.key:
        # Validate provided auth key
        key = args.key.strip()
        if len(key) == 32 and all(c in '0123456789ABCDEFabcdef' for c in key):
            AUTH_KEY = key
        else:
            print(f"Error: Invalid auth key format. Expected 32 hex characters, got {len(key)}")
            sys.exit(1)
    
    # Prompt for MAC address if not configured anywhere
    if not MAC_ADDRESS:
        print("MAC address not configured in .env file or command line.")
        MAC_ADDRESS = get_mac_address()
    
    controller = InteractiveMiBand()
    
    print("Welcome to Mi Band 4 Interactive Controller!")
    print(f"Device MAC: {MAC_ADDRESS}")
    print(f"Auth key: {'✓ Configured' if AUTH_KEY else '✗ Not configured'}")
    
    if not AUTH_KEY:
        print("\n⚠ Authentication not configured!")
        print("To enable full functionality:")
        print("  1. Add AUTH_KEY=your_key_here to .env file, or")
        print("  2. Use --key command line flag, or")
        print("  3. Put your key in auth_key.txt file")
        print("  Auth key format: 32 hex characters (0-9, a-f)")
    
    menu_actions = {
        "1": controller.discover_devices,
        "2": controller.connect_and_authenticate,
        "3": controller.get_device_info,
        "4": controller.get_battery_info,
        "5": controller.manage_time,
        "6": controller.send_custom_notification,
        "7": controller.send_message_notification,
        "8": controller.send_call_notification,
        "9": controller.send_missed_call_notification,
        "10": controller.send_mail_notification,
        "11": controller.set_music_info,
        "12": controller.get_steps_info,
        "13": controller.set_alarm,
        "14": controller.get_heart_rate_once,
        "15": controller.start_realtime_heart_rate,
        "16": controller.set_device_time,
        "17": controller.run_all_tests,
        "18": controller.disconnect,
    }
    
    try:
        while True:
            print_menu()
            choice = input("\nEnter your choice: ").strip()
            
            if choice == "0":
                if controller.connected:
                    await controller.disconnect()
                print("Goodbye!")
                break
            elif choice in menu_actions:
                try:
                    await menu_actions[choice]()
                except Exception as e:
                    print(f"Error: {e}")
                
                input("\nPress Enter to continue...")
            else:
                print("Invalid choice! Please try again.")
                
    except KeyboardInterrupt:
        print("\nExiting...")
        if controller.connected:
            await controller.disconnect()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nExited by user")
    except Exception as e:
        print(f"Application error: {e}")