UART
The UART (Universal Asynchronous Receiver/Transmitter) interface provides asynchronous serial communication. Supports the standard baud rates from 300 to 4,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 also fails with 409 CONFIGURATION_CONFLICT while its TX or RX pin is configured as a GPIO. Reset the pin to its default analog input (reset() on the pin) before enabling UART.
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, DataBits, SerialParity, SerialStopBits, UartConfigPatch};
use std::{thread, time::Duration};
let devices = At1000::find_devices(Duration::from_millis(500))?; // Find devices
let tester = At1000::open(&devices[0], None)?; // 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(UartConfigPatch {
vcc: Some(3.3),
baud_rate: Some(115200),
data_bits: Some(DataBits::Eight),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
..Default::default()
})?; // 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/usize): 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::{DataBits, SerialParity, SerialStopBits, UartConfigPatch};
let uart = tester.com.uart(0)?;
uart.configure(UartConfigPatch {
baud_rate: Some(9600),
data_bits: Some(DataBits::Eight),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
..Default::default()
})?;
Parameters:
config(UartConfigPatch): Configuration object with optional properties:enabled(boolean/bool/Option<bool>, optional): Enable or disable the interfacebaud_rate(number/int/Option<u32>, optional): Baud rate, one of the standard rates listed below (300 to 4,000,000)data_bits(number/int/Option<DataBits>, optional): Data bits per character (7 or 8;DataBits::SevenorDataBits::Eightin Rust)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)
Valid baud rates: 300, 600, 1200, 1800, 2400, 4800, 9600, 19200, 38400, 57600, 115200, 230400, 460800, 500000, 576000, 921600, 1000000, 1152000, 1500000, 2000000, 2500000, 3000000, 3500000, 4000000
Default configuration (after tester.reset()): disabled, baud_rate 115200, 8 data bits, no parity, 1 stop bit (8N1), vcc 3.3 V.
Returns: Promise<UartConfig> / UartConfig / Result<UartConfig, At1000Error> - The updated configuration
Exceptions:
- All three SDKs reject
baud_ratelocally unless it is one of the 24 valid rates listed above; the interval 300–4,000,000 is not a continuous range. No request is sent for an unsupported rate. - 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] - Throws error
409 UNSAFE_PIN_VOLTAGEif enabling, or changingvccwhile enabled, finds TX or RX below −0.5 V or abovevcc+ 0.5 V - Throws error
409 CONFIGURATION_CONFLICTwhen enabling while TX or RX is configured as a GPIO (not an analog input), or while I2C, SPI or PWM uses the same channel
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::{DataBits, SerialParity, SerialStopBits, UartConfigPatch};
let uart = tester.com.uart(0)?;
uart.enable(UartConfigPatch {
baud_rate: Some(115200),
data_bits: Some(DataBits::Eight),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
..Default::default()
})?;
Parameters:
config(UartEnableConfig, optional /UartConfigPatchin Rust, required): Configuration to apply. In Rust, passUartConfigPatch::default()to keep the current settings;enableignores the patch'senabledfield and always enables the interface
Returns: Promise<UartConfig> / UartConfig / Result<UartConfig, At1000Error> - The updated configuration with enabled: true
Exceptions:
- Same validation errors as
configure() - Throws error
409 UNSAFE_PIN_VOLTAGEif TX or RX reads below −0.5 V or abovevcc+ 0.5 V - Throws error
409 CONFIGURATION_CONFLICTwhile TX or RX is configured as a GPIO (not an analog input), or while I2C, SPI or PWM uses the same channel
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<Vec<u8>, At1000Error> - The transmitted data as a byte array (number[], Vec<u8> in Rust)
Exceptions:
- Throws validation error if data exceeds 65536 bytes
- Throws error
400 INVALID_CONFIGURATIONif the interface is disabled
start_rx()​
Starts continuous reception. Received data is buffered internally until read with rx(). start_rx() empties the device's receive buffer, but bytes that arrived while the UART was enabled and reception was not started may still be held by the interface and delivered by the next rx(). Discard them in your script if they matter.
- 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<Vec<u8>, 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 - The receive buffer holds up to 65536 bytes. Once it is full, incoming bytes are dropped until
rx()empties it
Exceptions:
- Throws error
400 INVALID_CONFIGURATIONif the interface is disabled
Types​
UartConfig​
Complete configuration object returned by configuration methods.
{
enabled: boolean,
baud_rate: number, // standard rate, 300 to 4,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: DataBits,
pub stop_bits: SerialStopBits,
pub parity: SerialParity,
}
pub enum DataBits {
Seven,
Eight,
}
#[non_exhaustive]
pub enum SerialStopBits {
One,
OnePointFive,
Two,
Unknown(String),
}
#[non_exhaustive]
pub enum SerialParity {
None,
Odd,
Even,
Mark,
Space,
Unknown(String),
}
Unknown(String) keeps a value reported by a newer firmware that this SDK version does not know, so a match on SerialStopBits or SerialParity needs a _ arm.
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<DataBits>,
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 has no separate enable type: enable() takes a UartConfigPatch and sets enabled: true itself.
SerialData​
Output data type, always returned as a byte array.
number[] // Array of bytes (0-255)
Rust: 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 |
| 4000000 | 4 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, DataBits, SerialParity, SerialStopBits, UartConfigPatch};
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], None)?;
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(UartConfigPatch {
baud_rate: Some(115200),
data_bits: Some(DataBits::Eight),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
..Default::default()
})?;
// 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::{DataBits, SerialParity, SerialStopBits, UartConfigPatch};
use std::{thread, time::Duration};
let uart = tester.com.uart(0)?;
// Configure for GPS module (9600 baud typical)
uart.enable(UartConfigPatch {
baud_rate: Some(9600),
data_bits: Some(DataBits::Eight),
parity: Some(SerialParity::None),
stop_bits: Some(SerialStopBits::One),
vcc: Some(3.3),
..Default::default()
})?;
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)
vccis at most 3.3 V: use external level shifters to talk to 5 V logic- Flow control (RTS/CTS) is not supported
- The receive buffer holds up to 65536 bytes; bytes received while it is full are dropped, so 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