pyqso/src/adif.py

#!/usr/bin/env python 
# File: adif.py

#    Copyright (C) 2012 Christian Jacobs.

#    This file is part of PyQSO.

#    PyQSO is free software: you can redistribute it and/or modify
#    it under the terms of the GNU General Public License as published by
#    the Free Software Foundation, either version 3 of the License, or
#    (at your option) any later version.
#
#    PyQSO is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU General Public License for more details.
#
#    You should have received a copy of the GNU General Public License
#    along with PyQSO.  If not, see <http://www.gnu.org/licenses/>.

import re
import logging
from datetime import datetime

from record import *

# All the possible field names and their associated data types 
# from the ADIF specification (version 3.0.2)
AVAILABLE_FIELD_NAMES_TYPES = {"ADDRESS": "M", 
                              "ADDRESS__INTL": "G", 
                              "AGE": "N",
                              "CALL": "S", 
                              "FREQ": "N", 
                              "DATE": "D",
                              "QSO__RANDOM": "B"}

# A: AwardList
# B: Boolean
# N: Number
# S: String
# I: International string
# D: Date
# T: Time
# M: Multi-line string
# G: Multi-line international string
# L: Location
DATA_TYPES = ["A", "B", "N", "S", "I", "D", "T", "M", "G", "L", "E"]
   
class ADIF:
   
   def __init__(self):
      # Class for I/O of files using the Amateur Data Interchange Format (ADIF).
      logging.debug("New ADIF instance created!")
      
   def read(self, path):
      
      logging.debug("Reading in ADIF file with path: %s." % path)
      
      try:
         f = open(path, 'r')
         text = f.read()
         f.close() # Close the file, otherwise "bad things" might happen!
      except IOError as e:
         logging.error("I/O error %d: %s" % (e.errno, e.strerror))
         raise
      except:
         logging.error("Unknown error occurred when reading the ADIF file.")
         raise
      
      #if(format == ".adi"):
      records = self.parse_adi(text)
      #elif(format == ".adx"):
         #records = self.parse_adx(text)
      #else:
         #logging.error("Unknown file format.")
         
      if(records == []):
         logging.warning("No records found in the file. Empty file or wrong file type?")
         
      return records
      
   def parse_adi(self, text):
      records = []

      # Separate the text at the <eor> or <eoh> markers.
      tokens = re.split('(<eor>|<eoh>)', text, flags=re.IGNORECASE)
      tokens.pop() # Anything after the final <eor> marker should be ignored.
      
      # The header might tell us the number of records, but let's not assume
      # this and simply ignore it instead (if it exists).
      if(re.search('<eoh>', text, flags=re.IGNORECASE) is not None):
         # There is a header present, so let's ignore everything
         # up to and including the <eoh> marker. Note that
         # re.search has been used here to handle any case sensitivity.
         # Previously we were checking for <eoh>. <EOH> is also valid
         # but wasn't been detected before.
         while len(tokens) > 0:
            t = tokens.pop(0)
            if(re.match('<eoh>', t, flags=re.IGNORECASE) is not None):
               break
            
      n_eor = 0 
      n_record = 0
      records = []
      for t in tokens:
         if(re.match('<eor>', t, flags=re.IGNORECASE) is not None):
            n_eor = n_eor + 1
            continue
         else:
            n_record = n_record + 1
            # Each record will have field names and corresponding
            # data entries. Store this in a dictionary.
            # Note: This is based on the code written by OK4BX.
            # (http://web.bxhome.org/blog/ok4bx/2012/05/adif-parser-python)
            fields_and_data_dictionary = {}
            fields_and_data = re.findall('<(.*?):(\d*).*?>([^<\t\n\r\f\v\Z]+)', t)  
            for fd in fields_and_data:
               # Let's force all field names to be in upper case.
               # This will help us later when comparing the field names
               # against the available field names in the ADIF specification.
               fields_and_data_dictionary[fd[0].upper()] = fd[2][:int(fd[1])]
            records.append(Record(fields_and_data_dictionary))    
      
      assert n_eor == n_record
      
      return records

      
   def write(self, records, path):
      f = open(path, 'w') # Open file for writing
      
      # First write a header containing program version, number of records, etc.
      dt = datetime.now()
      
      f.write('''Amateur radio log file. Conforms to ADIF specification version 1.0.
Generated on %s. Contains %d record(s). 
      
<adif_ver:5>3.0.2
<programid:5>PyQSO
<programversion:11>Development

<eoh>\n''' % (dt, len(records)))
      
      # Then write each log to the file.
      for r in records:
         if(all([r.fields_and_data[key] is None for key in r.fields_and_data.keys()])):
            # Check if all entries of record[i] are None - if so, don't add it.
            continue
         for key in r.fields_and_data.keys():
            if( (r.fields_and_data[key] is not None) and (r.fields_and_data[key] != "") ):
               f.write("<%s:%d>%s\n" % (key.lower(), len(r.fields_and_data[key]), r.fields_and_data[key]))
         f.write("<eor>\n")

      f.close()
      
   
   def test_read(self):
      f = open("../ADIF.test_read.adi.test", 'w')
      f.write('''Some test ADI data.<eoh>

<call:4>TEST<band:3>40M<mode:2>CW
<qso_date:8:d>20130322<time_on:4>1955<eor>''')
      f.close()
    
      records = self.read("../ADIF.test_read.adi.test")
      
      assert records == [{'band': '40M', 'time_on': '1955', 'call': 'TEST', 'mode': 'CW', 'qso_date': '20130322'}]
              
if(__name__ == '__main__'):
   a = ADIF() 
   a.test_read()