""" 2023-11-17 - First release 2024-01-20 - - Bugfix for date decoding and parity check - Disable and enable debug messages with separate function 2025-06-09 - Use hardware timer if virtual timers are not 2026-01-23 - - Added the option to also use a signal source with an inverted signal. This resolves Issue #3 - Added some more debug information as suggested in Issue #3 - Changed the year format to 4 digits 2026-03-22 - Merged Pull-Request #5. Thanks to rwv """ import utime from machine import Pin, Timer class dcf77: # Constants for the irq handler IRQ_MINUTE = 0 IRQ_HOUR = 1 IRQ_DAY = 2 IRQ_MONTH = 3 IRQ_YEAR = 4 IRQ_DST = 5 # The offset for the year # The dcf protocol only provides two-digit year number YEAR_OFFSET = 2000 # Timer to use # negativ values are virtual timers. timer = -1 def __init__( self, tco_pin: Pin, false_time: list[int] = [50, 130], true_time: list[int] = [150, 230], pause_time: list[int] = [1700, 2500], inverted: bool = False, ): """ - ``tco_pin``: Pin object of the TCO pin - ``false_time``: Minimum/Maximum impulse width for a FALSE signal - ``true_time``: Minimum/Maximum impulse width for a TRUE signal - ``pause_time``: Minimum/Maximum pause to recognize tick 59 - ``inverterted``: DCF77 module sends an inverted signal """ self.FALSE_TIME = false_time self.TRUE_TIME = true_time self.TICK59_TIME = pause_time self.TCO = tco_pin self.DEBUG = False self.TIMEOUT_TIME = self.TICK59_TIME[1] + 1000 self.inverted = inverted self.irq_handler = None self.irq_start = 0 # Starttime of the interrupt self.irq_last = 0 # Time of last interrupt self.last_pulse = 0 # Length of last pulse # Try to initalize a virtual timer. # If the micropython port doesen't support virtual timers, switch to a hardware timer. if self.timer == -1: try: self.timer_timeout = Timer(self.timer) # Initalize virtual timer except ValueError: self.timer = 0 self.timer_timeout = Timer( self.timer ) # Virtual timer not supported, use hardware timer 0 else: self.timer_timeout = Timer( self.timer ) # Use custom timer (virtual or hardware) self.found59 = False # Tick59 was found self.tick = 0 # Actual tick self.signal = [] # Actual received signal self.signal_last = [] # Last received signal self.valid = False # Signal is valid self.datetime = [0, 0, 0, 0, 0, 0, 0, 0] self.datetime_last = [0, 0, 0, 0, 0, 0, 0, 0] self.dst_changed = False # Init TCO pin self.TCO.init(mode=Pin.IN, pull=Pin.PULL_UP) # Function to print debug messages def __print(self, *val): if self.DEBUG: text = "" for v in range(len(val)): text += str(val[v]) print("DCF77: ", text) # Interrupt handler for the DCF IRQ def __handler_interrupt(self, pin): self.__run() # Interrupt hanlder for the timeout timer def __timeout_callback(self, timer): self.found59 = False self.valid = False self.tick = 0 self.signal.clear() self.__print("Signal timeout - Start new scanning for tick 59") # Get the signal def __get_signal(self): val = self.TCO.value() if self.inverted: return not val return val # Main function. Called from the interrupt handler def __run(self): # Starting the timeout timer self.timer_timeout.init( mode=Timer.ONE_SHOT, period=self.TIMEOUT_TIME, callback=self.__timeout_callback, ) # If the last edge was detected between min/max TICK59_TIME # we have found the beginning of the telegram if ( utime.ticks_diff(utime.ticks_ms(), self.irq_last) > self.TICK59_TIME[0] and utime.ticks_diff(utime.ticks_ms(), self.irq_last) < self.TICK59_TIME[1] ): if not self.found59: self.__print("Found Tick 59") self.tick = 0 self.found59 = True # Saving the actual time for comparsion self.irq_last = utime.ticks_ms() # If the signal from the DCF module is low # we save the time if not self.__get_signal(): self.irq_start = utime.ticks_ms() else: # The signal from the DCF module is high # Calculate the pulse length diff = utime.ticks_diff(utime.ticks_ms(), self.irq_start) self.last_pulse = diff val = 2 # Is the pulse a 0 or 1? if diff >= self.FALSE_TIME[0] and diff <= self.FALSE_TIME[1]: val = 0 self.__print("Detected '0'") elif diff >= self.TRUE_TIME[0] and diff <= self.TRUE_TIME[1]: val = 1 self.__print("Detected '1'") else: self.__print( f"Wrong pulse width {diff} - Start new scanning for tick 59" ) self.found59 = False self.signal.clear() # If the beginning of the telegram was found let count the ticks and save the values if self.found59: # self.__print("Tick: ", self.tick, " - ", val) self.signal.append(val) self.irq_start = 0 self.tick += 1 # If we have tick 59 the telegram is completed # Copy the data to another list and clear the source list if self.tick > 58: self.signal_last = self.signal.copy() self.signal.clear() self.tick = 0 # If tick is 1 end the saved signal list is not empty # we have a valid telegram if self.tick == 1: if len(self.signal_last) > 0: self.valid = True self.__decode() self.__custom_irq() self.__print("Last Transmission: ", self.signal_last) # Function to decoding the time informations def __decode_time(self, time, check_parity): vals = [1, 2, 4, 8, 10, 20, 40, 80] # Values for the bit positions sum = 0 # Counter variable parity_check = 0 # Counter variable for the parity check # If check_parity is True the last value in the list is the parity # so we should not count this value end = len(time) if check_parity: end -= 1 # Counting the time or date for t in range(end): if time[t]: sum += vals[t] parity_check += 1 # Checking parity # If a even number of bits was TRUE the parity # has to be FALSE if check_parity: if (parity_check % 2) == 0: if time[-1]: self.valid = False self.__print("Parity is not correct") return 999 return sum # Function to decoding the date informations def __decode_date(self, date, check_parity=True): vals = [1, 2, 4, 8, 10, 20, 40, 80] # Values for the bit positions parity_check = 0 # Counter variable for the parity check # Checking parity # If a even number of bits was TRUE the parity # has to be FALSE if check_parity: for t in range(len(date) - 1): if date[t]: parity_check += 1 if (parity_check % 2) == 0: if date[-1]: self.valid = False self.__print("Parity is not correct") return [999, 999, 999, 999] # Decoding Day of month DayOfMonth = 0 for v, b in enumerate(date[0:6]): if b: DayOfMonth += vals[v] # Decoding the weekday Weekday = 0 for v, b in enumerate(date[6:9]): if b: Weekday += vals[v] # Decoding the month Month = 0 for v, b in enumerate(date[9:14]): if b: Month += vals[v] # Decoding the year Year = self.YEAR_OFFSET for v, b in enumerate(date[14:22]): if b: Year += vals[v] return [DayOfMonth, Weekday, Month, Year] # Decoding the telegram def __decode(self, with_seconds=False): """ Decoding the DCF77 signal. If ``None`` is returned the signal is not valid If some value return ``999`` the decoding failed Bit 00 = 0 Minutenmarke, immer 0-Bit Bit 01..14 = Wetterdaten Bit 15 = Rufbit, ist es 1, liegt ggf. Störung vor Bit 16 = Ankündigung Zeitumstellung Bit 17..18 = Offset in stunden zur UTC. Bit 19 = Ankündigung Schaltsekunde Bit 20 = Start des Zeitprotokolls, immer 1-Bit Bit 21..28 = Minuten mit Parität Bit 29..35 = Stunden mit Parität Bit 36..41 = Kalendertag Bit 42..44 = Wochentag, 1=Montag Bit 45..49 = Monat Bit 50..58 = Jahr mit Parität Bit 59 = Kein Impuls """ # Signal to short or to long if len(self.signal_last) != 59: self.__print("Signal not valid -- Shorter or longer as 59 ticks") self.valid = False return # Checking some bits that must be everytime 0 or 1 if self.signal_last[0]: self.__print("Bit 0 is not FALSE") self.valid = False return if not self.signal_last[20]: self.__print("Bit 20 is not TRUE") self.valid = False return # Decoding minutes = self.__decode_time(self.signal_last[21:29], True) hours = self.__decode_time(self.signal_last[29:36], True) day, weekday, month, year = self.__decode_date(self.signal_last[36:59], True) seconds = 0 if with_seconds: seconds = self.tick self.datetime_last = self.datetime.copy() self.datetime = [year, month, day, weekday - 1, hours, minutes, seconds, 0] # Function to run the custom irq handler def __custom_irq(self): if self.irq_handler is not None: run_handler = False # Minute has changed if self.IRQ_MINUTE in self.irq_mode: if self.datetime_last[5] != self.datetime[5]: run_handler = True # Hour has changed if self.IRQ_HOUR in self.irq_mode: if self.datetime_last[4] != self.datetime[4]: run_handler = True # Day has changed if self.IRQ_DAY in self.irq_mode: if self.datetime_last[2] != self.datetime[2]: run_handler = True # Month has changed if self.IRQ_MONTH in self.irq_mode: if self.datetime_last[1] != self.datetime[1]: run_handler = True # Year has changed if self.IRQ_YEAR in self.irq_mode: if self.datetime_last[0] != self.datetime[0]: run_handler = True # DST changed to TRUE if self.IRQ_DST in self.irq_mode: if self.signal_last[16]: if not self.dst_changed: run_handler = True self.dst_changed = True else: self.dst_changed = False if run_handler: self.irq_handler() # Function to start fetching the DCF77 signal def start(self): """ Start fetching time informations """ # Configure IRQ self.TCO.irq( trigger=Pin.IRQ_RISING | Pin.IRQ_FALLING, handler=self.__handler_interrupt ) self.__print("Start decoding") # Function to stop fetching the DCF77 signal def stop(self): """ Stop fetching time informations """ # Disable IRQ self.TCO.irq(handler=None) self.timer_timeout.deinit() self.valid = False self.__print("Stop decoding") # Function that returns the last fetched signal def get_LastSignal(self): """ Return the last fetched signal as list """ return self.signal_last # Function that returns the actual time and date informations def get_DateTime(self, with_seconds=False): """ Returns the actual time and date informations as an 8-tuple which contains: year, month, day, weekday, hours, minutes, seconds, subseconds This tuple can be directly used with the machine.RTC module - ``year`` contains only the last 2 digest - ``seconds`` is everytime 0, when ``with_seconds`` is TRUE the value will be set to the actual tick - ``subseconds`` is everytime 0 - If ``None`` is returned the signal is not valid - If some value return ``999`` the decoding failed """ return self.datetime # Function that returns some status informations def get_Infos(self): """ Returns some status informations as dict """ info = { "Found59": self.found59, "Valid": self.valid, "Last pulse length": self.last_pulse, "Tick": self.tick, } if self.valid: info.update( { "Call bit": self.signal_last[15], "Summer time announcement": self.signal_last[16], "CEST": self.signal_last[17], "CET": self.signal_last[18], "Leap second": self.signal_last[19], } ) return info # Enable an custom irq handler def irq(self, mode: list[int] = [0], handler=None): """ Enables an custom irq handler for various events. ```Mode``` has to be a list of the following modes: - ```IRQ_MINUTE``` = irq is fired when the minute changed - ```IRQ_HOUR``` = irq is fired when the hour changed - ```IRQ_DAY``` = irq is fired when the day changed - ```IRQ_MONTH``` = irq is fired when the month changed - ```IRQ_YEAR``` = irq is fired when the year changed - ```IRQ_DST``` = irq is fired when the DST flag changes to TRUE """ self.irq_mode = mode self.irq_handler = handler # Enable/Disable debug messages def debug(self, mode=False): self.DEBUG = mode