#!/usr/bin/env python3 """ ############################################################################## # # # icestim stimulation control script # # # # Requirements: # # - python3 # # - python packages: # # - argparse, icecream, termcolor # # # # In Debian GNU/Linux systems, python and the pip package manager (useful to # # install the required python packages) can be installed by issuing the # # following command: # # - sudo apt install python3 python3-pipl # # # # Using the pip package manager, the required python packages can be # # installed issuing the following command: # # - pip3 install argparse icecream termcolor # # # # Usage: to get usage information, just run: # # - python3 stimulate.py -h # # # # Author: # # - Hipólito Guzmán-Miranda (hguzman@us.es) # # # ############################################################################## """ # Some ic() and print() statements have been commented out, uncomment them for # extra visibility of internal variables # Import third-party libraries import argparse import serial import sys import csv import time from icecream import ic from termcolor import colored # Configure the argument parser parser = argparse.ArgumentParser(description='Connect to a configured FPGA device and stimulate.', formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('input0', help='Input stimuli file for channel 0 (csv format, only one column)') parser.add_argument('input1', help='Input stimuli file for channel 1 (csv format, only one column)') parser.add_argument('--freq', default=60, help='Stimulation frequency in KHz') parser.add_argument('--looped', default=False, action='store_true', help='If set, the stimuli will continuously repeat. If unset, it will only run once. (default: %(default)s)') parser.add_argument('--noninteractive', default=False, action='store_true', help='If set, do not prompt the user to press keys and stop after one repeat of the stimuli vectors. Ignores --looped, setting it to false. (default: %(default)s)') # Get arguments from command-line args = parser.parse_args() ic(args) ic(args.freq) ic(args.looped) ic(args.noninteractive) # Set parameters for serial port BAUDRATE = 1000000 #9600 DEVFILE = '/dev/ttyUSB1' ic(BAUDRATE) ic(DEVFILE) # Define a dictionary with all possible commands # TODO : this must always reflect the contents of commands.vhd CMD = { "CMD_ISDATA0" : 0x0, "CMD_ISDATA1" : 0x1, "CMD_SET_SIZE_LO" : 0x2, "CMD_SET_SIZE_HI" : 0x3, "CMD_SET_FREQ" : 0x4, "CMD_SET_LOOPED" : 0x5, "CMD_START" : 0x6, "CMD_PAUSE" : 0x7, "CMD_REWIND" : 0x8, "CMD_WR_REWIND" : 0x9, "CMD_UNDEFINED1" : 0xA, "CMD_DBG_ECHO" : 0xB, "CMD_DBG_READ" : 0xC, "CMD_DBG_DUMP0" : 0xD, "CMD_DBG_DUMP1" : 0xE, "CMD_NOP" : 0xF } #ic(CMD) # Define a dictionary with all possible values # TODO : this must always reflect the contents of commands.vhd VAL = { "VAL_SIZE_LO" : 0x0, "VAL_SIZE_HI" : 0x1, "VAL_FREQ" : 0x2, "VAL_CYCLE_COUNT" : 0x3, "VAL_RUNNING" : 0x4, "VAL_LOOPED" : 0x5, "VAL_RD_PTR_LO" : 0x6, "VAL_RD_PTR_HI" : 0x7, "VAL_WR_PTR0_LO" : 0x8, "VAL_WR_PTR0_HI" : 0x9, "VAL_WR_PTR1_LO" : 0xA, "VAL_WR_PTR1_HI" : 0xB, "VAL_UNUSED0" : 0xC, "VAL_UNUSED1" : 0xD, "VAL_UNUSED2" : 0xE, "VAL_UNUSED3" : 0xF } #ic(CMD) # Define a list with all possible frequencies # TODO : this list must always match the list in stimulator.vhd ALLOWED_FREQ = [30, 50, 60] #ic(FREQ) errorstring = f'ERROR: frequency {args.freq} is not one of the possible values {ALLOWED_FREQ}' assert int(args.freq) in ALLOWED_FREQ, errorstring # Function to send a 16-bit command to the serial port # LSB is sent first, MSB is sent second def send_cmd(dev, cmd): #print("send_cmd:") #ic(hex(cmd)) cmd = cmd & 0xFFFF lsb = cmd & 0xFF msb = cmd >> 8 #ic(hex(lsb)) dev.write(lsb.to_bytes(1, 'big')) #ic(hex(msb)) dev.write(msb.to_bytes(1, 'big')) # Function to receive a 16-bit response from the serial port # LSB is received first, MSB is received second def recv_res(dev): #print('waiting for lsb...') lsb = dev.read() #ic(lsb) #print('waiting for msb...') msb = dev.read() #ic(msb) retval = (int.from_bytes(msb, 'big') << 8) + int.from_bytes(lsb, 'big') #ic(hex(retval)) return retval # Open the serial device dev = serial.Serial(DEVFILE, BAUDRATE) #ic(dev) # Parse and sanitize input for ch0 # Samples are checked in the [-2048, 2047] but are afterwards converted to the # range [0, 4095] since the DAC receives inputs as unsigned decimals ch0 = [] with open(args.input0) as f: reader = csv.reader(f) for row in reader: assert len(row) == 1, f'ERROR: row {row} in {args.input0} containes {len(row)} values' val = int(row[0]) assert (val >= -2048) and (val <= 2047), f'ERROR: value {val} in {args.input0} is out of 12-bit range [-2048, 2047]' ch0.append(val + 2048) #ic(ch0) # Parse and sanitize input for ch1 ch1 = [] with open(args.input1) as f: reader = csv.reader(f) for row in reader: assert len(row) == 1, f'ERROR: row {row} in {args.input1} containes {len(row)} values' val = int(row[0]) assert (val >= -2048) and (val <= 2047), f'ERROR: value {val} in {args.input1} is out of 12-bit range [-2048, 2047]' ch1.append(val + 2048) #ic(ch1) # Assert that both channels contain the same number of samples assert len(ch0) == len(ch1), f'ERROR: {args.input0} and {args.input1} do not contain the same number of values (they contain {len(ch0)} and {len(ch1)} values respectively)' size = len(ch0) print(f'Found stimuli of length {size}') ic(size) # Assert stimuli size is not too big. The first assertion is due to the ice40 # limitations (maybe solvable by changing FPGA family/device), the second # assertion is due to our design (to solve, the architecture and command set # should be reconsidered and modified) assert size <= 2**15, f'ERROR: stimuli length {size} is bigger than what fits in the two icebreaker SPRAMs ({2**15} samples)' assert size <= 2**24, f'ERROR: stimuli length {size} is bigger than what can be expressed with 24 bits ({2**24} samples)' # Determine SIZE_LO and SIZE_HI SIZE_LO = size & 0xFFF SIZE_HI = size >> 12 ic(SIZE_LO) ic(SIZE_HI) # Determine FREQ FREQ = ALLOWED_FREQ.index(int(args.freq)) ic(FREQ) # Determine LOOPED LOOPED = int(args.looped) # args.noninteractive disables LOOPED, warn the user if they had both if args.noninteractive == True and LOOPED == 1 : print(colored("WARNING: ", 'yellow'), end='') print("args.noninteractive is True, disabling LOOPED") LOOPED = 0 ic(LOOPED) # Send an echo command to check that the FPGA is correctly configured print("Sending an echo command:") print("(If we do not get any answer, make sure the FPGA is configured with") print("the correct bitstream (just execute the command 'make prog_top'))") command = ((CMD["CMD_DBG_ECHO"] << 12) + 0x123) send_cmd(dev, command) print(f'Sent CMD_DBG_ECHO, VALUE = 0x123. ', end='') res = recv_res(dev) print(f'Response: {hex(res)} ', end='') if res == 0xB123 : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) # Rewind read and write pointers print("Rewinding read and write pointers:") # Send CMD_REWIND print('Sending CMD_REWIND') command = (CMD["CMD_REWIND"] << 12) + SIZE_LO send_cmd(dev, command) # Send CMD_WR_REWIND print('Sending CMD_WR_REWIND') command = (CMD["CMD_WR_REWIND"] << 12) + SIZE_LO send_cmd(dev, command) # Send the stimuli configuration to the FPGA device print("Sending the configuration to the FPGA device:") # Send CMD_SET_SIZE_LO print('Sending CMD_SET_SIZE_LO') command = (CMD["CMD_SET_SIZE_LO"] << 12) + SIZE_LO send_cmd(dev, command) # Send CMD_SET_SIZE_HI print('Sending CMD_SET_SIZE_HI') command = (CMD["CMD_SET_SIZE_HI"] << 12) + SIZE_HI send_cmd(dev, command) # Send CMD_SET_FREQ print('Sending CMD_SET_FREQ') command = (CMD["CMD_SET_FREQ"] << 12) + FREQ send_cmd(dev, command) # Send CMD_SET_LOOPED print('Sending CMD_SET_LOOPED') command = (CMD["CMD_SET_LOOPED"] << 12) + LOOPED send_cmd(dev, command) # Use CMD_DBG_READ to verify the design was correctly configured print("Verifying uploaded configuration...") command = ((CMD["CMD_DBG_READ"] << 12) + VAL["VAL_SIZE_LO"]) send_cmd(dev, command) print(f'Sent CMD_DBG_READ, VAL_SIZE_LO. ', end='') res = recv_res(dev) print(f'Response: {hex(res)} ', end='') if res == SIZE_LO : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) command = ((CMD["CMD_DBG_READ"] << 12) + VAL["VAL_SIZE_HI"]) send_cmd(dev, command) print(f'Sent CMD_DBG_READ, VAL_SIZE_HI. ', end='') res = recv_res(dev) print(f'Response: {hex(res)} ', end='') if res == SIZE_HI : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) command = ((CMD["CMD_DBG_READ"] << 12) + VAL["VAL_FREQ"]) send_cmd(dev, command) res = recv_res(dev) print(f'Sent CMD_DBG_READ, VAL_FREQ. Response: {hex(res)} ', end='') if res == FREQ : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) command = ((CMD["CMD_DBG_READ"] << 12) + VAL["VAL_LOOPED"]) send_cmd(dev, command) res = recv_res(dev) print(f'Sent CMD_DBG_READ, VAL_LOOPED. Response: {hex(res)} ', end='') if res == LOOPED : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) # If we have reached this point, stimuli configuration has been correctly # transferred to the FPGA, so we can send the stimuli samples to the device # Send data for channel 0 print('Sending stimuli for channel 0') for i in range(size): command = (CMD["CMD_ISDATA0"] << 12) + ch0[i] send_cmd(dev, command) # Send data for channel 1 print('Sending stimuli for channel 1') for i in range(size): command = (CMD["CMD_ISDATA1"] << 12) + ch1[i] send_cmd(dev, command) # Use CMD_DBG_DUMP0 and CMD_DBG_DUMP1 to verify the stimuli were correcly # loaded. For each channel, we read all samples from the FPGA and store them # inside a different variable in this script, which will be compared with the # original values sent to the FPGA print("Verifying uploaded stimuli...") command = CMD["CMD_DBG_DUMP0"] << 12 send_cmd(dev, command) print(f'Sent CMD_DBG_DUMP0. Response: ', end='') ch0_dump = [] for i in range(size): res = recv_res(dev) ch0_dump.append(res) #print(hex(res), end=',') #print() if ch0_dump == ch0 : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) command = CMD["CMD_DBG_DUMP1"] << 12 send_cmd(dev, command) print(f'Sent CMD_DBG_DUMP1. Response: ', end='') ch1_dump = [] for i in range(size): res = recv_res(dev) ch1_dump.append(res) #print(hex(res), end=',') #print() if ch1_dump == ch1 : print(colored("OK", 'green')) else : sys.exit(colored("ERROR", 'red')) # Begin the stimulation if args.noninteractive == False : input("Press any key to send CMD_START, or assert the 'start' input (button_center), or press Ctrl+C to exit...") else : print("Running non-interactively, sending CMD_START...") command = CMD["CMD_START"] << 12 send_cmd(dev, command) print(f'Sent CMD_START.', end='') if LOOPED : print(' Vectors are looped, stimulation is continuous.') else : print() # If running interactively, prompt the user to stop the script. # If not, wait until all the samples have been sent. if args.noninteractive == False : input('Press any key to send CMD_PAUSE (if still running), or press Ctrl+C to exit...') else : print('Running non-interactively, will wait size * sample_time and then try to stop stimulation...') sample_time = 1.0/(int(args.freq)*1000) print(f' size: {size} (samples), freq: {args.freq} (Hz), sample_time: {sample_time} (seconds)') seconds = size * sample_time print(f' Sleeping for size * sample_time = {seconds} seconds...') time.sleep(seconds) # Check if the stimulation is still running print('Trying to pause stimuli, checking if we are running...') command = ((CMD["CMD_DBG_READ"] << 12) + VAL["VAL_RUNNING"]) send_cmd(dev, command) print(f'Sent CMD_DBG_READ, VAL_RUNNING. ', end='') res = recv_res(dev) print(f'Response: {hex(res)} ', end='') if res == 1 : print(colored("is running", 'white')) elif res == 0 : print(colored("is stopped", 'white')) else : sys.exit(colored("ERROR: unexpected response", 'red')) # Due to system's sleep (on which python's time.sleep() depends) not being # infinitely precise, at high frecuencies we may need some extra time to wait # for the stimulation to finish if (res == 1) and (args.noninteractive == True) : print(" Polling FPGA until stimuli finish: ", end='') while res == 1 : send_cmd(dev, command) res = recv_res(dev) print('.', end='') print(' ', end='') print(colored("OK", 'green')) # Pause stimulator if it is not already paused if res == 1 : command = CMD["CMD_PAUSE"] << 12 send_cmd(dev, command) print(f'Sent CMD_PAUSE.') # Exit if args.noninteractive == False : input('Press any key to exit') else : print('Running non-interactively, exiting')