dengweijie/lib_xua
1
0
Fork 0
forked from PAWPAW-Mirror/lib_xua
Code Pull Requests Activity

Add initial test tx + checkers from fwk_io

Browse Source
This commit is contained in:
Ed
2024-04-15 12:59:55 +01:00
parent 17206d4b8f
commit 4e4ae01a35
4 changed files with 505 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/test_midi/src/app_midi_simple.xc
View File

@@ -46,8 +46,6 @@ void ctrlPort();
#define VELOCITY 80
void test(chanend c_midi){
printf("Test\n");
struct midi_in_parse_state mips;
reset_midi_state(mips);
@@ -103,7 +101,7 @@ int main()
on tile[0]: test(c_midi);
on tile[1]: usb_midi(p_midi_rx, p_midi_tx, clk_midi, c_midi, 0);
// Setup HW so we can run this on the MC board
on tile[0]: ctrlPort();
}

83
tests/test_midi_tx.py Normal file
View File

@@ -0,0 +1,83 @@
# Copyright 2014-2024 XMOS LIMITED.
# This Software is subject to the terms of the XMOS Public Licence: Version 1.
import pytest
import Pyxsim
from Pyxsim import testers
from pathlib import Path
from uart_tx_checker import UARTTxChecker
# from spdif_test_utils import (
# Clock,
# Spdif_rx,
# Frames,
# freq_for_sample_rate,
# )
MAX_CYCLES = 15000000
MIDI_RATE = 31250
CONFIGS = ["xs2", "xs3"]
CONFIGS = ["xs3"]
class Midi_expect:
def __init(self):
pass
def expect(self):
expected = "Hello"
return expected
#####
# This test builds the spdif transmitter app with a verity of presets and tests that the output matches those presets
#####
@pytest.mark.parametrize("config", CONFIGS)
def test_tx(capfd, config):
xe = str(Path(__file__).parent / f"test_midi/bin/{config}/test_midi_{config}.xe")
p_midi_out = "tile[1]:XS1_PORT_4C"
# tester = testers.ComparisonTester(
# Frames(channels=audio, no_of_blocks=no_of_blocks, sam_freq=sam_freq).expect()[
# : no_of_samples * len(audio)
# ]
# )
tester = testers.ComparisonTester(Midi_expect().expect())
tx_port = "tile[1]:XS1_PORT_4C"
rx_port = None
baud = MIDI_RATE
bpb = 8
parity = 0
stop = 1
length_of_test = 3 # characters
simthreads = [
# UARTTxChecker(rx_port, tx_port, parity, baud, length_of_test, stop, bpb)
]
simargs = ["--max-cycles", str(MAX_CYCLES)]
simargs.extend(["--trace-to", "trace.txt", "--vcd-tracing", "-tile tile[1] -ports -o trace.vcd"]) #This is just for local debug so we can capture the run, pass as kwarg to run_with_pyxsim
# result = Pyxsim.run_on_simulator(
result = Pyxsim.run_on_simulator(
xe,
simthreads=simthreads,
instTracing=True,
# clean_before_build=True,
clean_before_build=False,
tester=tester,
# capfd=capfd,
capfd=None,
timeout=1500,
simargs=simargs,
build_options=[
"-j",
f"CONFIG={config}",
"EXTRA_BUILD_FLAGS="
+ f" -DMIDI_RATE_HZ={MIDI_RATE}"
,
],
)
assert result

177
tests/uart_rx_checker.py Normal file
View File

@@ -0,0 +1,177 @@
# Copyright 2022 XMOS LIMITED.
# This Software is subject to the terms of the XMOS Public Licence: Version 1.
import Pyxsim as px
from typing import Sequence
from functools import partial
# We need to disable output buffering for this test to work on MacOS; this has
# no effect on Linux systems. Let's redefine print once to avoid putting the
# same argument everywhere.
print = partial(print, flush=True)
Parity = dict(
UART_PARITY_EVEN=0,
UART_PARITY_ODD=1,
UART_PARITY_NONE=2,
UART_PARITY_BAD=3
)
class DriveHigh(px.SimThread):
def __init__(self, p):
self._p = p
def run(self):
xsi = self.xsi
xsi.drive_port_pins(self._p, 1);
class UARTRxChecker(px.SimThread):
def __init__(self, rx_port, tx_port, parity, baud, stop_bits, bpb, data=[0x7f, 0x00, 0x2f, 0xff],
intermittent=False):
"""
Create a UARTRxChecker instance.
:param rx_port: Receive port of the UART device under test.
:param tx_port: Transmit port of the UART device under test.
:param parity: Parity of the UART connection.
:param baud: BAUD rate of the UART connection.
:param stop_bits: Number of stop_bits for each UART byte.
:param bpb: Number of data bits per "byte" of UART data.
:param data: A list of bytes to send (default: [0x7f, 0x00, 0x2f, 0xff])
:param intermittent: Add a random delay between sent bytes.
"""
self._rx_port = rx_port
self._tx_port = tx_port
self._parity = parity
self._baud = baud
self._stop_bits = stop_bits
self._bits_per_byte = bpb
self._data = data
self._intermittent = intermittent
# Hex value of stop bits, as MSB 1st char, e.g. 0b11 : 0xC0
def send_byte(self, xsi, byte):
"""
Send a byte to the rx_port
:param xsi: XMOS Simulator Instance.
:param byte: Byte to send
"""
# Send start bit
self.send_start(xsi)
# Send data
self.send_data(xsi, byte)
# Send parity
self.send_parity(xsi, byte)
# Send stop bit(s)
self.send_stop(xsi)
def send_start(self, xsi):
"""
Send a start bit.
:param xsi: XMOS Simulator Instance.
"""
xsi.drive_port_pins(self._rx_port, 0)
self.wait_baud_time(xsi)
def send_data(self, xsi, byte):
"""
Write the data bits to the rx_port
:param xsi: XMOS Simulator Instance.
:param byte: Data to send.
"""
# print "0x%02x:" % byte
for x in range(self._bits_per_byte):
# print " Sending bit %d of 0x%02x (%d)" % (x, byte, (byte >> x) & 0x01)
xsi.drive_port_pins(self._rx_port, (byte & (0x01 << x)) >= 1)
# print " (x): %d" % ((byte & (0x01 << x))>=1)
self.wait_baud_time(xsi)
def send_parity(self, xsi, byte):
"""
Send the parity bit to the rx_port
:param xsi: XMOS Simulator Instance.
:param byte: Data to send parity of.
"""
parity = (self._parity - 1) % 3 #parity enum in lib_uart (old XC) different from SDK
if parity < 2:
crc_sum = 0
for x in range(self._bits_per_byte):
crc_sum += ((byte & (0x01 << x)) >= 1)
crc_sum += parity
# print "Parity for 0x%02x: %d" % (byte, crc_sum%2)
xsi.drive_port_pins(self._rx_port, crc_sum % 2)
self.wait_baud_time(xsi)
elif parity == Parity['UART_PARITY_BAD']:
# print "Sending bad parity bit"
self.send_bad_parity(xsi)
def send_stop(self, xsi):
"""
Send the stop bit(s) to the rx_port
:param xsi: XMOS Simulator Instance.
"""
for x in range(self._stop_bits):
xsi.drive_port_pins(self._rx_port, 1)
self.wait_baud_time(xsi)
def send_bad_parity(self, xsi):
"""
Send a parity bit of 1 to simulate an incorrect parity state.
:param xsi: XMOS Simulator Instance.
"""
# Always send a parity bit of 1
xsi.drive_port_pins(self._rx_port, 0)
self.wait_baud_time(xsi)
def get_bit_time(self):
"""
Returns the expected time between bits for the currently set BAUD rate.
Returns float value in nanoseconds.
"""
# Return float value in ps
return (1.0 / self._baud) * 1e12
def wait_baud_time(self, xsi):
"""
Wait for 1 bit time, as determined by the baud rate.
"""
self.wait_until(xsi.get_time() + self.get_bit_time())
def wait_half_baud_time(self, xsi):
"""
Wait for half a bit time, as determined by the baud rate.
"""
self.wait_until(xsi.get_time() + (self.get_bit_time() / 2))
def run(self):
xsi = self.xsi
# Drive the uart line high.
xsi.drive_port_pins(self._rx_port, 1)
# Wait for the device to bring up it's tx port, indicating it is ready
self.wait((lambda _x: self.xsi.is_port_driving(self._tx_port)))
# If we're doing an intermittent send, add a delay between each byte
# sent. Delay is in ns. 20,000ns = 20ms, 100,000ns = 100ms. Delays could
# be more variable, but it hurts test time substantially.
if self._intermittent:
for x in self._data:
k = randint(20000, 100000)
self.wait_until(xsi.get_time() + k)
self.send_byte(xsi, x)
else:
for x in self._data:
self.send_byte(xsi, x)

244
tests/uart_tx_checker.py Normal file
View File

@@ -0,0 +1,244 @@
# Copyright 2022 XMOS LIMITED.
# This Software is subject to the terms of the XMOS Public Licence: Version 1.
import Pyxsim as px
from typing import Sequence
from functools import partial
# We need to disable output buffering for this test to work on MacOS; this has
# no effect on Linux systems. Let's redefine print once to avoid putting the
# same argument everywhere.
print = partial(print, flush=True)
class UARTTxChecker(px.SimThread):
"""
This simulator thread will act as a UART device, and will check sent and
transations caused by the device, by looking at the tx pins.
"""
def __init__(self, rx_port, tx_port, parity, baud, length, stop_bits, bpb):
"""
Create a UARTTxChecker instance.
:param tx_port: Transmit port of the UART device under test.
:param parity: Parity of the UART connection.
:param baud: BAUD rate of the UART connection.
:param length: Length of transmission to check.
:param stop_bits: Number of stop_bits for each UART byte.
:param bpb: Number of data bits per "byte" of UART data.
"""
self._tx_port = tx_port
self._parity = parity
self._baud = baud
self._length = length
self._stop_bits = stop_bits
self._bits_per_byte = bpb
# Hex value of stop bits, as MSB 1st char, e.g. 0b11 : 0xC0
def get_port_val(self, xsi, port):
"""
Sample the state of a port
:rtype: int
:param xsi: XMOS Simulator Instance.
:param port: Port to sample.
"""
is_driving = xsi.is_port_driving(port)
if not is_driving:
return 1
else:
return xsi.sample_port_pins(port)
def get_bit_time(self):
"""
Returns the expected time between bits for the currently set BAUD rate.
Returns float value in nanoseconds.
:rtype: float
"""
# Return float value in ps
return (1.0/self._baud) * 1e12
def wait_baud_time(self, xsi):
"""
Wait for 1 bit time, as determined by the baud rate.
"""
self.wait_until(xsi.get_time() + self.get_bit_time())
return True
def wait_half_baud_time(self, xsi):
"""
Wait for half a bit time, as determined by the baud rate.
"""
self.wait_until(xsi.get_time() + (self.get_bit_time() / 2))
def read_packet(self, xsi, parity, length=4):
"""
Read a given number of bytes of UART traffic sent by the device.
Returns a list of bytes sent by the device.
:rtype: list
:param xsi: XMOS Simulator Instance.
:param parity: The UART partiy setting. See Parity.
:param length: The number of bytes to read. Defaults to 4.
"""
packet = []
start_time = 0
got_start_bit = False
initial_port_val = self.get_port_val(xsi, self._tx_port)
print("tx starts high: %s" % ("True" if initial_port_val else "False"))
for x in range(length):
packet.append(chr(self.read_byte(xsi, parity)))
return packet
def read_byte(self, xsi, parity):
"""
Read 1 byte of UART traffic sent by the device
Returns an int, representing a byte read from the uart. Should be in the range 0 <= x < 2^bits_per_byte
:rtype: int
:param xsi: XMOS Simulator Instance.
:param parity: The UART partiy setting. See Parity.
"""
byte = 0
val = 0
# Recv start bit
initial_port_val = self.get_port_val(xsi, self._tx_port)
if initial_port_val == 1:
self.wait_for_port_pins_change([self._tx_port])
#else go for it as assume tx has just fallen with no interframe gap
# print("Byte start time: ", xsi.get_time())
# The tx line should go low for 1 bit time
if self.get_val_timeout(xsi, self._tx_port) == 0:
print("Start bit recv'd")
else:
print("Start bit issue")
return False
# recv the byte
crc_sum = 0
for j in range(self._bits_per_byte):
val = self.get_val_timeout(xsi, self._tx_port)
byte += (val << j)
crc_sum += val
print(f"Sampled {self._bits_per_byte} data bits")
# Check the parity if needs be
self.check_parity(xsi, crc_sum, parity)
# Get the stop bit
self.check_stopbit(xsi)
# Print a new line to split bytes in output
print()
return byte
def check_parity(self, xsi, crc_sum, parity):
"""
Read the parity bit and check it against a crc sum. Print correctness.
:param xsi: XMOS Simulator Instance.
:param crc_sum: The checksum to test parity against.
:param parity: The UART partiy setting. See Parity.
"""
if parity > 0:
parity_val = 0 if parity == 1 else 1
read = self.get_val_timeout(xsi, self._tx_port)
if read == (crc_sum + parity_val) % 2:
print("Parity bit correct")
else:
print("Parity bit incorrect. Got %d, expected %d" % (read, (crc_sum + parity_val) % 2))
else:
print("Parity bit correct")
def check_stopbit(self, xsi):
"""
Read the stop bit(s) of a UART transmission and print correctness.
:param xsi: XMOS Simulator Instance.
"""
stop_bits_correct = True
for i in range(self._stop_bits):
# The stop bits should stay high for this time
if self.get_val_timeout(xsi, self._tx_port) == 0:
stop_bits_correct = False
print("Stop bit correct: %s" % ("True" if stop_bits_correct else "False"))
def get_val_timeout(self, xsi, port):
"""
Get a value from a given port of the device, with a timeout determined
by the BAUD rate.
Returns whether the pin is high (True) or low (False)
:rtype: bool
:param xsi: XMOS Simulator Instance.
:param port: The port to sample.
"""
# This intentionally has a 0.3% slop. It is per-byte and gives some
# wiggle-room if the clock doesn't divide into ns nicely.
timeout = self.get_bit_time() * 0.5
short_timeout = self.get_bit_time() * 0.2485
# Allow for "rise" time
self.wait_until(xsi.get_time() + short_timeout)
# Get val
K = self.wait_time_or_pin_change(xsi, timeout, port)
# Allow for "fall" time
self.wait_until(xsi.get_time() + short_timeout)
return K
def wait_time_or_pin_change(self, xsi, timeout, port):
"""
Waits for a given timeout, or until a port changes state. Which ever
occurs 1st. Prints an error if the former causes the function to break.
Returns whether the pin is high (True) or low (False)
:rtype: bool
:param xsi: XMOS Simulator Instance.
:param timeout: Time to wait.
:param port: Port to sample.
"""
start_time = xsi.get_time()
start_val = self.get_port_val(xsi, port)
transitioned_during_wait = False
def _continue(_timeout, _start_time, _start_val):
if xsi.get_time() >= _start_time + _timeout:
return True
if self.get_port_val(xsi, port) != _start_val:
transitioned_during_wait = True
return True
return False
wait_fun = (lambda x: _continue(timeout, start_time, start_val))
self.wait(wait_fun)
# Start value should *not* have changed during timeout
if transitioned_during_wait:
print("FAIL :: Unexpected Transition.")
return start_val
def run(self):
# Wait for the xcore to bring the uart tx port up
self.wait((lambda x: self.xsi.is_port_driving(self._tx_port)))
self.wait((lambda x: self.get_port_val(self.xsi, self._tx_port) == 1))
K = self.read_packet(self.xsi, self._parity, self._length)
# Print each member of K as a hex byte
# inline lambda function mapped over a list? awh yiss.
print(", ".join(map((lambda x: "0x%02x" % ord(x)), K)))