samples: uart_async: implement uart async test

For testing purpose, it could be removed after the PR
is finished.

Signed-off-by: Jun Li <[email protected]>

Author: jli157

samples/drivers/uart_async/CMakeLists.txt

@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: Apache-2.0
+
+cmake_minimum_required(VERSION 3.13.1)
+include($ENV{ZEPHYR_BASE}/cmake/app/boilerplate.cmake NO_POLICY_SCOPE)
+project(uart_async)
+
+FILE(GLOB app_sources src/*.c)
+target_sources(app PRIVATE ${app_sources})

samples/drivers/uart_async/prj.conf

@@ -0,0 +1,14 @@
+#Enable async UART support
+CONFIG_UART_ASYNC_API=y
+
+#Enable log on UART
+CONFIG_THREAD_NAME=y
+CONFIG_LOG=y
+CONFIG_LOG_BACKEND_UART=y
+CONFIG_LOG_BACKEND_SHOW_COLOR=y
+## Enable debug log level
+CONFIG_LOG_DEFAULT_LEVEL=4
+
+## Enable shell
+CONFIG_SHELL=y
+CONFIG_PRINTK=y

samples/drivers/uart_async/src/main.c

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2019 Intel Corporation
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <zephyr.h>
+#include <sys/printk.h>
+#include <drivers/uart.h>
+#include <string.h>
+#include <kernel.h>
+
+#define ASYNC_BUF_SZ 128
+
+static struct main_data {
+	struct device *uart_dev;
+	uint8_t rx_buf[ASYNC_BUF_SZ];
+	uint8_t tx_buf[ASYNC_BUF_SZ];
+	struct k_sem tx_sem;
+} _main_data;
+
+static void uart_async_callback(struct uart_event *evt, void *priv_data)
+{
+	struct main_data *data = priv_data;
+
+	if (evt->type == UART_TX_DONE) {
+		printk("TX done: %d\n", evt->data.tx.len);
+		memset(data->tx_buf, 0, sizeof(data->tx_buf));
+
+	} else if (evt->type == UART_RX_RDY) {
+		uint8_t *src_buf = evt->data.rx.buf
+				+ evt->data.rx.offset;
+		memcpy(data->tx_buf, src_buf, evt->data.rx.len);
+		printk("RX done: %d - %s\n", evt->data.rx.len, data->tx_buf);
+		if (evt->data.rx.len > 0) {
+			k_sem_give(&data->tx_sem);
+		}
+	}
+}
+
+int main(void)
+{
+	uint32_t version = sys_kernel_version_get();
+
+	printk(
+		"UART-Async API Echo ARCH: %s, BOARD: %s, version: %d.%d.%d\n",
+		CONFIG_ARCH, CONFIG_BOARD,
+		SYS_KERNEL_VER_MAJOR(version),
+		SYS_KERNEL_VER_MINOR(version),
+		SYS_KERNEL_VER_PATCHLEVEL(version));
+
+	struct device *uart = device_get_binding("UART_6");
+
+	if (uart == NULL) {
+		printk("UART open failed!\n");
+		return -1;
+	}
+
+	struct main_data *data = &_main_data;
+
+	data->uart_dev = uart;
+
+	uint8_t *buf = data->rx_buf;
+
+	memset(buf, 0, sizeof(ASYNC_BUF_SZ));
+
+
+	uart_callback_set(uart, uart_async_callback, data);
+	k_sem_init(&data->tx_sem, 0, 1);
+
+	/* start rx */
+	uart_rx_enable(uart, buf, ASYNC_BUF_SZ, K_FOREVER.ticks);
+	while (1) {
+		k_sem_take(&data->tx_sem, K_FOREVER);
+		int len = strlen(data->tx_buf);
+
+		if (len > 0) {
+			uart_tx(data->uart_dev, data->tx_buf,
+					len, K_FOREVER.ticks);
+		}
+	}
+
+	return 0;
+}

samples/drivers/uart_async/test_uart.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+#
+# Copyright (c) 2019 Intel Corporation
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import sys, traceback
+import getopt
+import serial
+import os
+import time
+from threading import Thread
+from threading import Timer
+import click
+import signal
+
+class uart_test(Thread):
+	def __init__(self, tty_port, baud):
+		signal.signal(signal.SIGTERM, self.handleSignal)
+		signal.signal(signal.SIGINT, self.handleSignal)
+		signal.signal(signal.SIGTSTP, self.handleSignal)
+
+		Thread.__init__(self)
+
+		self.console = serial.Serial(
+			port=tty_port,
+			baudrate=baud,
+			parity=serial.PARITY_NONE,
+			stopbits=serial.STOPBITS_ONE,
+			bytesize=serial.EIGHTBITS,
+			timeout = 0.01, # 10 ms
+			)
+		self.running = True
+		if self.console.isOpen():
+			self.console.flushInput()
+			self.console.flushOutput()
+		self.start()
+
+	def __del__(self):
+		self.console.close()
+
+	def close(self):
+		self.running = False
+		time.sleep(1)
+		self.join()
+		if self.console.isOpen():
+			self.console.close()
+
+	def packet_send(self, cmd):
+		if self.console.isOpen():
+			self.console.write(cmd.encode())
+
+	def handleSignal(self, signum, frame):
+		self.close()
+
+	def tx_test(self):
+		if self.running:
+			my_packet = "This is a stm32 F429zi packet!\r\n"
+			print("Write: " + my_packet)
+			self.packet_send(my_packet)
+			self.tx_timer = Timer(1.0, self.tx_test)
+			self.tx_timer.start();
+
+	def run(self):
+
+		self.tx_test()
+		while self.running:
+			if self.console.inWaiting() > 0:
+				input_str = self.console.readline()
+				print("rcv: %s" % input_str)
+
+		self.tx_timer.cancel();
+		self.console.close();
+
+
+@click.command("Start dma test")
+@click.option('-b', '--baud', default=115200, type=int, help='Baud rate')
+@click.argument("tty_port")
+def main(tty_port, baud):
+	try:
+		test_shell =  uart_test(tty_port, baud)
+
+	except KeyboardInterrupt:
+		test_shell.close()
+		exit()
+	except serial.SerialException:
+		print("Open serial port %s failed!" % tty_port)
+	except SystemExit:
+		print('Exiting...')
+	except:
+		traceback.print_exc()
+		sys.exit(3)
+
+if __name__ == "__main__":
+	main()