Digimodes auf dem Raspberry Pi 4

Seit knapp einem Jahr betreibe ich jetzt einen Raspberry Pi 4 im Büro, den ich per VNC aus dem Wohnzimmer bediene, um damit Digimodes wie FT8, PSK31, RTTY und so weiter zu machen. Neben WSJT-X für die „automatischen“ Betriebsarten, nutze ich auch FLDIGI auf dem Rechner. Da ich seit längerem auch Cloudlog nutze als Alltagslog, musste hier also eine Lösung hier, die möglichst ohne Handarbeit die entsprechenden Logeinträge der einzelnen Programme ins Cloudlog importiert.

Ich wäre nicht so ein verkappter Amateur-Programmierer (ich weiß, ich habe das mal vor vielen Jahren gelernt beruflich, aber leider seit der Ausbildung nur sehr spärlich benutzt), wenn ich mir da nicht eine eigenständige Lösung schaffen würde, die dieses Problem behebt. Das Ganze funktioniert also bei mir über ein Python-Script, welches per Crontab auf dem Raspberry Pi aufgerufen wird (in einem durch mich vorgegebenen Intervall) und was Neben dem Upload des Logs ins Cloudlog auch gleich den Transport nach Clublog und LoTW vornimmt. Die Lösung besteht aus dem Script selbst und einer Konfigurations-Datei (ini-Datei), die verschiedene Informationen beinhaltet.

Ich will euch das Script nicht vorenthalten und packe es (samt ini-File) einfach mal in diesen Beitrag rein:

#!/usr/bin/python3
import requests
import sys, getopt
import configparser
import os.path, time 
import adif_io
from pathlib import Path
from datetime import datetime
from pytz import utc, timezone
from time import mktime
from tzlocal import get_localzone
import urllib.request

def export_adif(src):
    # init configparser
    config = configparser.ConfigParser()
    config.read('sync3.ini')
    
    # get API-data
    apiKey = config['DEFAULT']['ApiKey']
    apiURL = config['DEFAULT']['ApiURL']
    
    # init ADIF-configs
    adifPaths = config.sections()
    for path in config.sections():
        if path == src:
            path = config[src]['Path']
            last_modified = time.ctime(os.path.getmtime(path))
            last_uploaded = time.ctime(os.path.getmtime(path + '.last_uploaded'))
            print("last modified: %s" % last_modified)
            print("last uploaded: %s" % last_uploaded)

            last_modified_time = datetime.strptime(last_modified, "%a %b %d %H:%M:%S %Y")
            last_uploaded_time = datetime.strptime(last_uploaded, "%a %b %d %H:%M:%S %Y")
            
            if last_modified_time < last_uploaded_time:
                print("No upload needed")
            else:
                print("Exporting file {}".format(path))
                
                qsos_raw, adif_header = adif_io.read_from_file(path)

                # The QSOs are probably sorted by QSO time already, but make sure:
                for qso in qsos_raw:
                    qso["t"] = adif_io.time_on(qso)
                qsos_raw_sorted = sorted(qsos_raw, key = lambda qso: qso["t"])
        
                post_line = ""
                for qso in qsos_raw_sorted:
                    adif_line = ""
                    
                    tz = get_localzone()
                    local_dt = tz.localize(datetime(2010, 4, 27, 12, 0, 0, 0), is_dst=None)

                    if (mktime(datetime.utctimetuple(qso["t"])) >= mktime(datetime.utctimetuple(tz.localize(last_uploaded_time)))):
                        print("new qso")
                        del (qso["t"])
                        for key in qso:
                            adif_line+="<" + key.lower() + ":" + str(len(qso[key])) + ">" + qso[key]
                        post_line += adif_line.replace("OLIVIA-", "OLIVIA ")
                        post_line += "<eor>"

                    

                # building post-request
                post = {"key": apiKey, "type": "adif", "string": post_line}
                print (post)
                try:
                    resp = requests.post(apiURL, json=post)
                    if resp.status_code != 201:
                        print('Error: POST {} Result: {}'.format(apiURL,resp.status_code))
                    else:
                        print('Success: POST {} Result: {}'.format(apiURL,resp.status_code))
                        Path(path + '.last_uploaded').touch()
                        
                        with urllib.request.urlopen(config['DEFAULT']['CloudlogURL']) as response:
                            html = response.read()
                        
                        with urllib.request.urlopen(config['DEFAULT']['LoTWURL']) as response:
                            html = response.read()

                except:
                    pass
                
                #adif_file.close()


def main(argv):
   try:
      opts, args = getopt.getopt(argv[1:],"hs:",["source=",])
   except getopt.GetoptError:
      print ('{} -s|--source <PROGRAM-NAME>'.format(argv[0]))
      sys.exit(2)
   for opt, arg in opts:
      if opt == '-h':
         print ('{} -s|--source <PROGRAM-NAME>'.format(argv[0]))
         sys.exit()
      elif opt in ("-s", "--source"):
         src = arg
         export_adif(src)
         sys.exit()
   print ('{} -s|--source <PROGRAM-NAME>'.format(argv[0]))
   sys.exit(2)


if __name__ == "__main__":
   main(sys.argv)

Und hier die dazugehörige ini-Datei (sync3.ini):

###################################################
# Sync2.ini - configfile for ADIF-synchronisation #
###################################################
[DEFAULT]
# Cloudlog API-Key
ApiKey = xxxxxxxxxxxxxxxx

# URL of API within Cloudlog-instance
ApiURL = https://cloudlog.dg9vh.de/api/QSO

CloudlogURL = https://cloudlog.dg9vh.de/clublog/upload/DG9VH
LoTWURL = https://cloudlog.dg9vh.de/lotw/lotw_upload

#
# Here follows configuration of software-instances with ADIF-exports
#
[WSJT-X]
# Path to ADIF-File
Path = /home/pi/.local/share/WSJT-X/wsjtx_log.adi

[FLDIGI]
# Path to ADIF-File
Path = /home/pi/.fldigi/logs/logbook.adif

[JS8CALL]
# Path to ADIF-File
Path = /home/pi/.local/share/JS8Call/js8call_log.adi

Aufgerufen wird das Programm mit z.B. python3 sync3.py -s WSJT-X

Wenn ihr Fragen habt, dafür ist der Kommentarbereich eigentlich ganz hervorragend geeignet 🙂