Code Gorgy Time

Startseite

Nachricht beantworten
Autor: Frédéric
Datum:  
To: Guilde
Betreff: Code Gorgy Time


-- 
    Frédéric

# -*- coding: utf-8 -*-

""" IRIG.B122 timecode generator.

An IRIG.B122 timecode is written over a 1KHz sinusoidal carrier.
The timecode consists of 100 bits, each bit is made of 10 carrier periods.

Bits are encoded as followed:

- a marker bit is 8 periods at large amplitude, and 2 periods at reduced (1/3) amplitude;
- a 0 bit is 2 periods at large amplitude, and 8 periods at reduced (1/3) amplitude.
- a 1 bit is 5 periods at large amplitude, and 5 periods at reduced (1/3) amplitude;
"""

import time
import math
import array
import uctypes

import rp2

from irigBTimecode import IrigBTimecode

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

CARRIER_FREQ          = 1_000  # IRIG.B carrier clock (1kHz)
NB_SAMPLES_PER_PERIOD = 256    # nb samples in a single period
NB_PERIODS_PER_BIT    = 10     # nb periods in a single IRIG.B122 bit
CYCLIC_RATIO = {               # cyclic ratio per bit type
    'm': 0.8,  # marker bit
    '0': 0.2,  # 0 bit
    '1': 0.5   # 1 bit
}
CARRIER_RATIO = 3  # ratio bewteen large and small carrier amplitudes



class IrigB122:
    """
    """


    def __init__(self, offset, amplitude, ratio=CARRIER_RATIO, nbSamples=NB_SAMPLES_PER_PERIOD):
        """
        """
        carrierLarge = [round(offset + amplitude * math.sin(2 * math.pi * t / nbSamples)) for t in range(nbSamples)]
        carrierSmall = [round(offset + amplitude / CARRIER_RATIO * math.sin(2 * math.pi * t / nbSamples)) for t in range(nbSamples)]


        self._bitMarkerWave = array.array('I')
        for b in int(CYCLIC_RATIO['m'] * NB_PERIODS_PER_BIT) * carrierLarge + int((1-CYCLIC_RATIO['m']) * NB_PERIODS_PER_BIT) * carrierSmall:
            self._bitMarkerWave.append(b)
        print(f"DEBUG::_bitMarkerWave address = {uctypes.addressof(self._bitMarkerWave)}")


        self._bit0Wave = array.array('I')
        for b in int(CYCLIC_RATIO['0'] * NB_PERIODS_PER_BIT) * carrierLarge + int((1-CYCLIC_RATIO['0']) * NB_PERIODS_PER_BIT) * carrierSmall:
            self._bit0Wave.append(b)
        print(f"DEBUG::_bit0Wave address = {uctypes.addressof(self._bit0Wave)}")


        self._bit1Wave = array.array('I')
        for b in int(CYCLIC_RATIO['1'] * NB_PERIODS_PER_BIT) * carrierLarge + int((1-CYCLIC_RATIO['1']) * NB_PERIODS_PER_BIT) * carrierSmall:
            self._bit1Wave.append(b)
        print(f"DEBUG::_bit1Wave address = {uctypes.addressof(self._bit1Wave)}")


        self._dmaFrame = rp2.DMA()
        self._dmaFrame.active(False)
        # self._dmaFrame.irq(self._dmaFrameIsr)  # only 1 hard interupt to avoid conflict
        print(f"DMA Frame channel = {self._dmaFrame.channel}")


        self._dmaWave = rp2.DMA()
        self._dmaWave.active(False)
        # self.dmaWave.irq(self._dmaWaveIsr, hard=True)
        print(f"DMA Wave channel = {self._dmaWave.channel}")


        self.


        self._currentTime = None  # in s, since epoch
        self._frame = None


    # def _dmaFrameIsr(self, dma):
    #     """
    #     """
    #     print("DEBUG::_dmaFrameIsr::dma.read =", dma.read)
    #
    # def _dmaWaveIsr(self, dma):
    #     """
    #     """
    #     print("DEBUG::_dmaWaveIsr::dma.read =", dma.read)


    @property
    def currentTime(self):
        """
        """
        return self._currentTime


    def _dmaWaveConfig(self):
        """ Configure waveform generation DMA.
        """
        raise NotImplementedError("_dmaWaveConfig() must be implemented in real class!")


    def timeToTimecode(self, time_=None):
        """
        """
        return IrigBTimecode(time_)


    def timecodeToFrame(self, timecode):
        """ Convert IRIG.B Timecode to frame (array('I'))
        """
        frame = array.array('I')
        for b in timecode:
            if b in ('R', 'P'):
                frame.append(uctypes.addressof(self._bit0Wave))
            if b in ('I', 'E', '0'):
                frame.append(uctypes.addressof(self._bit0Wave))
            if b == '1':
                frame.append(uctypes.addressof(self._bit1Wave))
        frame.append(0)  # mark end of frame


        return frame


    def send(self, timecode):
        """ Send IRIG.B122 timecode.
        """
        self._frame = self.timecodeToFrame(timecode)
        print(f"DEBUG::frame address = {uctypes.addressof(self._frame)} ({self._frame})")


        self._dmaWaveConfig()


        # Configure Frame DMA
        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,  # disable chaining
                                                treq_sel=TREQ_PERMANENT,          # unpaced transfert, synchronization is done by PWM wrap
                                                irq_quiet=True,  #False,
                                                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


        # self._dmaFrame.active(True)  # start DMA


    def cancel(self):
        """ Cancel generator
        """
        self._dmaWave.active(False)   # cancel DMA
        self._dmaFrame.active(False)  # cancel DMA


    def isBusy(self):
        """ Check if generator is busy
        """
        return self._dmaFrame.read < uctypes.addressof(self._frame) + 4 * len(self._frame)


    def run(self):
        """ Continuously generates IRIG.B122 Timecodes.


        Get current time at startup and send to clock.


        Check if we need to send the timecode over and over, or if the clock can run by
        itself between synchronizations.
        If not, use internal RTC, and send new timecode every second.


        If the RTC drifts, resync every day at midnight from a ntp server.
        """
        self._currentTime = 1729512607  # "2024-10-21 14:10:07" (2024, 10, 21, 14, 10, 7, 0, 295, 0)


# -*- coding: utf-8 -*-

""" IRIG.B122 timecode generator.

Waveforms are generated using a DAC build arround a PIO + R2R network.
For each bit, a DMA feeds the PIO, and sends the 10 periods according to the bit encoding.
To send the complete timecode, a second DMA feeds the first DMA with the address of each bit in the timecode.
"""

import rp2

from irigB122 import CARRIER_FREQ, NB_SAMPLES_PER_PERIOD, NB_PERIODS_PER_BIT, IrigB122

PIO_0_TXF0    = 0x50200010
DREQ_PIO0_TX0 = 0x00


NB_DAC_BITS = 8                                     # nb bits for dac value
SM_NUM      = 0                                     # state machine number
SM_FREQ     = CARRIER_FREQ * NB_SAMPLES_PER_PERIOD  # Hz - TODO: ensure accurate frequency, and frequency without fractional clock division to avoid jitter


OFFSET    = (2 << NB_DAC_BITS) // 2
AMPLITUDE = OFFSET - 1


DAC_LSB_PIN = 0


class IrigB122Dac(IrigB122):
    """ IRIG.B122 timecode generator class
    """
    def __init__(self, dacLsbPin):
        """ Init generator
        """
        super().__init__(OFFSET, AMPLITUDE)


        self._dacLsbPin = dacLsbPin


        self._sm = rp2.StateMachine(SM_NUM, self._pioCode, freq=SM_FREQ, out_base=dacLsbPin)
        self._sm.active(True)


    @staticmethod
    @rp2.asm_pio(out_init=NB_DAC_BITS*(rp2.PIO.OUT_LOW,), out_shiftdir=rp2.PIO.SHIFT_RIGHT, autopull=True, pull_thresh=NB_DAC_BITS)
    def _pioCode():
        """ PIO code


        pins are automatically pulled from the OSR and written to the bus at the SM frequency
        """
        out(pins, 8)  # NB_DAC_BITS


    def _dmaWaveConfig(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_PIO0_TX0,
                                              irq_quiet=False,
                                              bswap=False,
                                              sniff_en=False,
                                              write_err=False,
                                              read_err=False)


        self._dmaWave.config(write=PIO_0_TXF0,
                            count=NB_PERIODS_PER_BIT*NB_SAMPLES_PER_PERIOD,
                            ctrl=dmaWaveCtrl)



def test():
    """


    # See: https://en.wikipedia.org/wiki/IRIG_timecode#IRIG_timecode
    # R: reference marker (marker bit)
    # I: index marker (0 bit)
    # P: position identifier (marker bit)
    # E: empty bit (0 bit)
    # 0: bit clear (0 bit)
    # 1: bit set (1 bit)
    # Example: 2024-10-21 14:10:07 (Day of year=295)
    timecode = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"
    #            |||| ||| |||| |||  |||| ||   |||| |||| ||
    #               7   0    0   1     4  1      5    9  2
    """
    irigB122Dac = IrigB122Dac(DAC_LSB_PIN)
    try:
        timecode = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"
        irigB122Dac.send(timecode)


        while irigB122Dac.isBusy():
            pass


    finally:
        print("INFO::End")
        irigB122Dac.cancel()



if __name__ == "__main__":
    test()

# -*- coding: utf-8 -*-

""" IRIG.B122 timecode generator.

Waveforms are generated using a PWM (and a low pass filter).
For each bit, a DMA feeds the PWM, and sends the 10 periods according to the bit encoding.
To send the complete timecode, a second DMA feeds the first DMA with the address of each bit in the timecode.
"""

import machine

from irigB122 import CARRIER_FREQ, NB_SAMPLES_PER_PERIOD, NB_PERIODS_PER_BIT, IrigB122

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
]


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 # 125MHz

PWM_FREQ = CARRIER_FREQ * NB_SAMPLES_PER_PERIOD # Hz

OFFSET    = (F_CPU / PWM_FREQ) // 2
AMPLITUDE = OFFSET - 1


PWM_PIN = 0


class IrigB122Pwm(IrigB122):
    """ IRIG.B122 timecode generator class
    """
    def __init__(self, pwmPin):
        """ Init generator
        """
        super().__init__(OFFSET, AMPLITUDE)


        self._pwmPin = pwmPin


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


    def _dmaWaveConfig(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,  #False,
                                              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)



def test():
    """


    # See: https://en.wikipedia.org/wiki/IRIG_timecode#IRIG_timecode
    # R: reference marker (marker bit)
    # I: index marker (0 bit)
    # P: position identifier (marker bit)
    # E: empty bit (0 bit)
    # 0: bit clear (0 bit)
    # 1: bit set (1 bit)
    # Example: 2024-10-21 14:10:07 (Day of year=295)
    timecode = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"
    #            |||| ||| |||| |||  |||| ||   |||| |||| ||
    #               7   0    0   1     4  1      5    9  2
    """
    irigB122Pwm = IrigB122Pwm(PWM_PIN)
    try:
        timecode = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"
        irigB122Pwm.send(timecode)


        while irigB122Pwm.isBusy():
            pass


    finally:
        print("INFO::End")
        irigB122Pwm.cancel()



if __name__ == "__main__":
    test()

# -*- coding: utf-8 -*-

""" IRIG.B122 timecode generator.
"""
import time

# Time tuple index
YEAR = 0
MONTH = 1
DAY = 2
HOURS = 3
MINUTES = 4
SECONDS = 5
DUMMY = 6
DAY_OF_YEAR = 7
TZ = 8 # only availbale on unix port

DAYS_PER_MONTH = (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31)


class IrigBTimecode:
    """ IRIG.B122 timecode builder class


    # See: https://en.wikipedia.org/wiki/IRIG_timecode#IRIG_timecode
    # R: reference marker (marker bit)
    # I: index marker (0 bit)
    # P: position identifier (marker bit)
    # E: empty bit (0 bit)
    # 0: bit clear (0 bit)
    # 1: bit set (1 bit)
    # Example: 2024-10-21 14:10:07 (Day of year=295)
    timecode = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"
    #            |||| ||| |||| |||  |||| ||   |||| |||| ||
    #               7   0    0   1     4  1      5    9  2
    """
    def _dec2Bin(self, decimal, digit=4):
        """
        """
        return f"{decimal:0{digit}b}"


    def computeDayOfYear(self, year, month, day):
        """ Compute Day of Year from year, month and day


        Note about leap year (https://en.wikipedia.org/wiki/Leap_year#Gregorian_calendar):
        Every year that is exactly divisible by four is a leap year, except for years that are exactly
        divisible by 100, but these centurial years are leap years if they are exactly divisible by 400.
        For example, the years 1700, 1800, and 1900 are not leap years, but the years 1600 and 2000 are.
        """
        # Current month
        doy = day


        # Full months
        for m in range(month-1):
            doy += DAYS_PER_MONTH[m]


        # Leap year
        if not year % 4 and year % 100 or not year % 400:
            doy += 1


        return doy


    def __call__(self, t=None):
        """ Convert given time to IRIG.B timecode.


        @param t: time as returned by time.localtime/mktime(), or by time.time()
                      If None, use current time
        @type t: tuple or int


        @return
        """
        if t is None:
            t = time.localtime()
        elif isinstance(t, int) or isinstance(t, float):
            t = time.localtime(t)


        seconds = t[SECONDS]
        minutes = t[MINUTES]
        hours = t[HOURS]
        dayOfYear = t[DAY_OF_YEAR]


        timecode = "R"  # start of timecode


        # Set seconds
        for s in reversed(self._dec2Bin(seconds % 10, 4)):
            timecode += s
        timecode += 'I'  # index marker
        for s in reversed(self._dec2Bin(seconds // 10, 3)):
            timecode += s


        timecode += 'P'  # position identifier


        # set minutes
        for s in reversed(self._dec2Bin(minutes % 10, 4)):
            timecode += s
        timecode += 'I'  # index marker
        for s in reversed(self._dec2Bin(minutes // 10, 3)):
            timecode += s
        timecode += 'E'  # empty bit


        timecode += 'P'  # position identifier


        # Set hours
        for s in reversed(self._dec2Bin(hours % 10, 4)):
            timecode += s
        timecode += 'I'  # index marker
        for s in reversed(self._dec2Bin(minutes // 10, 2)):
            timecode += s
        timecode += "EE"  # empty bits


        timecode += 'P'  # position identifier


        # set day of year - TODO: check if needed
        for s in reversed(self._dec2Bin(dayOfYear % 10, 4)):
            timecode += s
        timecode += 'I'  # index marker
        for s in reversed(self._dec2Bin((dayOfYear // 10) % 10, 4)):
            timecode += s
        timecode += 'P'  # position identifier
        for s in reversed(self._dec2Bin((dayOfYear // 100) % 10, 2)):
            timecode += s


        # Complete unused part of timecode
        timecode += "EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"


        return timecode



def test():
    """
    """
    TIMECODE = "R1110I000P0000I100EP0010I10EEP1010I1001P01EEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEPEEEEEEEEEP"


    irigbTimecode = IrigBTimecode()
    # time.strptime("2024-10-21 14:10:07", "%Y-%m-%d %H:%M:%S") unsupported under MicroPython
    print("2024-10-21 14:10:07 -> doy =", irigbTimecode.computeDayOfYear(2024, 10, 21))
    t = (2024, 10, 21, 14, 10, 7, 0, 295)  # 1729512607
    timecode = irigbTimecode(t)


    print(f"original : {TIMECODE}")
    print(f"generated: {timecode}")
    assert TIMECODE == timecode, "Incorrect timecode!"



if __name__ == "__main__":
    test()