FT601を使った通信システムを設計してみよう（その２）

import ftd3xx
import sys
import numpy as np
from random import randint
import logging
import time

FT_OPEN_BY_SERIAL_NUMBER = 0x00000001
FT_OPEN_BY_DESCRIPTION = 0x00000002
FT_OPEN_BY_LOCATION = 0x00000004
FT_OPEN_BY_GUID = 0x00000008
FT_OPEN_BY_INDEX = 0x00000010


def information_fifo_mode(num):
    # The following defines refer to "FTD3XX.h"
    fifo_mode_list = ["CONFIGURATION_FIFO_MODE_245", "CONFIGURATION_FIFO_MODE_600",
                      "CONFIGURATION_MODE_COUNT"]
    return fifo_mode_list[num]


def information_channel_config(num):
    # The following defines refer to "FTD3XX.h"
    channel_config_list = ["CONFIGURATION_CHANNEL_CONFIG_4", "CONFIGURATION_CHANNEL_CONFIG_2",
                           "CONFIGURATION_CHANNEL_CONFIG_1", "CONFIGURATION_CHANNEL_CONFIG_1_OUTPIPE",
                           "CONFIGURATION_CHANNEL_CONFIG_1_INPIPE", "CONFIGURATION_CHANNEL_CONFIG_COUNT"]
    return channel_config_list[num]


def open_ft_usb_device(device_type, device_name):
    ascii_device_name = bytes(device_name, encoding="ASCII")
    for device_id in range(16):
        usb_object = ftd3xx.create(device_id, FT_OPEN_BY_INDEX)
        if sys.platform == 'win32' and usb_object.getDriverVersion() < 0x01020006:
            usb_object.close()
            return None, 'D3XX driver version is old. Please update driver!'
        if usb_object.getDeviceInfo()['Description'] != ascii_device_name:
            usb_object.close()
            continue
        return usb_object, 'Successfully opened %s USB device: %s' % (device_type, device_name)
    return None, 'open usb device failed!'


def close_ft_usb_device(usb_object):
    usb_object.close()


def get_random_data(data_length):
    data_array = [randint(0, 255) for i in range(data_length)]
    return bytes(data_array), len(data_array)


def get_incremental_data(data_length):
    # repeat = int(data_length / 256)
    data_array = []
    # for i in range(repeat):
    #     for j in range(256):
    #         data_array.append(j)
    # data_array = [j for i in range(repeat) for j in range(256)]
    # return bytes(data_array), len(data_array)
    size_of_int_data = int(data_length / 4)
    data_array = np.arange(0, size_of_int_data, dtype=np.int32)
    return bytes(data_array), len(data_array)


def get_register_test_data(data_length, index, write_data_hex):
    size_of_int_data = int(data_length / 4)
    data_array = np.zeros(size_of_int_data, dtype=np.int32)
    data_array[index] = write_data_hex
    return bytes(data_array), len(data_array)


def WritePipe(D3XX, pipe, buffer, size):
    return D3XX.writePipe(pipe, buffer, size)


def ReadPipe(D3XX, pipe, size):
    received = D3XX.readPipeEx(pipe, size, True)
    return received['bytesTransferred'], received['bytes']


def upstream_test(usb_handle, option, inv):
    send_data_length = 4096
    if inv:
        data = 0xAA
    else:
        data = 0x55

    if option == 3:
        send_data, length = get_incremental_data(send_data_length)
    else:
        send_data, length = get_register_test_data(send_data_length, 1, data)

    transferred_value = WritePipe(usb_handle, 0x02, send_data, send_data_length)
    if transferred_value != send_data_length:
        # close_ft_usb_device(usb)
        print("WritePipe failed")
        return
    print("WritePipe Successful")
    return


def downstream_test(usb_handle):
    receive_data_length = 4096
    value, receive_data = ReadPipe(usb_handle, 0x82, receive_data_length)
    if value == receive_data_length:
        f_receive = open("receive_data.dat", "wb")
        f_receive.write(receive_data)
        print("ReadPipe Successful, receive data saved")
    return


def downstream_test_3x(usb_handle, turn):
    receive_data_length = 4096
    value_list = []
    receive_data_list = []
    result = [False, False, False]
    for i in range(turn):
        value, receive_data = ReadPipe(usb_handle, 0x82, receive_data_length)
        value_list.append(value)
        receive_data_list.append(receive_data)
        filename_prefix = 'receive_data_'
        filename_number = str(i)
        filename_suffix = '.dat'
        filename = filename_prefix + filename_number + filename_suffix
        if value_list[i] == receive_data_length:
            f_receive = open(filename, "wb")
            f_receive.write(receive_data_list[i])
            result[i] = True
        else:
            print("ReadPipe Failed")
            return
    if all(result):
        print("ReadPipe Successful, receive data saved")
    else:
        print("ReadPipe Failed")
    return


def loop_back_test(usb_handle, option):
    send_data_length = 4096
    receive_data_length = 4096
    flg = False
    if option == 1:
        data_type = "random"
        send_data, length = get_random_data(send_data_length)
    else:
        data_type = "incremental"
        send_data, length = get_incremental_data(send_data_length)
    # start_write = time.perf_counter()
    transferred_value = WritePipe(usb_handle, 0x02, send_data, send_data_length)
    # end_write = time.perf_counter() - start_write
    # print(f'{end_write}秒！')
    if transferred_value != send_data_length:
        close_ft_usb_device(usb)
        print("WritePipe failed")
        return
    else:
        # start_read = time.perf_counter()
        value, receive_data = ReadPipe(usb_handle, 0x82, receive_data_length)
        # end_read = time.perf_counter() - start_read
        # print(f'{end_read}秒！')
        if value == receive_data_length:
            flg = True
        if send_data != receive_data:
            print(" ** send_data and receive_data mismatch")
            return
        else:
            print(" send_data and receive_data match, loopback test passed **")

    if flg:
        f_send = open("send_data_" + data_type + ".dat", "wb")
        f_send.write(send_data)
        f_receive = open("receive_data_" + data_type + ".dat", "wb")
        f_receive.write(receive_data)


def main(handle):
    inversion = True
    while True:
        try:
            print('***** 1: loopback_test_random')
            print('***** 2: loopback_test_incremental')
            print('***** 3: upstream_test_incremental')
            print('***** 4: upstream_test_FD0004')
            print('***** 5: downstream_test_4K')
            print('***** 6: downstream_test_4K x 3 turn')
            print('***** 0: end test')
            cmd = input('command input >> ')
        except KeyboardInterrupt:
            break

        cmd = int(cmd)
        if cmd == 0:
            close_ft_usb_device(handle)
            break
        elif cmd == 1 or cmd == 2:
            loop_back_test(handle, cmd)
        elif cmd == 3 or cmd == 4:
            upstream_test(handle, cmd, inversion)
        elif cmd == 5:
            downstream_test(handle)
        elif cmd == 6:
            downstream_test_3x(handle, 3)
        else:
            print("invalid command")

        inversion = not inversion


if __name__ == '__main__':
    usb, message = open_ft_usb_device("FT60X", "FTDI SuperSpeed-FIFO Bridge")
    print(message)
    print("VendorID : 0x%X" % usb.getChipConfiguration().VendorID)
    print("ProductID : 0x%X" % usb.getChipConfiguration().ProductID)
    print("ChannelConfig : %s" % information_channel_config(usb.getChipConfiguration().ChannelConfig))
    print("FIFOMode : %s" % information_fifo_mode(usb.getChipConfiguration().FIFOMode))
    print("--------------------------------------------------------------------------")
    print("------------------------  Loop back test    ------------------------------")
    main(usb)
    # loop_back_test(1)