door-detect/csv_parser.py

from data import LocationData
from collections import namedtuple

ColumnSpec = namedtuple("ColumnSpec", "name format")

field_fmt = {
    'lat' : ColumnSpec('latitude', 'f'),
    'lon' : ColumnSpec('longitude', 'f'),
    'eig' : ColumnSpec('eigenvalue', 'f'),
    'prec' : ColumnSpec("precision", 'f'),
    'hwid' : ColumnSpec('bind_number', 's'),
    'yaw' : ColumnSpec('pitch', 'f'),
    'press' : ColumnSpec('airPressure', 'f'),
    'timestamp' : ColumnSpec('trigtime', 'd'),
}

def do_convert(s, fmt):
    if fmt == 's':
        return s
    if fmt == 'f':
        return float(s)
    if fmt == 'd':
        return int(s)
    if fmt == 'n':
        return "null"
    throw("what is this format?")

def data_from_row(d):
    d1 = {}
    for field in field_fmt:
        name = field_fmt[field].name
        fmt = field_fmt[field].format
        if name in d:
            st = d[name]
        elif ('\ufeff' + name) in d:
            # workaround stupid Windoge BOM
            st = d['\ufeff' + name]
        else:
            st = ''
        if st == '' and fmt != 'n':
            raise KeyError
        d1[field] = do_convert(st, fmt)
    if d1['lat'] == 0:
        d1['lat'] = None
    if d1['lon'] == 0:
        d1['lon'] = None
    return LocationData(**d1)

def parse_data_from_csv(f):
    if type(f) == str:
        with open(f, encoding='utf-8') as fd:
            return parse_data_from_csv(fd)
    from csv import DictReader
    rd = DictReader(f)
    ret = []
    for row in rd:
        try:
            entry = data_from_row(row)
            if entry.yaw < 0:
                continue
            ret.append(entry)
        except KeyError:
            print("suspicous line:", row)
    return ret

if __name__ == '__main__':
    print(parse_data_from_csv("test.csv"))