UART
The UART (Universal Asynchronous Receiver/Transmitter) interface provides asynchronous serial communication. Supports configurable baud rates from 300 to 12,000,000 bps with various data formats and voltage levels from 1.8 V to 3.3 V.
Some interfaces share the same GPIO pins, so only one can be used at a time. Check out the pin alternate functions page for more details.
The PWM pins use the same channels: enabling UART fails with 409 CONFIGURATION_CONFLICT while PWM is enabled on a pin of the same channel (DA17/DA18 for UART 0, DA21/DA22 for UART 1).
Enabling UART, or changing vcc while it is enabled, first measures its RX and TX pins. If any of them reads below −0.5 V or above vcc + 0.5 V, the request fails with 409 UNSAFE_PIN_VOLTAGE and the interface stays disabled. The error message names the pin and the voltage measured. Reconfiguring with an unchanged vcc does not repeat the check.
Quick Start​
The example code below shows how to enable the UART interface, configure it, and transmit/receive data.
- NodeJS
- Python
- Rust
import { AT1000 } from '@ikalogic/at1000';
let devices = await AT1000.findDevices(); // Find devices (returns host strings)
let tester = await AT1000.open(devices[0]); // Open the first detected device
await tester.reset(); // Reset the device to a known state
let uart = tester.com.uart(0); // Select the first UART interface
await uart.enable({
vcc: 3.3,
baud_rate: 115200,
data_bits: 8,
parity: 'none',
stop_bits: '1'
}); // Enable UART with standard 115200 8N1 configuration
// Transmit data
await uart.tx("Hello, World!\r\n");
// Start continuous receive
await uart.start_rx();
// Wait for some data
await new Promise(resolve => setTimeout(resolve, 100));
// Read received data
let data = await uart.rx();
console.log("Received:", data);
// Stop reception
await uart.stop_rx();
// Disable UART interface
await uart.disable();
from ikalogic_at1000 import AT1000
import time
devices = AT1000.find_devices() # Find devices (returns host strings)
tester = AT1000.open(devices[0]) # Open the first detected device
tester.reset() # Reset the device to a known state
uart = tester.com.uart(0) # Select the first UART interface
uart.enable(
vcc=3.3,
baud_rate=115200,
data_bits=8,
parity='none',
stop_bits='1'
) # Enable UART with standard 115200 8N1 configuration
# Transmit data
uart.tx("Hello, World!\r\n")
# Start continuous receive
uart.start_rx()
# Wait for some data
time.sleep(0.1)
# Read received data
data = uart.rx()
print(f"Received: {data}")
# Stop reception
uart.stop_rx()
# Disable UART interface
uart.disable()
use ikalogic_at1000::{AT1000, SerialParity, SerialStopBits, UartEnableConfig};
use std::{thread, time::Duration};
let devices = AT1000::find_devices(Duration::from_millis(500))?; // Find devices (returns host strings)
let tester = AT1000::open(&devices[0])?; // Open the first detected device
tester.reset()?; // Reset the device to a known state
let uart = tester.com.uart(0)?; // Select the first UART interface
uart.enable(&UartEnableConfig {
vcc: Some(3.3),
baud_rate: Some(115200),
data_bits: Some(8),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
})?; // Enable UART with standard 115200 8N1 configuration
// Transmit data
uart.tx(b"Hello, World!\r\n")?;
// Start continuous receive
uart.start_rx()?;
// Wait for some data
thread::sleep(Duration::from_millis(100));
// Read received data
let data = uart.rx()?;
println!("Received: {data:?}");
// Stop reception
uart.stop_rx()?;
// Disable UART interface
uart.disable()?;
API Reference​
Accessing UART​
- NodeJS
- Python
- Rust
const uart = tester.com.uart(id);
uart = tester.com.uart(id)
let id = 0; // 0-based interface index
let uart = tester.com.uart(id)?;
Parameters:
id(number/int/u32): The UART interface identifier (0-based index)
Returns: Uart instance / Result<Uart, At1000Error> in Rust
Methods​
configure(config)​
Configures the UART interface parameters without changing the enabled state.
- NodeJS
- Python
- Rust
await uart.configure({
baud_rate: 9600,
data_bits: 8,
parity: 'none',
stop_bits: '1',
vcc: 3.3
});
uart.configure(
baud_rate=9600,
data_bits=8,
parity='none',
stop_bits='1',
vcc=3.3
)
use ikalogic_at1000::{SerialParity, SerialStopBits, UartConfigPatch};
let uart = tester.com.uart(0)?;
uart.configure(&UartConfigPatch {
baud_rate: Some(9600),
data_bits: Some(8),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
..Default::default()
})?;
Parameters:
config(UartConfigPatch/&UartConfigPatchin Rust): Configuration object with optional properties:enabled(boolean/bool/Option<bool>, optional): Enable or disable the interfacebaud_rate(number/int/Option<u32>, optional): Baud rate (300 to 12,000,000)data_bits(number/int/Option<u8>, optional): Data bits per character (7 or 8)parity(string/str/Option<SerialParity>, optional): Parity mode -'none','odd','even','mark', or'space'stop_bits(string/str/Option<SerialStopBits>, optional): Stop bits -'1','1.5', or'2'vcc(number/float/Option<f64>, optional): Logic level voltage (1.8 to 3.3)
Returns: Promise<UartConfig> / UartConfig / Result<UartConfig, At1000Error> - The updated configuration
Exceptions:
- Throws validation error if
baud_rateis outside [300, 12000000] - Throws validation error if
data_bitsis not 7 or 8 - Throws validation error if
parityis not'none','odd','even','mark', or'space' - Throws validation error if
stop_bitsis not'1','1.5', or'2' - Throws validation error if
vccis outside [1.8, 3.3]
enable([config])​
Enables the UART interface and optionally applies configuration.
- NodeJS
- Python
- Rust
await uart.enable({
baud_rate: 115200,
data_bits: 8,
parity: 'none',
stop_bits: '1',
vcc: 3.3
});
uart.enable(
baud_rate=115200,
data_bits=8,
parity='none',
stop_bits='1',
vcc=3.3
)
use ikalogic_at1000::{SerialParity, SerialStopBits, UartEnableConfig};
let uart = tester.com.uart(0)?;
uart.enable(&UartEnableConfig {
baud_rate: Some(115200),
data_bits: Some(8),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
})?;
Parameters:
config(UartEnableConfig, optional /&UartEnableConfigin Rust): Optional configuration to apply
Returns: Promise<UartConfig> / UartConfig / Result<UartConfig, At1000Error> - The updated configuration with enabled: true
disable()​
Disables the UART interface.
- NodeJS
- Python
- Rust
await uart.disable();
uart.disable()
let uart = tester.com.uart(0)?;
uart.disable()?;
Returns: Promise<UartConfig> / UartConfig / Result<UartConfig, At1000Error> - The updated configuration with enabled: false
tx(data)​
Transmits data over the UART interface.
- NodeJS
- Python
- Rust
// Transmit string data
await uart.tx("Hello\r\n");
// Transmit raw bytes
await uart.tx([0x48, 0x65, 0x6C, 0x6C, 0x6F, 0x0D, 0x0A]);
# Transmit string data
uart.tx("Hello\r\n")
# Transmit raw bytes
uart.tx([0x48, 0x65, 0x6C, 0x6C, 0x6F, 0x0D, 0x0A])
let uart = tester.com.uart(0)?;
// Transmit string data
uart.tx(b"Hello\r\n")?;
// Transmit raw bytes
uart.tx(&[0x48, 0x65, 0x6C, 0x6C, 0x6F, 0x0D, 0x0A])?;
Parameters:
data(SerialDataInput/&[u8]in Rust): Data to transmit, either astringor anumber[]byte array (up to 65536 bytes). Rust accepts bytes only - write text asb"..."ortext.as_bytes()
Returns: Promise<SerialData> / SerialData / Result<SerialData, At1000Error> - The transmitted data as a byte array (number[], Vec<u8> in Rust)
Exceptions:
- Throws validation error if data exceeds 65536 bytes
start_rx()​
Starts continuous reception. Data is buffered internally until read with rx().
- NodeJS
- Python
- Rust
await uart.start_rx();
uart.start_rx()
let uart = tester.com.uart(0)?;
uart.start_rx()?;
Returns: Promise<void> / None / Result<(), At1000Error>
stop_rx()​
Stops continuous reception.
- NodeJS
- Python
- Rust
await uart.stop_rx();
uart.stop_rx()
let uart = tester.com.uart(0)?;
uart.stop_rx()?;
Returns: Promise<void> / None / Result<(), At1000Error>
rx()​
Reads received data from the internal buffer.
- NodeJS
- Python
- Rust
// Read all available data
const data = await uart.rx();
console.log("Received:", data);
# Read all available data
data = uart.rx()
print(f"Received: {data}")
let uart = tester.com.uart(0)?;
// Read all available data
let data = uart.rx()?;
println!("Received: {data:?}");
Returns: Promise<SerialData> / SerialData / Result<SerialData, At1000Error> - Received data as byte array (number[], Vec<u8> in Rust)
Notes:
- Returns immediately with available data (non-blocking)
- Clears the receive buffer after reading
- Must call
start_rx()before receiving data
Types​
UartConfig​
Complete configuration object returned by configuration methods.
{
enabled: boolean,
baud_rate: number, // 300 to 12,000,000
data_bits: number, // 7 or 8
parity: 'none' | 'odd' | 'even' | 'mark' | 'space',
stop_bits: '1' | '1.5' | '2',
vcc: number // 1.8 to 3.3
}
Rust: ikalogic_at1000::UartConfig
pub struct UartConfig {
pub enabled: bool,
pub baud_rate: u32,
pub vcc: f64,
pub data_bits: u8,
pub stop_bits: SerialStopBits,
pub parity: SerialParity,
}
pub enum SerialStopBits { One, OnePointFive, Two }
pub enum SerialParity { None, Odd, Even, Mark, Space }
UartConfigPatch​
Partial configuration for the configure() method. All properties are optional.
{
enabled?: boolean,
baud_rate?: number,
data_bits?: number,
parity?: 'none' | 'odd' | 'even' | 'mark' | 'space',
stop_bits?: '1' | '1.5' | '2',
vcc?: number
}
Rust: ikalogic_at1000::UartConfigPatch (derives Default, so unset fields stay None)
pub struct UartConfigPatch {
pub enabled: Option<bool>,
pub baud_rate: Option<u32>,
pub vcc: Option<f64>,
pub data_bits: Option<u8>,
pub stop_bits: Option<SerialStopBits>,
pub parity: Option<SerialParity>,
}
UartEnableConfig​
Optional configuration for the enable() method.
{
baud_rate?: number,
data_bits?: number,
parity?: 'none' | 'odd' | 'even' | 'mark' | 'space',
stop_bits?: '1' | '1.5' | '2',
vcc?: number
}
Rust: ikalogic_at1000::UartEnableConfig (derives Default, so unset fields stay None)
pub struct UartEnableConfig {
pub baud_rate: Option<u32>,
pub vcc: Option<f64>,
pub data_bits: Option<u8>,
pub stop_bits: Option<SerialStopBits>,
pub parity: Option<SerialParity>,
}
SerialData​
Output data type, always returned as a byte array.
number[] // Array of bytes (0-255)
Rust: ikalogic_at1000::SerialData
pub type SerialData = Vec<u8>;
Common Baud Rates​
| Baud Rate | Description |
|---|---|
| 300 | Low-speed legacy |
| 1200 | Legacy modems |
| 2400 | Legacy modems |
| 4800 | Legacy devices |
| 9600 | Common default |
| 19200 | Common |
| 38400 | Common |
| 57600 | Common |
| 115200 | Most common high-speed |
| 230400 | High-speed |
| 460800 | High-speed |
| 921600 | Very high-speed |
| 1000000 | 1 Mbps |
| 2000000 | 2 Mbps |
| 12000000 | 12 Mbps (maximum) |
Usage Example​
- NodeJS
- Python
- Rust
import { AT1000 } from '@ikalogic/at1000';
// Find and connect to device
const devices = await AT1000.findDevices();
const tester = await AT1000.open(devices[0]);
await tester.reset(); // Reset the device to a known state
// Access UART interface 0
const uart = tester.com.uart(0);
// Enable with standard 115200 8N1 configuration
await uart.enable({
baud_rate: 115200,
data_bits: 8,
parity: 'none',
stop_bits: '1',
vcc: 3.3
});
// Start continuous reception
await uart.start_rx();
// Send a command
await uart.tx("AT\r\n");
// Wait for response
await new Promise(resolve => setTimeout(resolve, 100));
// Read response
const response = await uart.rx();
console.log("Response:", Buffer.from(response).toString());
// Stop reception and disable
await uart.stop_rx();
await uart.disable();
from ikalogic_at1000 import AT1000
import time
# Find and connect to device
devices = AT1000.find_devices()
tester = AT1000.open(devices[0])
tester.reset() # Reset the device to a known state
# Access UART interface 0
uart = tester.com.uart(0)
# Enable with standard 115200 8N1 configuration
uart.enable(
baud_rate=115200,
data_bits=8,
parity='none',
stop_bits='1',
vcc=3.3
)
# Start continuous reception
uart.start_rx()
# Send a command
uart.tx("AT\r\n")
# Wait for response
time.sleep(0.1)
# Read response
response = uart.rx()
print(f"Response: {bytes(response).decode()}")
# Stop reception and disable
uart.stop_rx()
uart.disable()
use ikalogic_at1000::{AT1000, SerialParity, SerialStopBits, UartEnableConfig};
use std::{thread, time::Duration};
// Find and connect to device
let devices = AT1000::find_devices(Duration::from_millis(500))?;
let tester = AT1000::open(&devices[0])?;
tester.reset()?; // Reset the device to a known state
// Access UART interface 0
let uart = tester.com.uart(0)?;
// Enable with standard 115200 8N1 configuration
uart.enable(&UartEnableConfig {
baud_rate: Some(115200),
data_bits: Some(8),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
})?;
// Start continuous reception
uart.start_rx()?;
// Send a command
uart.tx(b"AT\r\n")?;
// Wait for response
thread::sleep(Duration::from_millis(100));
// Read response
let response = uart.rx()?;
println!("Response: {}", String::from_utf8_lossy(&response));
// Stop reception and disable
uart.stop_rx()?;
uart.disable()?;
Common Patterns​
AT Command Interface​
- NodeJS
- Python
- Rust
async function sendATCommand(uart, command, timeout = 1000) {
// Clear any pending data
await uart.rx();
// Send command
await uart.tx(command + "\r\n");
// Wait for response
await new Promise(resolve => setTimeout(resolve, timeout));
// Read response
const response = await uart.rx();
return Buffer.from(response).toString().trim();
}
// Usage
await uart.start_rx();
const info = await sendATCommand(uart, "ATI");
console.log("Device info:", info);
const signal = await sendATCommand(uart, "AT+CSQ");
console.log("Signal quality:", signal);
import time
def send_at_command(uart, command, timeout=1.0):
# Clear any pending data
uart.rx()
# Send command
uart.tx(command + "\r\n")
# Wait for response
time.sleep(timeout)
# Read response
response = uart.rx()
return bytes(response).decode().strip()
# Usage
uart.start_rx()
info = send_at_command(uart, "ATI")
print(f"Device info: {info}")
signal = send_at_command(uart, "AT+CSQ")
print(f"Signal quality: {signal}")
use ikalogic_at1000::{At1000Error, Uart};
use std::{thread, time::Duration};
fn send_at_command(uart: &Uart, command: &str, timeout: Duration) -> Result<String, At1000Error> {
// Clear any pending data
uart.rx()?;
// Send command
uart.tx(format!("{command}\r\n").as_bytes())?;
// Wait for response
thread::sleep(timeout);
// Read response
let response = uart.rx()?;
Ok(String::from_utf8_lossy(&response).trim().to_string())
}
// Usage
let uart = tester.com.uart(0)?;
uart.start_rx()?;
let info = send_at_command(&uart, "ATI", Duration::from_secs(1))?;
println!("Device info: {info}");
let signal = send_at_command(&uart, "AT+CSQ", Duration::from_secs(1))?;
println!("Signal quality: {signal}");
GPS NMEA Receiver​
- NodeJS
- Python
- Rust
// Configure for GPS module (9600 baud typical)
await uart.enable({
baud_rate: 9600,
data_bits: 8,
parity: 'none',
stop_bits: '1',
vcc: 3.3
});
await uart.start_rx();
// Read NMEA sentences
let buffer = '';
while (true) {
const data = await uart.rx();
if (data.length > 0) {
buffer += Buffer.from(data).toString();
// Process complete sentences
const lines = buffer.split('\r\n');
buffer = lines.pop(); // Keep incomplete sentence
for (const line of lines) {
if (line.startsWith('$GPGGA')) {
console.log("Position:", line);
} else if (line.startsWith('$GPRMC')) {
console.log("Navigation:", line);
}
}
}
await new Promise(resolve => setTimeout(resolve, 100));
}
import time
# Configure for GPS module (9600 baud typical)
uart.enable(
baud_rate=9600,
data_bits=8,
parity='none',
stop_bits='1',
vcc=3.3
)
uart.start_rx()
# Read NMEA sentences
buffer = ''
while True:
data = uart.rx()
if len(data) > 0:
buffer += bytes(data).decode()
# Process complete sentences
lines = buffer.split('\r\n')
buffer = lines.pop() # Keep incomplete sentence
for line in lines:
if line.startswith('$GPGGA'):
print(f"Position: {line}")
elif line.startswith('$GPRMC'):
print(f"Navigation: {line}")
time.sleep(0.1)
use ikalogic_at1000::{SerialParity, SerialStopBits, UartEnableConfig};
use std::{thread, time::Duration};
let uart = tester.com.uart(0)?;
// Configure for GPS module (9600 baud typical)
uart.enable(&UartEnableConfig {
baud_rate: Some(9600),
data_bits: Some(8),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
})?;
uart.start_rx()?;
// Read NMEA sentences
let mut buffer = String::new();
loop {
let data = uart.rx()?;
if !data.is_empty() {
buffer.push_str(&String::from_utf8_lossy(&data));
// Process complete sentences, keeping the incomplete tail in the buffer
while let Some(index) = buffer.find("\r\n") {
let sentence: String = buffer.drain(..index + 2).collect();
let line = sentence.trim_end();
if line.starts_with("$GPGGA") {
println!("Position: {line}");
} else if line.starts_with("$GPRMC") {
println!("Navigation: {line}");
}
}
}
thread::sleep(Duration::from_millis(100));
}
Binary Protocol​
- NodeJS
- Python
- Rust
// Send binary command with header and checksum
function buildPacket(command, data) {
const packet = [0xAA, 0x55, command, data.length, ...data];
const checksum = packet.reduce((a, b) => a ^ b, 0);
return [...packet, checksum];
}
// Read binary response
async function readPacket(uart, timeout = 1000) {
const startTime = Date.now();
let buffer = [];
while (Date.now() - startTime < timeout) {
const data = await uart.rx();
buffer.push(...data);
// Check for complete packet
if (buffer.length >= 4 && buffer[0] === 0xAA && buffer[1] === 0x55) {
const length = buffer[3];
if (buffer.length >= 5 + length) {
return buffer.slice(0, 5 + length);
}
}
await new Promise(resolve => setTimeout(resolve, 10));
}
throw new Error("Timeout waiting for packet");
}
// Usage
const packet = buildPacket(0x01, [0x10, 0x20]);
await uart.tx(packet);
const response = await readPacket(uart);
import time
from functools import reduce
# Send binary command with header and checksum
def build_packet(command, data):
packet = [0xAA, 0x55, command, len(data)] + list(data)
checksum = reduce(lambda a, b: a ^ b, packet, 0)
return packet + [checksum]
# Read binary response
def read_packet(uart, timeout=1.0):
start_time = time.time()
buffer = []
while time.time() - start_time < timeout:
data = uart.rx()
buffer.extend(data)
# Check for complete packet
if len(buffer) >= 4 and buffer[0] == 0xAA and buffer[1] == 0x55:
length = buffer[3]
if len(buffer) >= 5 + length:
return buffer[:5 + length]
time.sleep(0.01)
raise TimeoutError("Timeout waiting for packet")
# Usage
packet = build_packet(0x01, [0x10, 0x20])
uart.tx(packet)
response = read_packet(uart)
use ikalogic_at1000::{At1000Error, Uart};
use std::{thread, time::{Duration, Instant}};
// Send binary command with header and checksum
fn build_packet(command: u8, data: &[u8]) -> Vec<u8> {
let mut packet = vec![0xAA, 0x55, command, data.len() as u8];
packet.extend_from_slice(data);
let checksum = packet.iter().fold(0u8, |a, b| a ^ b);
packet.push(checksum);
packet
}
// Read binary response, returning `None` when the timeout elapses
fn read_packet(uart: &Uart, timeout: Duration) -> Result<Option<Vec<u8>>, At1000Error> {
let start = Instant::now();
let mut buffer: Vec<u8> = Vec::new();
while start.elapsed() < timeout {
let data = uart.rx()?;
buffer.extend_from_slice(&data);
// Check for complete packet
if buffer.len() >= 4 && buffer[0] == 0xAA && buffer[1] == 0x55 {
let length = usize::from(buffer[3]);
if buffer.len() >= 5 + length {
return Ok(Some(buffer[..5 + length].to_vec()));
}
}
thread::sleep(Duration::from_millis(10));
}
Ok(None)
}
// Usage
let uart = tester.com.uart(0)?;
let packet = build_packet(0x01, &[0x10, 0x20]);
uart.tx(&packet)?;
match read_packet(&uart, Duration::from_secs(1))? {
Some(response) => println!("Response: {response:?}"),
None => println!("Timeout waiting for packet"),
}
Line-Based Protocol​
- NodeJS
- Python
- Rust
class LineReader {
constructor(uart) {
this.uart = uart;
this.buffer = '';
}
async readLine(timeout = 5000) {
const startTime = Date.now();
while (Date.now() - startTime < timeout) {
const data = await this.uart.rx();
this.buffer += Buffer.from(data).toString();
const newlineIndex = this.buffer.indexOf('\n');
if (newlineIndex !== -1) {
const line = this.buffer.substring(0, newlineIndex).replace('\r', '');
this.buffer = this.buffer.substring(newlineIndex + 1);
return line;
}
await new Promise(resolve => setTimeout(resolve, 10));
}
throw new Error("Timeout waiting for line");
}
async writeLine(text) {
await this.uart.tx(text + '\r\n');
}
}
// Usage
const reader = new LineReader(uart);
await reader.writeLine("GET STATUS");
const status = await reader.readLine();
console.log("Status:", status);
import time
class LineReader:
def __init__(self, uart):
self.uart = uart
self.buffer = ''
def read_line(self, timeout=5.0):
start_time = time.time()
while time.time() - start_time < timeout:
data = self.uart.rx()
self.buffer += bytes(data).decode()
newline_index = self.buffer.find('\n')
if newline_index != -1:
line = self.buffer[:newline_index].replace('\r', '')
self.buffer = self.buffer[newline_index + 1:]
return line
time.sleep(0.01)
raise TimeoutError("Timeout waiting for line")
def write_line(self, text):
self.uart.tx(text + '\r\n')
# Usage
reader = LineReader(uart)
reader.write_line("GET STATUS")
status = reader.read_line()
print(f"Status: {status}")
use ikalogic_at1000::{At1000Error, Uart};
use std::{thread, time::{Duration, Instant}};
struct LineReader<'a> {
uart: &'a Uart,
buffer: String,
}
impl<'a> LineReader<'a> {
fn new(uart: &'a Uart) -> Self {
Self { uart, buffer: String::new() }
}
// Returns `None` when no complete line arrives before the timeout
fn read_line(&mut self, timeout: Duration) -> Result<Option<String>, At1000Error> {
let start = Instant::now();
while start.elapsed() < timeout {
let data = self.uart.rx()?;
self.buffer.push_str(&String::from_utf8_lossy(&data));
if let Some(index) = self.buffer.find('\n') {
let line: String = self.buffer.drain(..index + 1).collect();
return Ok(Some(line.trim_end().to_string()));
}
thread::sleep(Duration::from_millis(10));
}
Ok(None)
}
fn write_line(&self, text: &str) -> Result<(), At1000Error> {
self.uart.tx(format!("{text}\r\n").as_bytes())?;
Ok(())
}
}
// Usage
let uart = tester.com.uart(0)?;
let mut reader = LineReader::new(&uart);
reader.write_line("GET STATUS")?;
match reader.read_line(Duration::from_secs(5))? {
Some(status) => println!("Status: {status}"),
None => println!("Timeout waiting for line"),
}
Echo Test​
- NodeJS
- Python
- Rust
// Loop TX to RX for echo test
async function echoTest(uart, testData) {
await uart.start_rx();
// Send test data
await uart.tx(testData);
// Wait for echo
await new Promise(resolve => setTimeout(resolve, 100));
// Read response
const received = await uart.rx();
// Compare
const sent = typeof testData === 'string'
? Array.from(Buffer.from(testData))
: testData;
const match = sent.length === received.length &&
sent.every((v, i) => v === received[i]);
console.log(`Echo test: ${match ? 'PASS' : 'FAIL'}`);
return match;
}
await echoTest(uart, "Hello, World!");
import time
# Loop TX to RX for echo test
def echo_test(uart, test_data):
uart.start_rx()
# Send test data
uart.tx(test_data)
# Wait for echo
time.sleep(0.1)
# Read response
received = uart.rx()
# Compare
if isinstance(test_data, str):
sent = list(test_data.encode())
else:
sent = list(test_data)
match = len(sent) == len(received) and all(
s == r for s, r in zip(sent, received)
)
print(f"Echo test: {'PASS' if match else 'FAIL'}")
return match
echo_test(uart, "Hello, World!")
use ikalogic_at1000::{At1000Error, Uart};
use std::{thread, time::Duration};
// Loop TX to RX for echo test
fn echo_test(uart: &Uart, test_data: &[u8]) -> Result<bool, At1000Error> {
uart.start_rx()?;
// Send test data
uart.tx(test_data)?;
// Wait for echo
thread::sleep(Duration::from_millis(100));
// Read response
let received = uart.rx()?;
// Compare
let matched = received == test_data;
println!("Echo test: {}", if matched { "PASS" } else { "FAIL" });
Ok(matched)
}
let uart = tester.com.uart(0)?;
echo_test(&uart, b"Hello, World!")?;
Notes​
- The UART interface supports TTL-level signals only (not RS-232 voltage levels)
- Use appropriate level shifters for 5V logic when
vccis set higher - Flow control (RTS/CTS) is not supported
- The receive buffer has limited capacity; read frequently to avoid overflow
- For very high baud rates (> 1 Mbps), ensure signal integrity and short cable lengths
- Always call
start_rx()before expecting to receive data - The maximum single transmission is 65536 bytes