Closed KurtE closed 2 years ago
Quick notes on this: So far it looks like it is not implemented or working on other ports as well including (RP2040, STM32F05) Does appear like it is implemented on ESP32
Notes to self and questions on how to best code stuff within CircuitPython.
Side comment: I still wonder why the writes are synchronous. They don't appear to be with MicroPython, and my quick look at PySerial, shows methods like: flush() and out_waiting()...
I guess the next question is, is it worth implementing? That is logically if the busio code is running synchronous, do you gain anything by using this option versus adding your own set of the value of the direction pin around the call to write?
So I tried this on ESP32-S3 as so far only ESP code appears to implement it. I setup two uart objects one using the option and one rolling our own:
import time
import board
import busio
import supervisor
from digitalio import DigitalInOut, Direction, Pull
import neopixel
led_colors = [
(255, 0, 0),
(0, 255, 0),
(0, 0, 255),
(255, 255, 0),
(255, 0, 255),
(0, 255, 255),
(255, 255, 255),
(0, 0, 0),
]
color_index = 0
check_pin = DigitalInOut(board.IO4)
check_pin.direction = Direction.INPUT
check_pin.pull = Pull.DOWN
uart1 = None
uart2 = None
led = neopixel.NeoPixel(board.NEOPIXEL, 1)
led.brightness = 0.1
last_check_value = check_pin.value
uart2_dir = DigitalInOut(board.IO16)
uart2_dir.direction = Direction.OUTPUT
uart2_dir.value = 0
times_1 = 0
times_2 = 0
loop_count = 0
while True:
if check_pin.value:
if uart1 == None:
print("*** initialize the uarts ***")
# we have not yet tried to iniatialize the uarts
uart1 = board.UART()
uart1.deinit()
uart1 = busio.UART(board.TX, board.RX, rs485_dir= board.IO1, baudrate=1000000)
# on second one roll our own direction pin
uart2 = busio.UART(board.IO17, board.IO18, baudrate=1000000)
if uart1 != None:
start_time = time.monotonic_ns()
uart1.write(b"abcdefghijklmnopqrstuvwxyz")
after1 = time.monotonic_ns()
times_1 += int(after1 - start_time)
uart2_dir.value = 1
uart2.write(b"abcdefghijklmnopqrstuvwxyz")
uart2_dir.value = 0
after2 = time.monotonic_ns()
times_2 += int(after2 - after1)
loop_count += 1
if (loop_count == 10):
print(times_1/10, " ", times_2/10)
times_1 = 0
times_2 = 0
loop_count = 0
# cycle colors
led[0] = led_colors[color_index]
color_index += 1
if color_index >= len(led_colors):
color_index = 0
time.sleep(0.1)
And so far it looks like the option is slower... Or maybe there is a difference in these two UARTS?
Adafruit CircuitPython 7.3.0-beta.0-2-g87e59a444 on 2022-04-02; ESP32-S3-DevKitC-1-N8R8 with ESP32S3
*** initialize the uarts ***
866699.0 686646.0
860596.0 674437.0
869752.0 680540.0
851443.0 677490.0
897217.0 729371.0
857545.0 674437.0
860594.0 680543.0
854492.0 686647.0
863647.0 680541.0
Which is also shown differences in Logic Analyzer output
But assuming the answer is, yes, still worth implementing. Question to self (and others) In the port code:
void common_hal_busio_uart_construct(busio_uart_obj_t *self,
const mcu_pin_obj_t *tx, const mcu_pin_obj_t *rx,
const mcu_pin_obj_t *rts, const mcu_pin_obj_t *cts,
const mcu_pin_obj_t *rs485_dir, bool rs485_invert,
uint32_t baudrate, uint8_t bits, busio_uart_parity_t parity, uint8_t stop,
mp_float_t timeout, uint16_t receiver_buffer_size, byte *receiver_buffer,
bool sigint_enabled) {
self->baudrate = baudrate;
self->character_bits = bits;
self->timeout_ms = timeout * 1000;
if (self->character_bits != 7 && self->character_bits != 8) {
mp_raise_ValueError(translate("Invalid word/bit length"));
}
mp_printf(&mp_plat_print, "uart_construct: tx:%p rx:%p rts:%p cts:%p rs485:%p\n", tx, rx, rts, cts, rs485_dir);
Is there a proper/preferred way in the HAL code to convert/use the parameter: const mcu_pin_obj_t *rs485_dir
as a standard output (DigitalInOut) object set to output....
if so then can probably easily implement the code in:
// Write characters.
size_t common_hal_busio_uart_write(busio_uart_obj_t *self, const uint8_t *data, size_t len, int *errcode) {
if (self->tx == NULL) {
mp_raise_ValueError(translate("No TX pin"));
}
LPUART_WriteBlocking(self->uart, data, len);
return len;
}
Where before calling the WriteBlocking we see that we are setup for rs485_dir pin and set the correct value
Then after the return, we probably have to loop calling the get status method, until the transmit complete flag is set, at which point we would set the pin status back to not writing state.
Now back to investigating...
we want to mimic pyserial's behavior so if you implement non-blocking write, please mimic the API as well! https://stackoverflow.com/questions/25424056/pyserial-non-blocking-write
we want to mimic pyserial's behavior so if you implement non-blocking write, please mimic the API as well! https://stackoverflow.com/questions/25424056/pyserial-non-blocking-write
I will be honest and say again I am not a python developer. Nor had great experiences with PySerial... But maybe it has improved since then.
I was looking at the documents at: https://pyserial.readthedocs.io/en/latest/pyserial_api.html#serial.Serial.write And saw that it then had methods for flush() and out_waiting...
As for non-blocking writes, that is more of an issue for you all to decide on, as that if it were to be implemented it should apply to all (or at least most) of the ports.
Unless explicitly labelled otherwise (like audio playing), writes to various I/O devices are blocking in CircuitPython. We don't have non-blocking writes in the sense of writes happening the background for UART, I2C, SPI, etc.
We are thinking about non-blocking I/O operations using async/await and asyncio. This provides a nice framework for handling completion, etc., as opposed to explicit polling of completion.
(While composing this, I thought I would test and describe pyserial's behavior, but it's not that intuitive: I tested it with a low baudrate and large number of bytes with .write_timeout
being None, 0, and non-zero. Depending on whether all the bytes can be sent or not, various things happen, including exceptions.)
Thanks @dhalbert,
As for how Pyserial works, again I am a novice. My first usages of it was for ROS and did not have great experiences with it... So rewrote the couple of objects into c++...
So I don't understand the relationship with their asynchronous serial transfers and timeouts, if there is no hardware flow control involved. Than the timing should be pretty fixed... Time per character * number of characters.
One thing I noticed in this code and with IMXRT RM is there is some support for RS485 support by using the RTS pin. But none of the Teensy boards actually have exposed any RTS pins... The Arduino code handles RTS and RS485 direction pins through simple software...
I will still experiment with enabling it on normal GPIO pins.
At least I know understand the error message when I tried it: ValueError: Invalid RTS pin
The code sets the RTS variable to what is defined in the RS485 pin variable and it does not find it in the list.... And if not in the list of RTS pins it prints out error message. Probably without allowing other pins to work the current code should detect that you are wanting to use CTS pin for RS485 and change the error message.
I have the first part of the code in place, and at times trying to figure out why board resets.... Maybe it was sitting on a couple of stray wires...
I was again curious with two things with the code: a) is it faster than just rolling your own? dir_pin.value = 1, do your writes dir_pin.value= 0 b) I wondered if the code in a) would work right anyway...
Short answer is a) may be similar speed wise, but both currently have problems... I will now fix the rs485_dir version. picture worth...
Problem is the write code does not return, until the last byte was pushed into the output queue of the processor and returns before the queue is empty and transmit complete flag is set. Will add that check into the code I am adding.
Again you may want to consider adding a flush() member (like PySerial and Arduino) for allowing those who roll their own to work like a)
Forgot to mention, if you are curious of the python code I am running:
import time
import board
import busio
import supervisor
from digitalio import DigitalInOut, Direction, Pull
initial_baud = 115200
if hasattr(board, 'NEOPIXEL'):
import neopixel
led_colors = [
(255, 0, 0),
(0, 255, 0),
(0, 0, 255),
(255, 255, 0),
(255, 0, 255),
(0, 255, 255),
(255, 255, 255),
(0, 0, 0),
]
color_index = 0
led = neopixel.NeoPixel(board.NEOPIXEL, 1)
led.brightness = 0.3
else:
led = DigitalInOut(board.LED)
led.direction = Direction.OUTPUT
led_colors = None
if hasattr(board, 'TX2'):
dir1_pin = board.D2
dir2_pin = board.D3
tx2 = board.TX2
rx2 = board.RX2
else:
#ESP32-S3
dir1_pin = board.IO1
dir2_pin = board.IO16
tx2 = board.IO17
rx2 = board.IO18
uart1 = board.UART()
uart2 = None
#def setup_uarts(baud):
if True:
baud = initial_baud
print("*** initialize the uarts baud: ", baud, " ***")
# we have not yet tried to iniatialize the uarts
if (uart1 != None):
uart1.deinit()
uart1 = busio.UART(board.TX, board.RX, rs485_dir=dir1_pin, baudrate=baud)
print("After Uart1")
# on second one roll our own direction pin
if (uart2 != None):
uart2.deinit()
uart2 = busio.UART(tx2, rx2, baudrate=baud)
print("After Uart2")
#setup_uarts(initial_baud)
uart2_dir = DigitalInOut(dir2_pin)
uart2_dir.direction = Direction.OUTPUT
uart2_dir.value = 0
times_1 = 0
times_2 = 0
loop_count = 0
print("Start loop")
while True:
start_time = time.monotonic_ns()
uart1.write(b"abcdefghijklmnopqrstuvwxyz")
after1 = time.monotonic_ns()
times_1 += int(after1 - start_time)
uart2_dir.value = 1
uart2.write(b"abcdefghijklmnopqrstuvwxyz")
uart2_dir.value = 0
after2 = time.monotonic_ns()
times_2 += int(after2 - after1)
loop_count += 1
if (loop_count == 10):
print(int(times_1/10), " ", int(times_2/10))
times_1 = 0
times_2 = 0
loop_count = 0
if led_colors != None:
led[0] = led_colors[color_index]
color_index += 1
if color_index >= len(led_colors):
color_index = 0
else:
if led.value:
led.value = False
else:
led.value = True
time.sleep(0.025)
quick update: I updated the write code like:
size_t common_hal_busio_uart_write(busio_uart_obj_t *self, const uint8_t *data, size_t len, int *errcode) {
if (self->tx == NULL) {
mp_raise_ValueError(translate("No TX pin"));
}
if (self->rs485_dir && len) {
GPIO_PinWrite(self->rs485_dir->gpio, self->rs485_dir->number, !self->rs485_invert);
LPUART_WriteBlocking(self->uart, data, len);
// Probably need to verify we have completed output.
uint32_t dont_hang_count = 0xffff;
while (dont_hang_count--) {
if (LPUART_GetStatusFlags(self->uart) & kLPUART_TransmissionCompleteFlag) {
break; // hardware says it completed.
}
}
GPIO_PinWrite(self->rs485_dir->gpio, self->rs485_dir->number, self->rs485_invert);
} else {
// could combine with above but would go through two ifs
LPUART_WriteBlocking(self->uart, data, len);
}
return len;
}
I will probably make the loop count for not hanging even bigger, should never happen. but now the output:
CircuitPython version
Code/REPL
Behavior
uart_construct: tx:0x6007e45c rx:0x6007e480 rts:0x0 cts:0x0 rs485:0x6007eda4 Traceback (most recent call last): File "", line 1, in
ValueError: Invalid RTS pin
Description
Note: The first line above in is debug line I added to verify that the rs485_dir pin was passed through. Was NULL when I tried without specifying the pin.
Would be great if it actually handles the direction pin...
Note: on STM32F405 it gives different error, where it says that it is not implemented.
Additional information
No response