csv_parser.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364
  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. if fmt == 'n':
  19. return "null"
  20. throw("what is this format?")
  21. def data_from_row(d):
  22. d1 = {}
  23. for field in field_fmt:
  24. name = field_fmt[field].name
  25. fmt = field_fmt[field].format
  26. if name in d:
  27. st = d[name]
  28. elif ('\ufeff' + name) in d:
  29. # workaround stupid Windoge BOM
  30. st = d['\ufeff' + name]
  31. else:
  32. st = ''
  33. if st == '' and fmt != 'n':
  34. raise KeyError
  35. d1[field] = do_convert(st, fmt)
  36. if d1['lat'] == 0:
  37. d1['lat'] = None
  38. if d1['lon'] == 0:
  39. d1['lon'] = None
  40. return LocationData(**d1)
  41. def parse_data_from_csv(f):
  42. if type(f) == str:
  43. with open(f, encoding='utf-8') as fd:
  44. return parse_data_from_csv(fd)
  45. from csv import DictReader
  46. rd = DictReader(f)
  47. ret = []
  48. for row in rd:
  49. try:
  50. entry = data_from_row(row)
  51. if entry.yaw < 0:
  52. continue
  53. ret.append(entry)
  54. except KeyError:
  55. print("suspicous line:", row)
  56. return ret
  57. if __name__ == '__main__':
  58. print(parse_data_from_csv("test.csv"))