Using DMA and PIO for IRIG.B122 timecode generation

[fmafma]

Hi!

I would like to generate a French AFNOR NF S87-500 timecode, which is very close to the IRIG.B122. It needs a 1kHz sine carrier, each bit being 10 periods, with amplitude changing at zero crossing over the 10 periods depending on the bit value:

I plan to implement this using DDS, with a RP2040, in Micropython. My idea is to use some DMA, to feed a PIO with either a DAC value + R2R converter, or a PWM value + lowpass filter.

I first tried to use a single DMA to feed a PWM, which controls the brightness of the builtin LED of a RPi Pico. This code works fine:

import math
import array
import uctypes
import machine
import rp2

CARRIER_FREQ = 2  # Hz
NB_SAMPLES_PER_SINE = 8192
NB_SINES_PER_BIT = 2
PWM_FREQ = CARRIER_FREQ * NB_SAMPLES_PER_SINE  # Hz

# Values are set for that specific PWM_FREQ, and GPIO25
CARRIER  = [round(0xee6+0xee5*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
WAVE = array.array('I')
for b in NB_SINES_PER_BIT*CARRIER:
    WAVE.append(b << 16)


class Test:
    def __init__(self, pwmPin):
        self._dmaPwm = rp2.DMA()
        self._dmaPwm.irq(self._dmaPwmIsr, hard=True)

        self._pwm = machine.PWM(pwmPin, freq=PWM_FREQ, duty_u16=0)

    def _dmaPwmIsr(self, dma):
        print("dmaPwm ended at", dma.read)

    def start(self):
        dmaPwmCtrl = self._dmaPwm.pack_ctrl(size=2,
                                            inc_write=False,
                                            treq_sel=0x1c,  # DREQ_PWM_WRAP4
                                            irq_quiet=False,
                                            bswap=False)
        self._dmaPwm.config(read=WAVE,
                            write=0x40050000+0x5c,  # CH4_CC (PWM4 Counter Comparator)
                            count=NB_SAMPLES_PER_SINE*NB_SINES_PER_BIT,
                            ctrl=dmaPwmCtrl
        )

        self._dmaPwm.active(True)  # start DMA


irigb122 = Test(machine.Pin(25, machine.Pin.OUT))
irigb122.start()

Then, I used a second DMA to feed the first one with different waveforms. Which does not work:

import math
import array
import uctypes
import machine
import rp2

CARRIER_FREQ = 2  # Hz
NB_SAMPLES_PER_SINE = 8192
NB_SINES_PER_BIT = 2
PWM_FREQ = CARRIER_FREQ * NB_SAMPLES_PER_SINE  # Hz

# Values are set for that specific PWM_FREQ, and GPIO25
CARRIER  = [round(0xee6+0xee5*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_1_WAVE = array.array('I')
for b in CARRIER:
    BIT_1_WAVE.append(b << 16)

CARRIER  = [round(0xee6+0x3b9*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_0_WAVE = array.array('I')
for b in CARRIER:
     BIT_0_WAVE.append(b << 16)

print(f"BIT_1_WAVE address: {uctypes.addressof(BIT_1_WAVE)}")
print(f"BIT_0_WAVE address: {uctypes.addressof(BIT_0_WAVE)}")


class Test:
    def __init__(self, pwmPin):
        self._dmaFrame = rp2.DMA()
        self._dmaFrame.irq(self._dmaFrameIsr, hard=True)
        self._dmaPwm = rp2.DMA()
        self._dmaPwm.irq(self._dmaPwmIsr, hard=True)

        self._pwm = machine.PWM(pwmPin, freq=PWM_FREQ, duty_u16=0)

    def _dmaFrameIsr(self, dma):
        print("dmaFrame ended at", dma.read)

    def _dmaPwmIsr(self, dma):
        print("dmaPwm ended at", dma.read)

    def start(self, timecode):
        # Built the frame with bits waves addresses _dmaPwm need to send to PWM
        frame = array.array('I')
        for b in timecode:
            if b == '0':
                frame.append(uctypes.addressof(BIT_0_WAVE))
            elif b == '1':
                frame.append(uctypes.addressof(BIT_1_WAVE))
        frame.append(0)  # mark end of frame
        print(f"frame address: {uctypes.addressof(frame)}")

        # _dmaFrame reads from frame and writes each bit wave address in _dmaPwm READ_ADD_TRIG
        dmaFrameCtrl = self._dmaFrame.pack_ctrl(size=2,  # a transfert is 4 bytes width
                                                inc_write=False,  # always write to the same address (READ_ADD_TRIG)
                                                irq_quiet=False,  # we want interrupt at end of transfert
                                                bswap=False)  # no byte swap
        self._dmaFrame.config(read=frame,  # source
                              write=self._dmaPwm.registers[15],  # _dmaPwm read address (READ_ADD_TRIG)
                              count=1,  # 1 transfert
                              ctrl=dmaFrameCtrl)

        # _dmaPwm reads from a bit wave which address is given by _dmaFrame, and writes to PWM4 Counter Comp)
        dmaPwmCtrl = self._dmaPwm.pack_ctrl(size=2,  # a transfert is 4 bytes width
                                            inc_write=False,  # alsways write to the same address (CH4_CC)
                                            chain_to=self._dmaFrame.channel,  # trigger _dmaFrame at the end of transfert
                                            treq_sel=0x1c,  # transfert is 'clocked' by the PWM4 counter rollover (DREQ_PWM_WRAP4 )
                                            irq_quiet=False,  # we want interrupt at end of transfert
                                            bswap=False)  # no byte swap
        self._dmaPwm.config(write=0x40050000+0x5c,  # PWM4 Counter Compare register address (CH4_CC)
                            count=NB_SINES_PER_BIT*NB_SAMPLES_PER_SINE,  # number of transferts
                            ctrl=dmaPwmCtrl
        )

        # Start _dmaFrame which should feed _dmaPwm with read address. Which does not seem to work
        self._dmaFrame.active(True)


irigb122 = Test(machine.Pin(25, machine.Pin.OUT))
irigb122.start("0101")

This code is close to the Micropython example on the DMA doc page.
If I monitor the ctrl registers, I can see that the dmaPwm has a read error, and the dmaFrame has a write error.
So, it seems that the dmaFrame is unable to feed the dmaPwm.

Any idea why?

[sk8board]

I have some suggestions that might help.

1. could you format your code as shown in my screenshot below? This will help others better read your code.

2. could you add comments to your code to help others better understand your code?

3. here is an example that I created of how to chain two DMA's together.

[fmafma]

1 & 2 done (let me know if you need more comments)

3 in your example, DMAs feed PIOs in ping-pong mode. Here, the first DMA is chained to the second, but the second feeds the triggered read address of the first one, and does not use the chain mecanism.

My code is based on the Micropython example, which works fine. The only difference is I feed the second DMA with only 1 word (READ_ADD_TRIG) instead of 2. So, I don't use the wrap mecanism, and set inc_write to False. I tried using the wrap mecanism, with 4 bytes rollover, but it does not work either.

I tried the other way, giving an initial read address to _dmaPwm, and start this DMA instead of _dmaFrame: I get a first transfert (complete bit wave), but then it stops. I think the problem is feeding READ_ADD_TRIG, which does not trig _dmaPwm.

I may not use the Micropython DMA implementation the right way, but I don't see why.

[sk8board]

Thank you for the help. I have a question. If your program works with one DMA, then why do you want to use two DMA's?

If you need to use two DMA's then I believe you have two problems:

1. Only one of the DMA's is set to active state. You will need to set both active.

2. I believe the first DMA is not being initially triggered to start the chain.

[fmafma]

My example with one DMA only send 1 bit wave. I need to send 100 bits, either 0, 1 or marker. They all have differente waveform (0 and 1 bits are shown in the above picture). The idea is to use a second DMA to feed the first one with each bit of the timecode frame.

Both DMA are enabled, and only the master DMA (_dmaFrame) needs to be triggered. It should then starts to feed the _dmaPwm by writing in its READ_ADD_TRIG register. This is how it is done in the Micropython DMA example.

I made many many tests, by giving an inital read address to dmaPwm or not, activate both of them or only one (trying each one), etc.. but at best, I only get one wave out; either the chain_to or the READ_ADD_TRIG does not work.

I would like to find someone able to write this code in C, using the Pico SDK, to see how it should be done; there may also be a Micropython problem, I can't say. All I know, is I don't have anymore ideas. I spent the last 5 days on this problem...

[sk8board]

Both DMA are enabled, and only the master DMA (_dmaFrame) needs to be triggered. It should then starts to feed the _dmaPwm by writing in its READ_ADD_TRIG register. This is how it is done in the Micropython DMA example.

I made many many tests, by giving an inital read address to dmaPwm or not, activate both of them or only one (trying each one), etc.. but at best, I only get one wave out; either the chain_to or the READ_ADD_TRIG does not work.

You provided two examples. The second example only activates one DMA. This is why I recommended to activate both DMA's

I would like to find someone able to write this code in C, using the Pico SDK, to see how it should be done; there may also be a Micropython problem, I can't say. All I know, is I don't have anymore ideas. I spent the last 5 days on this problem...

I can understand your frustration. MicroPython is very versatile, but the state of documentation is less than desirable, which makes the learning curve unnecessarily difficult.

Here is an example using C.

Here is an example using MicroPython Viper which helps understand how DMA configuration is performed at a bit level. Here is the initial trigger to start the chain.

I also recommend checking DMA2 and DMA3 in my MicroPython example. Here is how I start the initial trigger of the chain by completing configuration of DMA after setting the DMA active.

[fmafma]

According to the RP240 documentation, Writing to the READ_ADDR_TRIG register should trigger the DMA. If not, I don't see the point of these _TRIG registers. And as said, the Micropython DMA example works fine, but it only uses memory to memory transfert, not memory to PWM.

I add a quick look at the C code, and it does not use the same technic as described in the Micropython DMA example. Instead of writing to the READ_ADDR_TRIG register, it writes to the normal READ_ADDR register, and both DMA are chained to each other. I made a test, and the code does more things, there are additionnal reconfigurations, but still not working correctly. I will dig further in both C and viper codes.

Thanks for the links!

PS: BTW, even in the C code, only one DMA is triggered (line 147). But the one corresponding to my _dmaPwm!

[sk8board]

Note: the C example and my example both use two DMA channels chained to each other, which causes an infinite loop of DMA's triggering each other.

The Viper example chains one DMA channel like you have in your example.

[fmafma]

I guess there is an inifinite loop only if all inc_xxx flags are False? If say inc_read is True, once the read address becomes nul, it should stop?

[sk8board]

If your DMA config does not ever change, then you would chain both DMA's together for an infinite loop. This is what I did in my example.

If one of the DMA's needs the config to change, (read, write, or trans count changes after a transfer is complete) then chain one DMA to reconfigure the other. The reconfigure DMA will write the new config to the DMA that needs restarted. Writing the new config will trigger the updated DMA to restart the chain. This is what was done in the Viper example.

[sk8board]

I believe my last edit to my last post should provide the missing information that you need.

[fmafma]

What you describe is exactly what I thought I did... I probably missed something; I will study the viper example this week-end.

[sk8board]

Try this.

import math
import array
import uctypes
import machine
import rp2
import time

CARRIER_FREQ = 2  # Hz
NB_SAMPLES_PER_SINE = 8192
NB_SINES_PER_BIT = 2
PWM_FREQ = CARRIER_FREQ * NB_SAMPLES_PER_SINE  # Hz

# Values are set for that specific PWM_FREQ, and GPIO25
CARRIER  = [round(0xee6+0xee5*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_1_WAVE = array.array('L')
for b in CARRIER:
    BIT_1_WAVE.append(b << 16)

CARRIER  = [round(0xee6+0x3b9*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_0_WAVE = array.array('L')
for b in CARRIER:
     BIT_0_WAVE.append(b << 16)

print(f"BIT_1_WAVE address: {hex(uctypes.addressof(BIT_1_WAVE))}")
print(f"BIT_0_WAVE address: {hex(uctypes.addressof(BIT_0_WAVE))}")

@micropython.viper
def configure_DMAs(nword:int, H_buffer_line_add:ptr32):
    IRQ_QUIET = 0  # Do not generate an interrupt
    RING_SEL = 0   # No wrapping
    RING_SIZE = 0  # No wrapping
    HIGH_PRIORITY = 1
    INCR_WRITE = 0  # Non increment while writing

    #Setting up the "data" DMA channel 1
    TREQ_SEL = 28    #  num of PWM4
    INCR_READ = 1   # 1 increment while reading
    DATA_SIZE = 2   # 32 bit transfer
    CHAIN_TO = 0    # Chain to configure channel DMA 0 so it starts again
    EN = 1          # Channel is enabled by the configure DMA chan0
    DMA_control_word = ((IRQ_QUIET << 21) | (TREQ_SEL << 15) | (CHAIN_TO  << 11) | (RING_SEL << 10) |
                        (RING_SIZE << 9) | (INCR_WRITE << 5) | (INCR_READ << 4) | (DATA_SIZE << 2) |
                        (HIGH_PRIORITY << 1) | (EN << 0))
    ptr32(0x5000007c)[0] = 0                        # DMA Channel 1 Read Address pointer <- not important because reset by DMA0 "configure" channel
    ptr32(0x50000074)[0] = uint(0x40050000+0x5c)         # DMA Channel 1 Write Address pointer -> PWM4 address
    ptr32(0x50000078)[0] = nword                    # DMA Channel 1 Transfer Count <- length of the Data array buffer
    ptr32(0x50000070)[0] = DMA_control_word         # DMA Channel 1 Control and Status (using alias to not start immediatly - will be started by DMA chanel 0)

    #Setting up the "control" DMA channel 0 - to run the Channel 1 - Vertical Visible Area lines 
    TREQ_SEL = 0x3f # Max speed, however synchronization is achieved via the PIO irq 1
    INCR_READ = 1   # No increment while reading
    CHAIN_TO = 0    # chain to itself (no chaining)
    EN = 1          # Start channel upon setting the trigger register
    DMA_control_word = ((IRQ_QUIET << 21) | (TREQ_SEL << 15) | (CHAIN_TO  << 11) | (RING_SEL << 10) |
                        (RING_SIZE << 9) | (INCR_WRITE << 5) | (INCR_READ << 4) | (DATA_SIZE << 2) |
                        (HIGH_PRIORITY << 1) | (EN << 0))
    ptr32(0x50000000)[0] = uint(H_buffer_line_add)       # DMA Channel 0 Read Address pointer <- data array to reconfigure DMA1
    ptr32(0x50000004)[0] = uint(0x5000007c)              # DMA Channel 0 Write Address pointer -> DMA1 read_adress alias register 3 (CH1_AL3_READ_ADDR_TRIG ) - trigger the DMA1 start
    ptr32(0x50000008)[0] = 1                             # DMA Channel 0 Transfer Count <- Just one data (long) array to transfer continuously
    ptr32(0x50000010)[0] = DMA_control_word              # DMA Channel 0 Control and Status (using alias to not start immediatly - will be started by DMA trigger register)

    ptr32(0x50000430)[0] |= 0b0001  #triggers DMA chan0

@micropython.viper
def stopsync():
    ptr32(0x50000444)[0] |= 0b000011         # Aborts DMA chan0 and 1

class Test:
    def __init__(self, pwmPin):
        self._dmaFrame = rp2.DMA()
        self._dmaFrame.irq(self._dmaFrameIsr, hard=False)
        self._dmaPwm = rp2.DMA()
        self._dmaPwm.irq(self._dmaPwmIsr, hard=False)

        self._pwm = machine.PWM(pwmPin, freq=PWM_FREQ, duty_u16=0)

    def _dmaFrameIsr(self, dma):
        print("dmaFrame ended at", dma.read)

    def _dmaPwmIsr(self, dma):
        print("dmaPwm ended at", dma.read)

    def start(self, timecode):
        # Built the frame with bits waves addresses _dmaPwm need to send to PWM
        frame = array.array('L')
        for b in timecode:
            if b == '0':
                frame.append(uctypes.addressof(BIT_0_WAVE))
            elif b == '1':
                frame.append(uctypes.addressof(BIT_1_WAVE))
        frame.append(0)  # mark end of frame
        
        dma_transfers = NB_SINES_PER_BIT*NB_SAMPLES_PER_SINE
        
        configure_DMAs(dma_transfers,frame)

irigb122 = Test(machine.Pin(25, machine.Pin.OUT))
irigb122.start("0101")
time.sleep(5)
stopsync()

[sk8board]

I believe the example from the MicroPython doc uses DMA.registers incorrectly. I created an issue to have the example corrected.

[sk8board]

I have verified DMA.registers was incorrectly used in the example.

This works:

        self._dmaFrame.config(read=frame,  # source
                              #write=self._dmaPwm.registers[15],  # example incorrectly uses DMA.registers
                              write=uctypes.addressof(self._dmaPwm.registers) + 0x3c, # _dmaPwm read address (READ_ADD_TRIG)
                              count=1,  # 1 transfert
                              ctrl=dmaFrameCtrl,
                              trigger=1) # add an initial trigger to start the chain

[fmafma]

@sk8board thank your very much for this investigation! This is great to have something working, I will be able to go further in this project :o)

I will still continue to investigate and try to make it in pure Python, as for now, I'm unable to make it work correctly. I don't think the registers is the only issue (in my previous tests, I did try to directly use the register address as uint, with no success).

And if registers are not what they are supposed to be, how can the provided example work? I'm confused.

[sk8board]

I will still continue to investigate and try to make it in pure Python, as for now, I'm unable to make it work correctly. I don't think the registers is the only issue (in my previous tests, I did try to directly use the register address as uint, with no success).

uctypes.addressof(self._dmaPwm.registers) will return the address of the DMA channel. For example, if the DMA channel is channel 0, then the returned address will be 0x50000000, as shown in section 2.5.7 of the RP2040 Datasheet.

If you want to write to READ_ADD_TRIG, then you need to add 0x3c to the returned address as shown here.

        self._dmaFrame.config(read=frame,  # source
                              #write=self._dmaPwm.registers[15],  # example incorrectly uses DMA.registers
                              write=uctypes.addressof(self._dmaPwm.registers) + 0x3c, # _dmaPwm read address (READ_ADD_TRIG)
                              count=1,  # 1 transfert
                              ctrl=dmaFrameCtrl,
                              trigger=1) # add an initial trigger to start the chain

You also need to add an initial trigger to start the chain with, trigger=1.

I simply copied your code from your initial post and replaced the _dmaFrame.config function with this code and it worked.

To avoid conflict between the two interrupts, you will need to change at least one of the interrupts to hard=False.

If you want the code to work a second time, then you will need to add code to deactivate the DMA channels DMA.active(0) or perform a reset machine.reset() to stop the DMA channel before restarting your code.

[fmafma]

Isn't dma.active(True) call the same as giving the trigger=True parameter in the config() method?

In my real code, I use a try/finally to call DMA.active(False) in order to reset it. In fact, I also do it right after creating the DMA...

[sk8board]

Try this:

import math
import array
import uctypes
import machine
import rp2
import time

CARRIER_FREQ = 2  # Hz
NB_SAMPLES_PER_SINE = 8192
NB_SINES_PER_BIT = 2
PWM_FREQ = CARRIER_FREQ * NB_SAMPLES_PER_SINE  # Hz

# Values are set for that specific PWM_FREQ, and GPIO25
CARRIER  = [round(0xee6+0xee5*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_1_WAVE = array.array('I')
for b in CARRIER:
    BIT_1_WAVE.append(b << 16)

CARRIER  = [round(0xee6+0x3b9*math.sin(2*math.pi*t/NB_SAMPLES_PER_SINE)) for t in range(NB_SAMPLES_PER_SINE)]
BIT_0_WAVE = array.array('I')
for b in CARRIER:
     BIT_0_WAVE.append(b << 16)

print(f"BIT_1_WAVE address: {uctypes.addressof(BIT_1_WAVE)}")
print(f"BIT_0_WAVE address: {uctypes.addressof(BIT_0_WAVE)}")


class Test:
    def __init__(self, pwmPin):
        self._dmaFrame = rp2.DMA()
        self._dmaFrame.irq(self._dmaFrameIsr, hard=False)
        self._dmaPwm = rp2.DMA()
        self._dmaPwm.irq(self._dmaPwmIsr, hard=False)

        self._pwm = machine.PWM(pwmPin, freq=PWM_FREQ, duty_u16=0)

    def _dmaFrameIsr(self, dma):
        print("dmaFrame ended at", dma.read)

    def _dmaPwmIsr(self, dma):
        print("dmaPwm ended at", dma.read)

    def start(self, timecode):
        # Built the frame with bits waves addresses _dmaPwm need to send to PWM
        frame = array.array('I')
        for b in timecode:
            if b == '0':
                frame.append(uctypes.addressof(BIT_0_WAVE))
            elif b == '1':
                frame.append(uctypes.addressof(BIT_1_WAVE))
        frame.append(0)  # mark end of frame
        print(f"frame address: {uctypes.addressof(frame)}")

        # _dmaFrame reads from frame and writes each bit wave address in _dmaPwm READ_ADD_TRIG
        dmaFrameCtrl = self._dmaFrame.pack_ctrl(size=2,  # a transfert is 4 bytes width
                                                inc_write=False,  # always write to the same address (READ_ADD_TRIG)
                                                irq_quiet=False,  # we want interrupt at end of transfert
                                                bswap=False)  # no byte swap
        self._dmaFrame.config(read=frame,  # source
                              write=uctypes.addressof(self._dmaPwm.registers) + 0x3c,  # _dmaPwm read address (READ_ADD_TRIG)
                              count=1,  # 1 transfert
                              ctrl=dmaFrameCtrl,
                              trigger=1)

        # _dmaPwm reads from a bit wave which address is given by _dmaFrame, and writes to PWM4 Counter Comp)
        dmaPwmCtrl = self._dmaPwm.pack_ctrl(size=2,  # a transfert is 4 bytes width
                                            inc_write=False,  # alsways write to the same address (CH4_CC)
                                            chain_to=self._dmaFrame.channel,  # trigger _dmaFrame at the end of transfert
                                            treq_sel=0x1c,  # transfert is 'clocked' by the PWM4 counter rollover (DREQ_PWM_WRAP4 )
                                            irq_quiet=False,  # we want interrupt at end of transfert
                                            bswap=False)  # no byte swap
        self._dmaPwm.config(write=0x40050000+0x5c,  # PWM4 Counter Compare register address (CH4_CC)
                            count=NB_SINES_PER_BIT*NB_SAMPLES_PER_SINE,  # number of transferts
                            ctrl=dmaPwmCtrl
        )

        # Start _dmaFrame which should feed _dmaPwm with read address. Which does not seem to work
        self._dmaFrame.active(True)


irigb122 = Test(machine.Pin(25, machine.Pin.OUT))
irigb122.start("0101")
time.sleep(5)
irigb122._dmaFrame.active(False)
irigb122._dmaPwm.active(False)
irigb122._dmaFrame.close()
irigb122._dmaPwm.close()

[fmafma]

Thanks! I finally got it running :o)

Just for fun, I will also do the DAC version...

[fmafma]

Still using the viper implementation, but it works great!

[sk8board]

Good to hear. Thanks for sharing the picture.

[fmafma]

Ok, now, I would like to continuously send the frame, without breaking the sine wave.

My first idea was to use another DMA, in ping-pong mode with the frame DMA, which feeds the wave DMA.

But the problem is the wave DMA is chained to the frame DMA, so a new wave address can be loaded as soon as it has finished to output the previous wave. The issue with the ping-pong config is it would require to change the chained DMA to alternatly point to the ping frame DMA, or the pong frame DMA.

Any idea if that is possible? Or do you see another way to do that?

[sk8board]

I am not sure I understand. What do you mean by breaking the sine wave? What is the source of the continuous input? How does the RP2040 receive the input?

[sk8board]

If I understand you correctly, I believe you want to use something like the example that is from the MicroPython documentation (RP2040 Datasheet). The gather_list array is a 2D array that has the address and length of transfers that are used to reconfigure both address and transfer count of the other DMA. Using a 2D array allows for complex and continuous transfers with data from different locations.

[fmafma]

Yes, that exaclty what I do (I only need a 1D array, as size of datas is constant, is an address).
The issue is to restart the entire process once done, without delay...

[sk8board]

I suspect you already know the answer. If not, then I apologize for my suspicion. I believe you need the two data buffers, Frame1 and Frame2. Read from one while writing to the other, then alternate when reading has completed.

Please let the investigator know that his collusion with my prior employer is corrupt and I would like the four year abuse of power to stop.

[fmafma]

Even with 2 buffers, how do I switch from one to another? I made some tests, and chaining the frame DMAs doesn't work: as they only send 1 data at a time (address of the next wave for the wave DMA), they ping-pong immediatly, instead of the end of the entire frame!

[fmafma]

It's the same process as before: one DMA feeds the PWM with the spécific sine wave (which is made of 10 periods with different amplitudes, depending on the bit value), and another DMA feeds the first DMA with the different sine waves addresses.

When the second DMA is over (ie has sent all the sine waves addresses matching a complete IRIG.B frame), I need to start over, without glitch in the sine carrier.

My problem is to trigger something very quickly when the second DMA is over, to start it again. I dont't think it can be done with an interruption and Python code...

[fmafma]

The frame is the time. I need to send 1 frame per second, incrementing the time by 1s between frames.
I don't know yet how I will manage the time, but I think using the internal RTC is good enough over a day, and I can resync at midnight on a ntp server.

[fmafma]

A frame is 100 bits long, but only half are used in my case.
So, to update the frame, I planned to monitor the read address of the second DMA; once it has sent the required bits, I can update them; I have more than half a second to do that.

But I do need to restart the DMA precisly once it has sent a complete frame.

[sk8board]

How did you create the sine wave signal from your code? My oscilloscope shows a very different signal when using your code. Do you have passive electronic components conditioning the signal?

[fmafma]

Yes, you need a low pass filter, to remove the PWM frequency (16kHz, in the above example).

I used a simple passive RC filter:

GPIO ------/\/\/\----+--- OUT
              R      |
                   -----
                   ----- C
                     |
                     |
                    ---
                     -  GND

You can use R=1k and C=100nF (cutoff freq. ~1,6kHz).

[fmafma]

But you may have to choose another pin than 25, the builtin LED may mess up the filter (or not, I don't know).
I only used the filter on my real design, using GPIO0, with a carrier frequency of 1kHz.

[fmafma]

Here is a version using interrupt to restart the frame DAM. As I expected, the carrier has some jitter...

import math
import time
import array
import uctypes
import machine

import rp2

TREQ_PERMANENT = 0x3f
READ_ADD_TRIG_REG = 15  # see RP2040 datasheet "2.5.2.1 Aliases and Triggers"

PWM_CH0_CC = 0x4005000c
PWM_CH1_CC = 0x40050020
PWM_CH2_CC = 0x40050034
PWM_CH3_CC = 0x40050048
PWM_CH4_CC = 0x4005005c
PWM_CH5_CC = 0x40050070
PWM_CH6_CC = 0x40050084
PWM_CH7_CC = 0x40050098

PWM_CHn_CC = [
    PWM_CH0_CC, PWM_CH0_CC,  # GPIO00, GPIO01
    PWM_CH1_CC, PWM_CH1_CC,
    PWM_CH2_CC, PWM_CH2_CC,
    PWM_CH3_CC, PWM_CH3_CC,
    PWM_CH4_CC, PWM_CH4_CC,
    PWM_CH5_CC, PWM_CH5_CC,
    PWM_CH6_CC, PWM_CH6_CC,
    PWM_CH7_CC, PWM_CH7_CC,
    PWM_CH0_CC, PWM_CH0_CC,
    PWM_CH1_CC, PWM_CH1_CC,
    PWM_CH2_CC, PWM_CH2_CC,
    PWM_CH3_CC, PWM_CH3_CC,
    PWM_CH4_CC, PWM_CH4_CC,
    PWM_CH5_CC, PWM_CH5_CC,
    PWM_CH6_CC, PWM_CH6_CC   # GPIO28, GPIO29
]

DREQ_PWM_WRAP0 = 0x18
DREQ_PWM_WRAP1 = 0x19
DREQ_PWM_WRAP2 = 0x1a
DREQ_PWM_WRAP3 = 0x1b
DREQ_PWM_WRAP4 = 0x1c
DREQ_PWM_WRAP5 = 0x1d
DREQ_PWM_WRAP6 = 0x1e
DREQ_PWM_WRAP7 = 0x1f

DREQ_PWM_WRAPn = [
    DREQ_PWM_WRAP0, DREQ_PWM_WRAP0,  # GPIO00, GPIO01
    DREQ_PWM_WRAP1, DREQ_PWM_WRAP1,
    DREQ_PWM_WRAP2, DREQ_PWM_WRAP2,
    DREQ_PWM_WRAP3, DREQ_PWM_WRAP3,
    DREQ_PWM_WRAP4, DREQ_PWM_WRAP4,
    DREQ_PWM_WRAP5, DREQ_PWM_WRAP5,
    DREQ_PWM_WRAP6, DREQ_PWM_WRAP6,
    DREQ_PWM_WRAP7, DREQ_PWM_WRAP7,
    DREQ_PWM_WRAP0, DREQ_PWM_WRAP0,
    DREQ_PWM_WRAP1, DREQ_PWM_WRAP1,
    DREQ_PWM_WRAP2, DREQ_PWM_WRAP2,
    DREQ_PWM_WRAP3, DREQ_PWM_WRAP3,
    DREQ_PWM_WRAP4, DREQ_PWM_WRAP4,
    DREQ_PWM_WRAP5, DREQ_PWM_WRAP5,
    DREQ_PWM_WRAP6, DREQ_PWM_WRAP6   # GPIO28, GPIO29
]

DMA_BASE = 0x50000000

DMA_READ_ADD_TRIG = [
    DMA_BASE + 0x03c,
    DMA_BASE + 0x07c,
    DMA_BASE + 0x0bc,
    DMA_BASE + 0x0fc,
    DMA_BASE + 0x13c,
    DMA_BASE + 0x17c,
    DMA_BASE + 0x1bc,
    DMA_BASE + 0x1fc,
    DMA_BASE + 0x23c,
    DMA_BASE + 0x27c,
    DMA_BASE + 0x2bc,
    DMA_BASE + 0x2fc
]

BIT_SHIFT = [
    0, 16,  # GPIO00, GPIO01
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,
    0, 16,  # GPIO28, GPIO29
]

F_CPU                 = 125_000_000
CARRIER_FREQ          = 1_000
NB_SAMPLES_PER_PERIOD = 1024
NB_PERIODS_PER_BIT    = 1
PWM_FREQ              = CARRIER_FREQ * NB_SAMPLES_PER_PERIOD
OFFSET                = (F_CPU / PWM_FREQ) // 2
AMPLITUDE             = OFFSET - 1
PWM_PIN               = 1


class Test:
    def __init__(self, pwmPin):
        self._pwmPin = pwmPin

        self._carrierWave = array.array('I')
        for t in range(NB_PERIODS_PER_BIT*NB_SAMPLES_PER_PERIOD):
            b = round(OFFSET + AMPLITUDE * math.sin(2 * math.pi * t / NB_SAMPLES_PER_PERIOD))
            self._carrierWave.append(b << BIT_SHIFT[self._pwmPin])

        self._dmaFrame = rp2.DMA()
        self._dmaFrame.active(False)

        self._dmaWave = rp2.DMA()
        self._dmaWave.active(False)
        self._dmaWave.irq(self._dmaWaveIsr, hard=True)

        self._pwm = machine.PWM(machine.Pin(pwmPin, machine.Pin.OUT), freq=PWM_FREQ, duty_u16=0)

        self._frame = array.array('I', 3 * [uctypes.addressof(self._carrierWave)])

    def _dmaWaveIsr(self, dma):
        # self._dmaFrame.registers[READ_ADD_TRIG_REG] = self._frame  # MemoryError!
        machine.mem32[DMA_READ_ADD_TRIG[self._dmaFrame.channel]] = uctypes.addressof(self._frame)


    def run(self):
        dmaWaveCtrl = self._dmaWave.pack_ctrl(enable=True,
                                              high_pri=True,
                                              size=2,
                                              inc_read=True,
                                              inc_write=False,
                                              ring_size=0, ring_sel=False,
                                              chain_to=self._dmaFrame.channel,
                                              treq_sel=DREQ_PWM_WRAPn[self._pwmPin],
                                              irq_quiet=True,                         # int only occurs at the end of the frame (or it should!)
                                              bswap=False,
                                              sniff_en=False,
                                              write_err=False,
                                              read_err=False)

        self._dmaWave.config(write=PWM_CHn_CC[self._pwmPin],
                             count=NB_PERIODS_PER_BIT*NB_SAMPLES_PER_PERIOD,
                             ctrl=dmaWaveCtrl)

        dmaFrameCtrl = self._dmaFrame.pack_ctrl(enable=True,
                                                high_pri=True,
                                                size=2,
                                                inc_read=True,
                                                inc_write=False,
                                                ring_size=0, ring_sel=False,
                                                chain_to=self._dmaFrame.channel,  # no chaining
                                                treq_sel=TREQ_PERMANENT,          # unpaced transfert, synchronization is done by PWM wrap
                                                irq_quiet=True,
                                                bswap=False,
                                                sniff_en=False,
                                                write_err=False,
                                                read_err=False)

        self._dmaFrame.config(read=self._frame,
                              write=self._dmaWave.registers[READ_ADD_TRIG_REG:],  # must use a slice - See https://github.com/micropython/micropython/issues/16083
                              count=1,
                              ctrl=dmaFrameCtrl,
                              trigger=True)  # start DMA

        while True:
            pass

    def cancel(self):
        self._dmaFrame.active(False)
        self._dmaWave.active(False)


def test():
    print("test()")
    test = Test(PWM_PIN)
    try:
        test.run()

    finally:
        test.cancel()


if __name__ == "__main__":
    test()

[mungewell]

Very interesting... I have my own project implementing SMPTE/LTC with the PIOs, but I didn't go this deep/comlex.... well not yet :-)

[mungewell]

@fmafma I have a question about the 'amplitude' modulation scheme; is it that a 'zero' is a longer period of low level, whilst a 'one' is a short period of low level, followed by a short period of high level? Is the total length for each bit the same?

I have some ideas about using PIO to generate (not just play) the freq, if you're interested?

[mungewell]

So my ideas are pretty much the same as the PicoTimecode project; data for the frames are pre-computed and shoved into FIFO. I have a collection of state-machines to do:

• sychronization
• FIFO playout
• modulation (manchester encoding)

For IRIG the modulation would have to be adjusted, but the rest could pretty much be same/similar. For manchester it's pretty simple:
https://github.com/mungewell/pico-timecode/blob/main/pico_timecode.py#L140

If my understand is correct you want this:

And I had a stab at implementing the ASK modulator. I'll admit that it was harder than I thought, the '4 sample waveforms' ate a lot of PIO code space and the '3 phrases' didn't help (this code has ZERO spare).
irig_test.py.txt

The 'low level' is set by resistors, my little board is set up for rendering an '8 sample sine wave'. You could increase the resistance (say to 20K for each R) to lower the level to the <1/3 that you need.... see:
https://github.com/orgs/micropython/discussions/16368

This is what it produces. If it looks correct I can have a go at implementing the FIFO to actually frame up the data-stream.
000101_0079.wav.zip

[mungewell]

Had the thought that the analogue waveforms are too 'triangular', so shortened the 'zero' section. But ultimately I think a better solution would be to split the 'encode' and 'modulate' functions into 2 separate (parallel/synchronized) state machines.
irig_test.py.txt

High Cycle:
    set(pindirs, 0b11) .side(0b01) [1]		# Zero = Low, High
    set(pindirs, 0b11) .side(0b11) [3]		# High = High, High
    set(pindirs, 0b11) .side(0b01) [1]		# Zero = Low, High
    set(pindirs, 0b11) .side(0b00) [3]		# Low  = Low, Low

Looking at the waveforms in Audacity, along with Low-Pass at 2KHz with either 6dB/oct or 12dB/oct.

[fmafma]

Very interesting! I need to take time to study your code, I have a lot to learn...
Thanks!

[mungewell]

I split the encoder and ASK-modulation, and implemented a basic packing system.... seems to work nicely - but again is right at the maximum PIO code size. A RP2350 adds another block, so that might be helpful.
irig-b.py.txt

If you'd like(/if it would help) I can adjust the resistors (to lower the peaks) and make an audio recording.

[mungewell]

In case anyone is interested in a follow up, I made a project based around this concept:
https://github.com/mungewell/pico-irig