test_usb_complete.py

#!/usr/bin/env python3
"""
Complete USB HID functionality test for iCharger 308 DUO

Tests ALL control capabilities:
1. Start charging
2. Modify current/voltage during charging
3. Monitor data continuously
4. Stop charging
"""

import sys
import time
from modbus_usb import ModbusUSB

# Addresses
ADDR_CHANNEL_1 = 0x0100
ADDR_CONTROL = 0x8000

# Control register offsets
CTRL_OPERATION = 0
CTRL_MEMORY = 1
CTRL_CHANNEL = 2
CTRL_ORDER_LOCK = 3
CTRL_ORDER = 4
CTRL_LIMIT_CURRENT = 5  # mA
CTRL_LIMIT_VOLTAGE = 6  # mV

# Orders
ORDER_STOP = 0
ORDER_RUN = 1
ORDER_MODIFY = 2
ORDER_LOCK = 0x55AA

# Operations
OP_CHARGE = 0
OP_STORAGE = 1
OP_DISCHARGE = 2


def read_channel_data(modbus, channel=0):
    """Read channel data, returns dict or None"""
    ch_base = ADDR_CHANNEL_1 + (channel * 0x100)

    # Read basic data (offset 0-10)
    regs1 = modbus.read_input_registers(ch_base, 11)
    if not regs1 or len(regs1) < 11:
        return None

    # Read status (offset 55-56)
    regs2 = modbus.read_input_registers(ch_base + 55, 2)
    if not regs2:
        return None

    # Parse
    current = regs1[4]
    if current >= 0x8000:
        current -= 0x10000

    int_temp = regs1[9]
    if int_temp >= 0x8000:
        int_temp -= 0x10000

    return {
        'timestamp_ms': (regs1[0] << 16) | regs1[1],
        'power_mw': (regs1[2] << 16) | regs1[3],
        'current_ma': current * 10,  # 10mA units
        'voltage_mv': regs1[6],
        'capacity_mah': (regs1[7] << 16) | regs1[8],
        'temp_c': int_temp / 10.0,
        'run_status': regs2[0],
        'run_error': regs2[1],
    }


def send_order(modbus, order):
    """Send order command with unlock"""
    if not modbus.write_multiple_registers(ADDR_CONTROL + CTRL_ORDER_LOCK, [ORDER_LOCK]):
        return False
    return modbus.write_multiple_registers(ADDR_CONTROL + CTRL_ORDER, [order])


def test_order_modify(modbus):
    """Test ORDER_MODIFY - change current during charging"""
    print("\n" + "=" * 70)
    print("TEST: ORDER_MODIFY (Change current during charging)")
    print("=" * 70)

    # Start storage charge
    print("\n1. Starting STORAGE on Channel 1...")
    if not modbus.write_multiple_registers(ADDR_CONTROL, [OP_STORAGE, 0, 0]):
        print("   FAILED to set operation")
        return False

    if not send_order(modbus, ORDER_RUN):
        print("   FAILED to send ORDER_RUN")
        return False
    print("   Started!")

    # Wait for operation to stabilize
    print("\n2. Waiting 3 seconds for operation to stabilize...")
    time.sleep(3)

    data = read_channel_data(modbus)
    if data:
        print(f"   Current: {data['current_ma']}mA, Status: {data['run_status']}")

    # Try to modify current
    print("\n3. Testing ORDER_MODIFY with new current limit (500mA)...")

    # Write new current limit
    new_current_ma = 500
    if not modbus.write_multiple_registers(ADDR_CONTROL + CTRL_LIMIT_CURRENT, [new_current_ma]):
        print(f"   FAILED to write current limit")
    else:
        print(f"   Wrote current limit: {new_current_ma}mA")

    # Send MODIFY order
    if not send_order(modbus, ORDER_MODIFY):
        print(f"   FAILED to send ORDER_MODIFY: {modbus.get_error_string()}")
    else:
        print("   ORDER_MODIFY sent!")

    # Read back
    time.sleep(1)
    regs = modbus.read_holding_registers(ADDR_CONTROL, 7)
    if regs:
        print(f"   Readback - Limit Current: {regs[5]}mA, Limit Voltage: {regs[6]}mV")

    # Stop
    print("\n4. Stopping...")
    send_order(modbus, ORDER_STOP)
    print("   Stopped")

    return True


def test_long_monitoring(modbus, duration=30):
    """Test long-term monitoring during charging"""
    print("\n" + "=" * 70)
    print(f"TEST: Long-term monitoring ({duration} seconds)")
    print("=" * 70)

    # Start storage
    print("\n1. Starting STORAGE on Channel 1...")
    if not modbus.write_multiple_registers(ADDR_CONTROL, [OP_STORAGE, 0, 0]):
        print("   FAILED to set operation")
        return False

    if not send_order(modbus, ORDER_RUN):
        print("   FAILED to start")
        return False
    print("   Started!")

    # Monitor
    print(f"\n2. Monitoring for {duration} seconds...")
    print("\nTime     Voltage     Current    Power    Capacity   Temp   Status")
    print("-" * 70)

    start = time.time()
    success = 0
    fail = 0
    last_fail_time = None

    data_log = []  # For graphing later

    while time.time() - start < duration:
        elapsed = time.time() - start

        data = read_channel_data(modbus)

        if data and data['run_status'] < 100:  # Sanity check
            success += 1
            last_fail_time = None

            data_log.append({
                'time': elapsed,
                **data
            })

            print(f"{elapsed:5.1f}s   {data['voltage_mv']/1000:7.3f}V   "
                  f"{data['current_ma']/1000:+6.3f}A   "
                  f"{data['power_mw']/1000:5.1f}W   "
                  f"{data['capacity_mah']:6d}mAh   "
                  f"{data['temp_c']:4.1f}C   {data['run_status']}")
        else:
            fail += 1
            if last_fail_time is None:
                last_fail_time = elapsed

            # Try reconnect after 3 consecutive failures
            if fail > 0 and fail % 5 == 0:
                print(f"{elapsed:5.1f}s   RECONNECTING...")
                modbus.disconnect()
                time.sleep(0.5)
                if modbus.connect():
                    print(f"{elapsed:5.1f}s   Reconnected!")
                else:
                    print(f"{elapsed:5.1f}s   Reconnect failed")
            else:
                err = modbus.get_error_string()
                print(f"{elapsed:5.1f}s   FAIL: {err if err else 'no data'}")

        time.sleep(0.5)

    print("-" * 70)

    # Stop
    print("\n3. Stopping...")
    send_order(modbus, ORDER_STOP)

    # Results
    print("\n" + "=" * 70)
    print("MONITORING RESULTS")
    print("=" * 70)
    print(f"Duration: {duration} seconds")
    print(f"Successful reads: {success}")
    print(f"Failed reads: {fail}")
    print(f"Success rate: {success/(success+fail)*100:.1f}%")

    if data_log:
        print(f"\nData range:")
        voltages = [d['voltage_mv'] for d in data_log]
        currents = [d['current_ma'] for d in data_log]
        print(f"  Voltage: {min(voltages)/1000:.3f}V - {max(voltages)/1000:.3f}V")
        print(f"  Current: {min(currents)/1000:.3f}A - {max(currents)/1000:.3f}A")
        print(f"  Capacity: {data_log[-1]['capacity_mah']}mAh accumulated")

    return success > fail


def test_all_orders(modbus):
    """Test all ORDER commands via USB HID"""
    print("\n" + "=" * 70)
    print("TEST: All ORDER commands")
    print("=" * 70)

    results = {}

    # ORDER_STOP (0)
    print("\n1. ORDER_STOP...")
    results['STOP'] = send_order(modbus, ORDER_STOP)
    print(f"   {'OK' if results['STOP'] else 'FAIL'}")

    # ORDER_RUN (1) - quick start/stop
    print("\n2. ORDER_RUN...")
    modbus.write_multiple_registers(ADDR_CONTROL, [OP_STORAGE, 0, 0])
    results['RUN'] = send_order(modbus, ORDER_RUN)
    print(f"   {'OK' if results['RUN'] else 'FAIL'}")
    time.sleep(1)
    send_order(modbus, ORDER_STOP)

    # ORDER_MODIFY (2)
    print("\n3. ORDER_MODIFY...")
    modbus.write_multiple_registers(ADDR_CONTROL, [OP_STORAGE, 0, 0])
    send_order(modbus, ORDER_RUN)
    time.sleep(2)
    modbus.write_multiple_registers(ADDR_CONTROL + CTRL_LIMIT_CURRENT, [1000])
    results['MODIFY'] = send_order(modbus, ORDER_MODIFY)
    print(f"   {'OK' if results['MODIFY'] else 'FAIL'}")
    send_order(modbus, ORDER_STOP)

    # ORDER_WRITE_SYS (3) - Skip, don't want to modify flash
    print("\n4. ORDER_WRITE_SYS - SKIPPED (flash write)")
    results['WRITE_SYS'] = 'SKIPPED'

    # ORDER_WRITE_MEM_HEAD (4) - Skip
    print("\n5. ORDER_WRITE_MEM_HEAD - SKIPPED (flash write)")
    results['WRITE_MEM_HEAD'] = 'SKIPPED'

    # ORDER_WRITE_MEM (5) - Skip
    print("\n6. ORDER_WRITE_MEM - SKIPPED (flash write)")
    results['WRITE_MEM'] = 'SKIPPED'

    # ORDER_TRANS_LOG_ON (6)
    print("\n7. ORDER_TRANS_LOG_ON...")
    results['LOG_ON'] = send_order(modbus, 6)
    print(f"   {'OK' if results['LOG_ON'] else 'FAIL'}")

    # ORDER_TRANS_LOG_OFF (7)
    print("\n8. ORDER_TRANS_LOG_OFF...")
    results['LOG_OFF'] = send_order(modbus, 7)
    print(f"   {'OK' if results['LOG_OFF'] else 'FAIL'}")

    # ORDER_MSGBOX_YES (8) - Only works if dialog visible
    print("\n9. ORDER_MSGBOX_YES...")
    results['MSGBOX_YES'] = send_order(modbus, 8)
    print(f"   {'OK' if results['MSGBOX_YES'] else 'FAIL (expected - no dialog)'}")

    # ORDER_MSGBOX_NO (9)
    print("\n10. ORDER_MSGBOX_NO...")
    results['MSGBOX_NO'] = send_order(modbus, 9)
    print(f"   {'OK' if results['MSGBOX_NO'] else 'FAIL (expected - no dialog)'}")

    print("\n" + "-" * 40)
    print("ORDER Commands Summary:")
    for cmd, result in results.items():
        status = "✅" if result == True else ("⏭️" if result == 'SKIPPED' else "❌")
        print(f"  {status} {cmd}")

    return results


def main():
    print("=" * 70)
    print("iCharger 308 DUO - Complete USB HID Functionality Test")
    print("=" * 70)

    modbus = ModbusUSB(debug=False, timeout=2.0)

    print("\nConnecting...")
    if not modbus.connect():
        print(f"FAILED: {modbus.get_error_string()}")
        return 1

    print(f"Connected: {modbus.device.get_product_string()}")

    try:
        # Test all ORDER commands
        test_all_orders(modbus)

        # Test ORDER_MODIFY in detail
        test_order_modify(modbus)

        # Long monitoring test
        print("\n" + "=" * 70)
        print("Ready for long monitoring test (30 seconds)")
        print("This will test if we can continuously read data during charging.")
        print("Press Enter to continue or Ctrl+C to skip...")
        try:
            input()
            test_long_monitoring(modbus, duration=30)
        except KeyboardInterrupt:
            print("\nSkipped")

    finally:
        # Make sure we stop any operation
        send_order(modbus, ORDER_STOP)
        modbus.disconnect()
        print("\nDisconnected.")

    return 0


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