Control FPGA LEDs by using Python Code (instead of Vitis Serial Terminal)

[ebatman]

Hi! I'm currently working on how to interface the baremetal design in Zynq FPGA with the HMI using PyQT designer and pyserial in PC. (actuaclly I'm trying  to use python code to send some characters to the FPGA board and make the led blinking)
Step1: My design (on Zynq) consist of PS GPIO and PS UART. PS UART for receving and sending the value from Vitis Serial Terminal and GPIO to control the LEDs on the Zynq Board.
Step2: Export the design onto Vitis and write the C code (Baremetal).
Step3: I can verify the design work well with Vitis Serial Terminal or GTKterm.
For example, I can enter a number 8 bits from Vitis Serial Terminal, this value is receiving by UART and control the LEDs on the FPGA Board by GPIO EMIO + display back the value in the terminal.
Step4: Verifying with the IHM (using Pyserial)
In this step, I used a script to enter the value from the Terminal of Visual Studio Code (VS code). I can get back the value that I entered, but that all. It seems that this value can not be used by the GPIO part to control the LEDs. It's not like when I verifying by using Vitis Serial Terminal or GTKterm.
The photo is the design I have done in Vivado
The C code is the bare-metal to receive the value and control the LEDs on the FPGA board.

int main(void)
{
	int Status;
    u8 RecvChar[MAX_MESSAGE_LENGTH]; // Define a buffer to store received characters
	/*
	 * Run the Uart Echo example , specify the Base Address that is
	 * generated in xparameters.h
	 */
	//Status = UartPsSendReceivePC(UART_BASEADDR);
    // Call the function and pass the UART base address and the receive buffer
    int result = UartPsSendReceivePC(UART_BASEADDR, RecvChar);
    // Check the result of the function call
    if (result == XST_SUCCESS) {
        // Function executed successfully
        printf("UartPsSendReceivePC executed successfully\n");
        // Now you can use the received characters stored in RecvChar buffer
        printf("Received message ok: %s\n", RecvChar);
    } else {
        // Function execution failed
        printf("UartPsSendReceivePC execution failed\n");
    }

}

int UartPsSendReceivePC(u32 UartBaseAddress, u8 *RecvChar)
//int UartPsSendReceivePC(u32 UartBaseAddress)
{
    xil_printf("Test system\n\r");

	u32 Running;
    u32 leds = 0; // Variable to store the LED values
    int Status, i;
	//u8 RecvChar[MAX_MESSAGE_LENGTH];
	u8 Receive_text;
	u32 CntrlRegister;
    u32 BufferIndex = 0;

	XGpioPs_Config *ConfigPtrPS;
	XGpioPs mio_emio;

	ConfigPtrPS = XGpioPs_LookupConfig(XPAR_XGPIOPS_0_DEVICE_ID);
	if (ConfigPtrPS == NULL) {
	return XST_FAILURE;
	}

	Status = XGpioPs_CfgInitialize(&mio_emio, ConfigPtrPS, ConfigPtrPS->BaseAddr);
	if (Status != XST_SUCCESS) {
		return XST_FAILURE;
	}
	// EMIO bank IO n°2 : input
	XGpioPs_SetDirection(&mio_emio, 2, 0xffffffff); // bank2 : emio 54 -> 85 = output
	XGpioPs_SetOutputEnable(&mio_emio, 2, 0xffffffff); // pas obligatoire

	CntrlRegister = XUartPs_ReadReg(UartBaseAddress, XUARTPS_CR_OFFSET);

	/* Enable TX and RX for the device */
	XUartPs_WriteReg(UartBaseAddress, XUARTPS_CR_OFFSET,
			  ((CntrlRegister & ~XUARTPS_CR_EN_DIS_MASK) |
			   XUARTPS_CR_TX_EN | XUARTPS_CR_RX_EN));

	Running = TRUE;
	while (Running)
		{
		 /* Wait until there is data */
		while (!XUartPs_IsReceiveData(UartBaseAddress));

		RecvChar[BufferIndex] = XUartPs_ReadReg(UartBaseAddress,
					    XUARTPS_FIFO_OFFSET);
		Receive_text = RecvChar[BufferIndex];
		/* Change the capitalization */
		if (('a' <= Receive_text) && ('z' >= Receive_text)) {
			/* Convert the Capital letter to a small. */
			Receive_text = Receive_text;
		}
		else if (('A' <= Receive_text) && ('Z' >= Receive_text)) {
			/* Convert the small letter to a Capital. */
			Receive_text = Receive_text;
		}
		else if (CHAR_ESC == Receive_text) {
			Running = FALSE;
		}
		/* Echo the character back */
		XUartPs_WriteReg(UartBaseAddress,  XUARTPS_FIFO_OFFSET,
				  RecvChar[BufferIndex]);
        /* Check for end of message */
        if (RecvChar[BufferIndex] == '\r' || RecvChar[BufferIndex] == '\n')
        	{
            /* Null-terminate the received message */
            RecvChar[BufferIndex] = '\0';
            /* Print out the received message */
            xil_printf("Received message: %s\n", RecvChar);
            /*for further processing with program of Fabien, return this variable*/
    	    leds = atoi(RecvChar);

        	//for (i = 0; i < 50; i++)
        		//{
                //leds++;
        	    // Use atoi() to convert the string to an integer
        	    XGpioPs_Write(&mio_emio, 2, leds); // Write to LEDs
                usleep(500000);
        		//}

			/* Reset buffer index for the next message */
            if (strcmp(RecvChar, ESC_STRING) == 0) // Compare against ESC_STRING
            {
                Running = FALSE; // Set Running to FALSE if ESC is received
            }
            BufferIndex = 0;
        	}
        else
        	{
            BufferIndex++;
			}
		}
	return XST_SUCCESS;
}

The attachment is my script in PYTHON and result of running.

 

import serial
import time
from serial.tools import list_ports
import threading

def find_serial_port():
    # Get a list of all available serial ports
    ports = list(list_ports.comports())
    
    # Iterate through the list of ports and find the one you want
    for port in ports:
        if 'USB' in port.description:  # You can change this condition based on your specific requirements
            return port.device  # Return the serial port device name
    
    # If no suitable port is found, return None
    return None

# Serial port configuration
BAUD_RATE = 115200

def establish_connection():
    while True:
        serial_port = find_serial_port()
        if serial_port is not None:
            try:
                ser = serial.Serial(serial_port, BAUD_RATE, timeout=1)
                print(f"Serial port {serial_port} opened successfully")
                return ser
            except serial.SerialException:
                print(f"Failed to open serial port {serial_port}. Retrying in 3 seconds...")
                time.sleep(3)
        else:
            print("No suitable serial port found. Retrying in 3 seconds...")
            time.sleep(3)

def main():
    ser = establish_connection()
    if ser is None:
        print("No serial port available. Exiting...")
        return

    try:
        # Interaction loop
        while True:
            # Read input from user
            user_input = input("Enter text to send (type 'exit' to quit): ")

            # Check if the user wants to exit
            if user_input.lower() == 'exit':
                break

            # Send user input to the Zynq device
            ser.write(user_input.encode('ascii'))
            ser.flushOutput()  # Flush output buffer

            # Wait for response
            ser.flushInput()  # Flush input buffer
            response = ser.readline().decode('ascii').strip()

            # Print response
            print(f"Response from Zynq: {response}")

    except KeyboardInterrupt:
        print("\nExiting...")
    finally:
        # Close serial port
        ser.close()

if __name__ == "__main__":
    # Start a thread to continuously attempt to establish the serial connection
    connection_thread = threading.Thread(target=establish_connection, daemon=True)
    connection_thread.start()

    # Start the main function
    main()

 

[artvvb]

Hi @ebatman

Apologies for the delay.

On 5/7/2024 at 6:38 AM, ebatman said:

ser.flushOutput()  # Flush output buffer

It looks like both flushInput and flushOutput are deprecated, and discard the current contents of the input and output buffers. Calling this right after writing to the output buffer might abort the data transfer. Just using ser.write ought to be sufficient.

Thanks,

Arthur