Async Morse class

[iv3wjr]

Hi to all,

developing an application for my Raspberry PICO W with no display I added visual feedback of the program behavior.

Being a ham I wrote a Morse class to flash the onboard LED (and sound optionally an 8 Ohm speaker) with a message. Now I am happy to share it with the community that helped me so many times.

The ex function is a usage example.

I use it to know when the application connects to WiFi, mounts the SD card , downloads data from the server and to monitor its correct operation.

from machine import Pin, PWM                                                    
import uasyncio as asyncio                                                      
# See:                                                                          
# https://www.embedded.com/storing-morse-code-in-c-a-comparison-of-techniques/  
# IV3WJR Roberto Vidmar, 20230130                                               
                                                                                
class AMorse:                                                                   
    code = {                                                                    
        "A": 66,  "B": 129, "C": 133, "D": 97,  "E": 32,                        
        "F": 132, "G": 99,  "H": 128, "I": 64,  "J": 142,                       
        "K": 101, "L": 130, "M": 67,  "N": 65,  "O": 103,                       
        "P": 134, "Q": 139, "R": 98,  "S": 96,  "T": 33,                        
        "U": 100, "V": 136, "W": 102, "X": 137, "Y": 141,                       
        "Z": 131, "0": 191, "1": 190, "2": 188, "3": 184,                       
        "4": 176, "5": 160, "6": 161, "7": 163, "8": 167,                       
        "9": 175, ",": 243, ".": 234, "?": 204, "/": 211,                       
        "@": 214}                                                               
                                                                                
    def __init__(self, msg="", wpm=8, led="LED", speaker=None):                 
        # Args:                                                                 
        #     msg (str): message to send                                        
        #     wpm (int): words per minute                                       
        #     led (int or "LED"): led Pin                                       
        #     speaker (int): speaker Pin                                        
        self.msg = msg                                                          
        self.led = Pin(led, Pin.OUT)                                            
        self.tdot = int(1200 / wpm)                                             
        self.tdash = self.tdot * 3                                              
        self.char_space = self.tdot * 3                                         
        self.word_space = self.tdot * 4                                         
        if speaker:                                                             
           self.speaker = PWM(speaker)                                          
           self.speaker.freq(1000)                                              
           self.speaker.duty_u16(0)                                             
        else:                                                                   
           self.speaker = None                                                  

                                                                                
    def decode(self, c):                                                        
        # Get number of symbols:                                                
        syms = c >> 5                                                           
        if syms < 6:                                                            
            mask = 31                                                           
        else:                                                                   
            syms = 6                                                            
            mask = 63                                                           
        code = c & mask                                                         
        dida = ""                                                               
        for s in range(syms):                                                   
            dida += '-' if (code % 2) else '.'                                  
            code = code >> 1                                                    
        #print("%3d %d %s" % (c, syms, dida))                                   
        return dida                                                             
                                                                                
    def append(self, msg):                                                      
        if self.msg:                                                            
            self.msg += " " + msg                                               
        else:                                                                   
            self.msg = msg                                                      
                                                                                
    async def keydown(self, t):                                                 
        self.led.high()                                                         
        if self.speaker:                                                        
            #self.speaker.duty_u16(32768)                                       
            self.speaker.duty_u16(1024)                                         
        await asyncio.sleep_ms(t)                                               
        self.led.low()                                                          
        if self.speaker:                                                        
            self.speaker.duty_u16(0)                                            
        await asyncio.sleep_ms(0)                                               
                                                                                
    async def wait(self):                                                       
        while self.msg:                                                         
            await asyncio.sleep_ms(10)                                          

                                                                               
    async def sender(self):                                                     
        self.led.low()                                                          
        while True:                                                             
            while self.msg:                                                     
                l = self.msg[0]                                                 
                if l == " ":                                                    
                    #print("space")                                             
                    await asyncio.sleep_ms(self.word_space)                     
                else:                                                           
                    dida = self.decode(self.code[l.upper()])                    
                    #print(dida)                                                
                    for e in dida:                                              
                        if e == ".":                                            
                            await self.keydown(self.tdot)                       
                        elif e == "-":                                          
                            await self.keydown(self.tdash)                      
                        await asyncio.sleep_ms(self.tdot)                       
                    await asyncio.sleep_ms(self.char_space)                     
                self.msg = self.msg[1:]                                         
            self.led.low()                                                      
            self.msg = ""                                                       
            await asyncio.sleep_ms(0)                                           
#============================================================================== 
def ex():                                                                       
    import sys                                                                  
    morse = AMorse("73 de iv3wjr", wpm=15, speaker=Pin(16))                     
                                                                                
    async def read(morse):                                                      
        while True:                                                             
            asyncio.run(morse.wait())                                           
            answ = input("Send more? ")                                         
            if answ:                                                            
                morse.append(answ)                                              
            else:                                                               
                sys.exit()                                                      
                                                                                
    async def forever():                                                        
        loop = asyncio.get_event_loop()                                         
        loop.create_task(morse.sender())                                        
        loop.create_task(read(morse))                                           
        loop.run_forever()                                                      
                                                                                
    asyncio.run(forever())                                                                                    

The speaker I use is a small 0.2 W 8 Ohm connected between GND and pin 16 on my PICO W through a 270 Ohm resistor.
speaker.duty_u16(1024) results in a weak sound, use greater values to make it louder.

Comments and suggestions are of course welcome.

73 de IV3WJR Bob

[peterhinch]

This is very nice and would be great for someone learning Morse!

I do have some comments about your use of uasyncio. Firstly there is no need to use the event loop. Secondly, calling .run() in a coroutine is incorrect because .run() is a blocking method. While this doesn't matter in your example, it's not good practice - see the tutorial.

I would rewrite your example as follows:

def ex():                                                                       
    morse = AMorse("73 de iv3wjr", wpm=15, speaker=Pin(16))                     
                                                                                
    async def read(morse):                                                      
        while True:                                                             
            await morse.wait()                                         
            answ = input("Send more? ")                                         
            if answ:                                                            
                morse.append(answ)                                              
            else:                                                               
                break
                                                                                
    async def forever():                                                        
        send_task = asyncio.create_task(morse.sender())                     
        await read(morse)
        send_task.cancel()
        await asyncio.sleep(0)  # Let cancellation occur
                                                                                
    asyncio.run(forever()) 

[iv3wjr]

Thank you Peter for your precious comment, it is for me a real pleasure to read your rewrite of my example. I promise that I will read with more attention the tutorial but reading how your code should be written is an unforgettable lesson. Really thank you for your time.

Unfortunately I noticed that in real use my AMorse.sender method does not behave as I expected if the rest of the code is not asynchronous. If the main program for example is waiting for a urequests.get call and the sender method is "sending" morse code, the timing is (obviously) incorrect. Some dots and/or dashes are longer and the resulting sound is unsatisfactory. But to my knowledge the is no way to await an urequests.get.

Working on a Raspberry PICO W however, I have two cores available so now I'm trying to run the sender method (without async) in a separate thread. I understand that uasyncio is better than _thread for many already discussed reasons but at present I have no other ideas to solve this problem.

[peterhinch]

Functions or methods that block are the enemy of uasyncio however there is a solution. The following function provides a way to delegate a blocking function to the second core. This allows a single task to wait on the function, with other uasyncio tasks running normally.

async def unblock(func, *args, **kwargs):
    def wrap(func, message, args, kwargs):
        message.set(func(*args, **kwargs))  # Run the blocking function.
    msg = Message()
    _thread.start_new_thread(wrap, (func, msg, args, kwargs))
    return await msg

See this ref for documentation.

[iv3wjr]

Great! I was trying to run a synchronous version of the sender method of my Morse class in a thread. In a simple example it seems to work but when I mix together threads and async functions in my application the PICO blocks and must be powerd off.
I will definitely try your hint.

[iv3wjr]

Unfortunately wrapping urequest.get as you suggested did not work for me. I tried to run the code three times and I got always the same result:

Unhandled exception in thread started by <function wrap at 0x20009b30>
Traceback (most recent call last):
  File "psfvg.py", line 98, in wrap
  File "urequests.py", line 180, in get
  File "urequests.py", line 163, in request
  File "urequests.py", line 93, in request
OSError: (-29312, 'MBEDTLS_ERR_SSL_CONN_EOF')

Maybe it's a PICO problem. I'm running MicroPython v1.19.1-841-g3446d440f on 2023-01-28; Raspberry Pi Pico W with RP2040.

[peterhinch]

Hmm. This seems to be a TLS/SSL issue which is well outside my comfort zone. Perhaps someone else will shed some light.