Recover call from WSJT-X

recover.py

#!/usr/bin/env python
import os
import sys

from collections import namedtuple
from datetime import datetime

from tqdm import tqdm

start_date = datetime(2023, 6, 15, 20, 55, 0)

MY_CALL = "W6BSD"
MY_GRID = "CM87"
POWER = "14"
FILE = os.path.expanduser('~/Library/Application Support/WSJT-X/ALL.TXT')

BANDS = {
 1: '160m',
 3: '80m',
 5: '60m',
 7: '40m',
 10: '30m',
 14: '20m',
 18: '17m',
 21: '15m',
 24: '12m',
 28: '10m',
 50: '6m',
 51: '6m',
 52: '6m',
 53: '6m',
 54: '6m',
}

def count_lines(filename):
  lines = 0
  with open(filename) as fdi:
    for _ in fdi:
      lines += 1
  return lines

def read_data(filename):
  entries = count_lines(filename)
  data = []
  with open(filename) as fdi:
    for line in tqdm(fdi, total=entries, desc="Reading data"):
      fields = line.split()
      date = datetime.strptime(fields[0], '%y%m%d_%H%M%S')
      if date < start_date:
        continue
      if fields[7] == MY_CALL and '73' in fields[-1]:
        data.append(fields)
  return data

ADIF_FIELDS = ("CALL", "FREQ", "BAND", "MODE", "QSO_DATE", "TIME_ON", "QSO_DATE_OFF", "TIME_OFF",
               "STATION_CALLSIGN", "MY_GRIDSQUARE", "TX_PWR", "COMMENT")
class Record(namedtuple("ADIFRecord", ADIF_FIELDS)):
  def __new__(cls, *args):
    date = datetime.strptime(args[0], '%y%m%d_%H%M%S')
    freq = round(float(args[1]) + float(args[6]) / 1000000, 5)

    items = [
      args[8],
      str(freq),
      BANDS[int(freq)],
      args[3],
      date.strftime('%Y%m%d'),
      date.strftime('%H%M%S'),
      date.strftime('%Y%m%d'),
      date.strftime('%H%M%S'),
      MY_CALL,
      MY_GRID,
      str(POWER),
      f'(R) {args[3]} - Recv: {args[4]}',
    ]
    return tuple.__new__(cls, items)

  def adif(self):
    lines = []
    for field in self._fields:
      data = getattr(self, field)
      lines.append(f"<{field}:{len(data)}>{data} ")
    lines.append('<EOR>')
    return '\n'.join(lines)

def main():
  data = read_data(FILE)
  print("Processing:", end=" ")
  with open('recovered.adi', 'w') as fdout:
    for count, field in enumerate(data):
      entry = Record(*field)
      print(entry.adif(), end="\n\n", file=fdout)

  print('Done.')
  print('Recovered contacts have been saved into recovered.adi')

if __name__ == "__main__":
  main()