uasyncio and SIM800 UART

[gpson]

I am building a python class to interface with SIM800 module.

After looking to this example

https://github.com/peterhinch/micropyth ... r/auart.py

I came up with this (simplified for the reference) code

Code: Select all

class SIM:
	@classmethod
	def init(cls):
		cls.uart = UART(2, 9600, timeout=1)
		cls._sreader = asyncio.StreamReader(cls.uart)
		cls._swriter = asyncio.StreamWriter(cls.uart, {})
		
	@classmethod
	async def setup(cls):
		await cls.command('AT+CMGF=1')    # plain text SMS
	
	@classmethod
	async def command(cls, cmdstr, lines=1, waitfor=0.5 ):
		# empty the uart
		while True:
			chunk = yield from cls._sreader.read(1)
			if not chunk:
				break
		
		await cls._swriter.awrite("{}\n".format(message))
		await asyncio.sleep(waitfor)
		
		cls._buf = await cls._sreader.readline()
		
		if not cls._buf:
			return None
			
		result = cls._buf
		log.info("result::'{}'".format(result))
		
		if lines > 1:
			cls._savbuf = ''
			for i in range(lines):
				cls._buf = await cls._sreader.readline()
				if not cls._buf:
					break
				if not cls._buf == b'' and not cls._buf == b'OK':
					cls._savbuf += cls._buf+'\n'
				
		return result
		
# APP ----------------

sim.init()

loop = asyncio.get_event_loop()
loop.create_task(sim.setup())
loop.run_forever()

The problem is that after some AT commands it is not known how many response data will come and the StreamReader.at_eof() is not working for me.

StreamReader.readline() never times out, and the timeout setting in UART(2, 9600, timeout=1) is not working.

What is the right way to go with asyncio and UART?..

[pythoncoder]

I hope I've understood your problem correctly - apologies if I've got it wrong. But I would suggest thinking about the problem in a different way. Forget about the UART as a physical device and think in terms of messages, with timeouts being performed by the application rather than the hardware.

As I understand it you send a message, and you don't know in advance how many response messages you will receive. So you run a software retriggerable timer, started when you send the request, and retriggered each time a response has been received. If responses stop and the timer times out, you deem the process complete.

If you go that route you'll find a retriggerable timer class in aswitch.py.

[gpson]

Thank you for a good example!

I think you understand my problem correctly. One thing I do not get: should I use StreamReader.readline() or uart.readline()?

Delay_ms class has kill() and trigger() methods. Delay_ms.killer() loops and checks for deadline. The problem is that StreamReader.readline() stops the loop forever :/

[pythoncoder]

Use StreamReader.readline(). The uasyncio module enables you to abstract the UART as a data stream like a file object. So ignore the UART hardware and design your application in terms of messages to and from a stream.

Asynchronous programming requires a rather different mindset but once you've got the hang of it, it's magic

I don't understand your query about my Delay_ms class. I assume you've read the docs section 3.3. Basically you instantiate it with an optional callback. It will do nothing until triggered, when a timer is started. If you trigger it each time you receive a message, the time value will be reset to the value specified in that trigger call. So it will only time out if no message has been received in the specified period since the last one. In that instance the callback (if provided) will run and its running status will be cleared.

[gpson]

@pythoncoder check this out

https://github.com/olablt/micropython-s ... _sim800.py

What is bad and what is good? Thank you in advance.

[pythoncoder]

The first problem is that you're using self.uart. For asynchronous I/O you should be using the asyncio.StreamReader and asyncio.StreamWriter classes. Once you've instantiated those you can ignore the physical UART and just process the streams.

Methods like

Code: Select all

async def writeline(self, command):
    self.uart.write("{}\r\n".format(command))
    print("<", command)

aren't valid coroutines. A valid coroutine must have at least one await in it. The call to self.uart.write is a blocking call which you don't want in a coroutine: you want to await the stream writer. So if your class instantiates a StreamWriter as self.swriter you might have:

Code: Select all

async def writeline(self, command):
    await self.swriter.awrite("{}\r\n".format(command))
    print("<", command)

I think it would be worth your while to get more familiar with the basics of asyncio and suggest you work through my tutorial to get a handle on the fundamentals of asynchronous programming.

[pythoncoder]

Here is the sort of thing I had in mind. I can only test this with a loopback, but it seems to work OK: it times out if the loopback is removed from pins X3 and X4. As written it runs forever so you'll need to stop it with ctrl-c.

Code: Select all

from pyb import UART
import uasyncio as asyncio
import aswitch

class SIM_UART():
    def __init__(self, uart_no = 2, timeout=4000):
        self.uart = UART(uart_no, 9600)
        self.timeout = timeout
        self.loop = asyncio.get_event_loop()
        self.swriter = asyncio.StreamWriter(self.uart, {})
        self.sreader = asyncio.StreamReader(self.uart)
        self.delay = aswitch.Delay_ms()
        self.response = []
        loop = asyncio.get_event_loop()
        loop.create_task(self.recv())


    async def recv(self):
        while True:
            res = await self.sreader.readline()
            self.response.append(res)

    async def send_command(self, command):
        self.response = []
        await self.swriter.awrite("{}\r\n".format(command))
        print("<", command)
        self.delay.trigger(self.timeout)
        result = b''
        while self.delay.running():
            await asyncio.sleep(1)
            while self.response:
                result += self.response.pop(0)
                self.delay.trigger(self.timeout)  # Got something, retrigger timer
        return result

async def test():
    res = await sim_uart.send_command('rats')
    if res:
        print('Result is:' , res)
    else:
        print('Timed outwaiting for result.')

loop = asyncio.get_event_loop()
sim_uart = SIM_UART()
loop.create_task(test())
loop.run_forever()

[gpson]

Thank you! Will run you script and test it with the sim800 module.

Last year I did a small window decoration project while reading your async tutorial: https://github.com/olablt/esp8266-tree

I am more enthusiast than real programmer, but will learn to design my application in terms of messages to and from a stream!

[pythoncoder]

I did that in a hurry. On reflection this is a neater solution:

Code: Select all

from pyb import UART
import uasyncio as asyncio
import aswitch

class SIM_UART():
    def __init__(self, uart_no = 2, timeout=4000):
        self.uart = UART(uart_no, 9600)
        self.timeout = timeout
        self.loop = asyncio.get_event_loop()
        self.swriter = asyncio.StreamWriter(self.uart, {})
        self.sreader = asyncio.StreamReader(self.uart)
        self.delay = aswitch.Delay_ms()
        self.response = []
        loop = asyncio.get_event_loop()
        loop.create_task(self.recv())

    async def recv(self):
        while True:
            res = await self.sreader.readline()
            self.response.append(res)  # Append to list of lines
            self.delay.trigger(self.timeout)  # Got something, retrigger timer

    async def send_command(self, command):
        self.response = []  # Discard any pending messages
        await self.swriter.awrite("{}\r\n".format(command))
        print("<", command)
        self.delay.trigger(self.timeout)  # Re-initialise timer
        while self.delay.running():
            await asyncio.sleep(1)  # Wait for 4s after last msg received
        return b''.join(self.response)

async def test():
    res = await sim_uart.send_command('rats')
    if res:
        print('Result is:' , res)
    else:
        print('Timed out waiting for result.')

loop = asyncio.get_event_loop()
sim_uart = SIM_UART()
loop.run_until_complete(test())

[EDIT]
I figured this was a good example of a practical problem. So I created a test program auart_hd.py, included it in the repo and amended the tutorial - section 5.1.1. You might like to take a look. It emulates the hardware device so the test can be run with just two wire links.

[Jim.S]

Asynchronous programming requires a rather different mindset but once you've got the hang of it, it's magic
@pythoncoder

I would thoroughly agree with this. I am just starting to get my head around uasyncio and the asyn.Event class in particular. For me, it makes the software feel more like hardware, in that I can have a number of coroutines working in paralell, coordinated by events..... That's it! It makes the software feel a bit like gears, shafts and levers...