csv_parser.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859
  1. from data import LocationData
  2. from collections import namedtuple
  3. ColumnSpec = namedtuple("ColumnSpec", "name format")
  4. field_fmt = {
  5. 'lat' : ColumnSpec('latitude', 'f'),
  6. 'lon' : ColumnSpec('longitude', 'f'),
  7. 'eig' : ColumnSpec('eigenvalue', 'f'),
  8. 'prec' : ColumnSpec("precision", 'f'),
  9. 'hwid' : ColumnSpec('bind_number', 's'),
  10. 'yaw' : ColumnSpec('pitch', 'f'),
  11. 'press' : ColumnSpec('airPressure', 'f'),
  12. }
  13. def do_convert(s, fmt):
  14. if fmt == 's':
  15. return s
  16. if fmt == 'f':
  17. return float(s)
  18. throw("what is this format?")
  19. def data_from_row(d):
  20. d1 = {}
  21. for field in field_fmt:
  22. name = field_fmt[field].name
  23. fmt = field_fmt[field].format
  24. if name in d:
  25. st = d[name]
  26. elif ('\ufeff' + name) in d:
  27. # workaround stupid Windoge BOM
  28. st = d['\ufeff' + name]
  29. else:
  30. st = ''
  31. if st == '':
  32. raise KeyError
  33. d1[field] = do_convert(st, fmt)
  34. if d1['lat'] == 0:
  35. d1['lat'] = None
  36. if d1['lon'] == 0:
  37. d1['lon'] = None
  38. return LocationData(**d1)
  39. def parse_data_from_csv(f):
  40. if type(f) == str:
  41. with open(f) as fd:
  42. return parse_data_from_csv(fd)
  43. from csv import DictReader
  44. rd = DictReader(f)
  45. ret = []
  46. for row in rd:
  47. try:
  48. ret.append(data_from_row(row))
  49. except KeyError:
  50. print("suspicous line:", row)
  51. return ret
  52. if __name__ == '__main__':
  53. print(parse_data_from_csv("test.csv"))