rtctimestampbias/rtctimestampbias.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

# the following packages need to be installed via pip: 
# serial, csv

import serial.tools.list_ports
import csv
import time
from datetime import datetime
from optparse import OptionParser
from sys import argv
from os import path

# RTC timestamp bias: saving and comparing timestamps
# of the microcontroller's real time clock
# https://gitlab.com/simplemicrocontrollers/rtctimestampbias

version = "0.1"
filename = "rtctimestampbias.csv"

def ShowPorts(option, opt, value, parser):
    number = 0
    for port in ports:
        if number > 0: print("---")
        print(f"port {number}: {port.name}")
        print(f"hwid: {port.hwid}")
        print(f"desc: {port.description}")
        number += 1

def SaveByNumber(number):
    SaveValue(ports[int(number)].device)

def SaveByName(name):
    for port in ports:
        if port.name == name:
            SaveValue(port.device)

def SaveValue(device):
    ser = serial.Serial(device, 9600)
    response = ser.readline()
    current_host_time = datetime.now()
    decoded_response = int(response.decode('utf-8'))
    ser.close()
    print("Serial device timestamp:")
    print(decoded_response)
    hosttimestamp = int(round(current_host_time.timestamp()))
    print("Host device timestamp:")
    print(hosttimestamp)
    csvdatarow = [[current_host_time, hosttimestamp, decoded_response]]
    with open(filename, 'a') as csvfile:
        csvwriter = csv.writer(csvfile)
        for row in csvdatarow:
            csvwriter.writerow(row)

def CalculateCSVdata(option, opt, value, parser):
    hostFirst = None
    deviceFirst = None
    hostLast = None
    deviceLast = None
    with open(filename, newline='') as csvfile:
        spamreader = csv.reader(csvfile, delimiter=' ', quotechar='|')
        rownumber = 0
        for row in spamreader:
            if rownumber == 0:
                hostFirst = ', '.join(row).split(',')[2]
                deviceFirst = ', '.join(row).split(',')[3]
            hostLast = ', '.join(row).split(',')[2]
            deviceLast = ', '.join(row).split(',')[3]
            rownumber += 1
    print(f"First timestamp of the host: {hostFirst}")
    print(f"First timestamp of the device: {deviceFirst}")
    print(f"Last timestamp of the host: {hostLast}")
    print(f"Last timestamp of the device: {deviceLast}")
    hostDiff = int(hostLast) - int(hostFirst)
    deviceDiff = int(deviceLast) - int(deviceFirst)
    totalDiff = deviceDiff - hostDiff
    print(f"Host diff: {hostDiff}; device diff: {deviceDiff}; host/device total diff: {totalDiff}")
    resultBias = (86400 / deviceDiff) * totalDiff
    print(f"Resulting daily clock bias in seconds: {resultBias}")
    resultBiasMs = int((86400 / resultBias) * 1000)
    print(f"Offset in one second per the number of milliseconds: {resultBiasMs}")

def CMDOptions():
    class FormatedParser(OptionParser):
        def format_epilog(self, formatter):
            return self.epilog
    usage = "usage: %prog [options]"
    progname = path.basename(__file__)
    epilog = ("".join(["Examples:\n  ",
        progname, " -l\n  ",
        progname, " -n 2\n  ",
        progname, " -d ttyUSB0\n  ",
        progname, " -c\n"]))
    opts = FormatedParser(usage=usage, version="%prog " + version,
                          epilog=epilog)
    opts.add_option('-l', '--list', action="callback",
                    callback=ShowPorts, help="show information about each port")
    opts.add_option('-n', '--num', action="store", type="string",
                    dest="num", metavar="number", default=False,
                    help="save data by port number")
    opts.add_option('-d', '--dev', action="store", type="string",
                    dest="dev", metavar="name", default=False,
                    help="save data by device name")
    opts.add_option('-c', '--calc', action="callback",
                    callback=CalculateCSVdata, help="calculate data from csv file")
    options, value = opts.parse_args()
    if options.num:
        SaveByNumber(options.num)
    if options.dev:
        SaveByName(options.dev)
    if len(argv[1:]) == 0:
        print ("no argument given\n")
        opts.print_help()

if __name__ == "__main__":
    ports = list(serial.tools.list_ports.comports())
    try:
        CMDOptions()
    except KeyboardInterrupt:
        exit()