csv_parser.py 1.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  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, yaw_col):
  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 yaw_col is None:
  35. d1['yaw'] = 0
  36. else:
  37. d1['yaw'] = d[yaw_col]
  38. if d1['lat'] == 0:
  39. d1['lat'] = None
  40. if d1['lon'] == 0:
  41. d1['lon'] = None
  42. return LocationData(**d1)
  43. def parse_data_from_csv(f, yaw_col = None):
  44. if type(f) == str:
  45. with open(f) as fd:
  46. return parse_data_from_csv(fd)
  47. from csv import DictReader
  48. rd = DictReader(f)
  49. ret = []
  50. for row in rd:
  51. try:
  52. ret.append(data_from_row(row, yaw_col))
  53. except KeyError:
  54. print("suspicous line:", row)
  55. return ret
  56. if __name__ == '__main__':
  57. print(parse_data_from_csv("test.csv"))