Organizing and storing arrays

[pythoncoder]

It occurred to me that my use of I2S.shift to dynamically change volume only works because I am constantly retrieving new data from a file. With a fixed sample set, shifting inside the loop will lead to a constantly changing volume. It needs to be done once:

Code: Select all

    async def play(self, fo, wname):
        #print("start")
        self.running = True
        swriter = asyncio.StreamWriter(audio_out)
        buflen = populate(fo, wname)
        I2S.shift(buf=samples_mv[:buflen], bits=16, shift=self.volume)  # memoryview into sample set
        while self.running:
            # HACK awaiting https://github.com/micropython/micropython/pull/7868
            swriter.out_buf = samples_mv[:buflen]
            await swriter.drain()

[sushyoshi]

Thanks pythoncoder.

To be honest, I am still trying to understand your code and how to schedule loops with the uasync. However your help was able to lead me in the right way. Now I just need to study about uasync more to get better.

[sushyoshi]

Hello pythoncoder,

I have been trying to understand your code and incorporate with push button inputs to change the frequency of the waves generator. For now I am still not able to change the frequency while the audio engine is working, however I am now able to use uasyncio to do other stuff at the same time.

Can you take a look at my code and see how can the play() function update the new frequency after I press a button?

Here is my play() function:

Code: Select all

playing = True
freq = 330
old_freq = 0
async def play(fo, wname):
    
    global freq
    global old_freq
    
    while playing == True:
        
        if old_freq != freq:                    
            buflen = 2 * populate(fo, wname)   # execute only when there is a frequency change
            old_freq = freq
                       
        # HACK awaiting https://github.com/micropython/micropython/pull/7868
        swriter.out_buf = samples_mv[:buflen]
        await swriter.drain()

Here is my main code with the multiple coroutines running along with the audio engine:

Code: Select all

from machine import Pin
import uasyncio as asyncio
import awg

# Settings
led = Pin(2, Pin.OUT)
btn = Pin(0, Pin.IN, Pin.PULL_UP)

ticks = 0



def button_interrupt(pin):  # Pressing the button increases the frequency +40
    global freq    
    awg.freq += 40
    print("freq = ", awg.freq)
    
btn.irq(trigger=Pin.IRQ_FALLING, handler=button_interrupt)


async def counter(period_ms):
    
    global ticks
    
    while True:
        ticks += 1
        print(ticks)
        ticks %= 16
        await asyncio.sleep_ms(period_ms)
        
async def blink(period_ms):
  
    while True:
        if led.value() == 0:
            led.value(1)       
        else:
            led.value(0)
        await asyncio.sleep_ms(period_ms)
        
async def monitor_input(period_ms): # monitors the current frequency

    while True:
        print("frequency now = ", awg.freq)
        await asyncio.sleep(period_ms)
        

        
async def main():
    
    asyncio.create_task(counter(500))
    asyncio.create_task(blink(250))
    asyncio.create_task(monitor_input(5))
    asyncio.create_task(awg.play(awg.freq, "saw"))
    
    print('Tasks are running')
    await asyncio.run_until_complete()
    
try:
    asyncio.run(main())
finally:
    awg.audio_out.deinit()
    asyncio.new_event_loop()  # Clear retained state

I am sure that looking at my code you realise I am very "green" to all this.

Overall, what I am trying to achieve is to be able to update the fo, wname with button presses and encoders. Do you think this can be achieved?