-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathcsv_parser.py
77 lines (66 loc) · 2.01 KB
/
csv_parser.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
'''
Created on Oct 28, 2016
@author: Klint
Module for Parsing CSV file
'''
import csv
CSV_EXT = '.csv'
class MissingRequiredKeyError(Exception):
pass
class CSVParseRowError(Exception):
pass
# public functions
def dictReadCsv(sFilename, setRequired=set()):
'''
returns a dictionary that matches variable names to value in CSV
'''
dictCSV = {}
bStarted = False
if not sFilename.endswith(CSV_EXT):
return
with open(sFilename, 'r') as csvfile:
try:
csvreader = csv.reader(csvfile, delimiter=',', dialect='excel')
for row in csvreader:
if bStarted:
key, value = tupleProcessRow(row)
value = fSafeCastFloat(value)
if value >= 0:
dictCSV[key] = value
else:
raise CSVParseRowError
else:
bStarted = bCheckStarted(row)
except CSVParseRowError:
print 'Row {} is invalid'.format(csvreader.line_num)
return {}
except:
print 'CSV is invalid'
return {}
if(bCheckRequired(dictCSV, setRequired)):
return dictCSV
else:
raise MissingRequiredKeyError
## private functions
def bCheckStarted(row):
return len(row) >= 3 and row[0] == 'KEY' and row[1] == 'VAL' and row[2] == 'NOTE'
def tupleProcessRow(row):
if len(row) >= 2:
return (row[0], row[1])
else:
raise MissingRequiredKeyError
def bCheckRequired(dictCSV, setRequired):
return setRequired.issubset(dictCSV.keys())
def fSafeCastFloat(s):
try:
return float(s)
except ValueError:
return -1
if __name__ == '__main__':
print dictReadCsv('test.csv')
try:
print dictReadCsv('test.csv', {'STANDBY'})
except MissingRequiredKeyError:
print 'MissingRequiredKeyError'
print dictReadCsv('test.csv', {'PAPS_ACTIVE'})
print dictReadCsv('test.csv', {'PAPS_ACTIVE', 'PAPS_STANDBY'})