Script for parsing DCF77 data captured as raw voltage data.

dcf77-parser.py

"""
    Script for parsing DCF77 data captured as raw voltage data.

    Input:
        Takes a CSV with two columns:
        | Timestamp in s | Voltage in V |
        ---------------------------------
        values ...
    Sampling rate needs to be no less than 100HZ

    Written by Jesse Born <[email protected]> on 2023-02-23

    Public Domain since 2023
"""
import csv


def parse_csv_file(file_path):
    with open(file_path, 'r') as csv_file:
        reader = csv.reader(csv_file)
        next(reader)  # Skip header row
        csv_list = []
        for row in reader:
            timestamp, voltage = float(row[0]), float(row[1])
            csv_list.append([timestamp, voltage > 3.5])

        # parse to binary
        last_val = [0.0, False]
        bin_list = []

        seq_start = []

        ts: float
        v: bool
        for i, (ts, v) in enumerate(csv_list):
            if last_val[1] == v:
                pass
                # still the same
            else:
                # value changed, reset timestamp and calculate delta
                delta_time = ts-last_val[0]
                last_val = [ts, v]
                # print(f"change at {ts} to {'HIGH' if v else 'LOW '}, delta {delta_time}")
                if not v:
                    bin_list.append(delta_time > 0.12)
                else:
                    if delta_time >= 1:
                        # print(f"found seq start at {len(bin_list)}")
                        seq_start.append(len(bin_list))
    return bin_list[seq_start[0]:seq_start[1]]


def parse_dcf77_to_iso8601_datetime(dcf77bits: list):
    def mask_multiply(bits, mask):
        result = 0
        for i in range(len(bits)):
            result += bits[i] * mask[i]
        return result
    min_mask = [1, 2, 4, 8, 10, 20, 40]
    hour_mask = [1, 2, 4, 8, 10, 20]
    calendarday_mask = [1, 2, 4, 8, 10, 20]
    weekday_mask = [1, 2, 4]
    month_mask = [1, 2, 4, 8, 10]
    year_mask = [1, 2, 4, 8, 10, 20, 40, 80]

    min = mask_multiply(dcf77bits[21:28], min_mask)
    h = mask_multiply(dcf77bits[29:35], hour_mask)
    calday = mask_multiply(dcf77bits[36:42], calendarday_mask)
    weekday_mask = mask_multiply(dcf77bits[42:45], weekday_mask)
    month = mask_multiply(dcf77bits[45:50], month_mask)
    year = mask_multiply(dcf77bits[50:58], year_mask) + 2000
    # 2007-08-31T16:47+00:00
    return f"{year}-{month:02d}-{calday:02d}T{h:02d}:{min:02d}"

def main():
    bits = parse_csv_file("/Users/jesseb0rn/Desktop/data.csv")
    print(f"It currently is {parse_dcf77_to_iso8601_datetime(bits)}")

if __name__ == "__main__":
    main()