Mercurial > hg > nsaunier > traffic-intelligence

comparison python/storage.py @ 215:5e2983b05d4e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
created first doctest tests for storage
author Nicolas Saunier <nicolas.saunier@polymtl.ca>
date Fri, 08 Jun 2012 18:44:27 -0400
parents ce44605f888a
children b5772df11b37
comparison
equal deleted inserted replaced
214:9c7fc6899c0e 215:5e2983b05d4e
54 connection = sqlite3.connect(filename) 54 connection = sqlite3.connect(filename)
55 cursor = connection.cursor() 55 cursor = connection.cursor()
56 56
57 try: 57 try:
58 cursor.execute('SELECT * from prototypes order by prototype_id, trajectory_id_matched') 58 cursor.execute('SELECT * from prototypes order by prototype_id, trajectory_id_matched')
59 except sqlite3.OperationalError: 59 except sqlite3.OperationalError as err:
60 return matched_indexes 60 print('DB Error: {0}'.format(err))
61 return []
61 62
62 for row in cursor: 63 for row in cursor:
63 matched_indexes.append((row[0],row[1])) 64 matched_indexes.append((row[0],row[1]))
64 65
65 connection.close() 66 connection.close()
66 return matched_indexes 67 return matched_indexes
67
68 def testloadPrototypeMatchIndexesFromSqlite():
69 'TODO: write as doctest'
70 empty_list = loadPrototypeMatchIndexesFromSqlite("bidon")
71 if empty_list == []:
72 print "Empty list test Ok"
73
74 matches=loadPrototypeMatchIndexesFromSqlite("/home/francois/Unison/École/12Été/CRSNG/TAAM-Experiments/resultats/prototypes-with-matches.sqlite")
75 if len(matches) == 66:
76 print "Matches test Ok"
77 return matches
78
79 68
80 def loadTrajectoriesFromSqlite(filename, trajectoryType, objectNumbers = -1): 69 def loadTrajectoriesFromSqlite(filename, trajectoryType, objectNumbers = -1):
81 '''Loads nObjects or the indices in objectNumbers from the database 70 '''Loads nObjects or the indices in objectNumbers from the database
82 TODO: load velocities (replace table name 'positions' by 'velocities' 71 TODO: load velocities (replace table name 'positions' by 'velocities'
83 TODO: load features as well, other ways of averaging trajectories 72 TODO: load features as well, other ways of averaging trajectories
85 import sqlite3 74 import sqlite3
86 75
87 connection = sqlite3.connect(filename) # add test if it open 76 connection = sqlite3.connect(filename) # add test if it open
88 cursor = connection.cursor() 77 cursor = connection.cursor()
89 78
90 if trajectoryType == 'feature': 79 try:
91 if type(objectNumbers) == int: 80 if trajectoryType == 'feature':
92 if objectNumbers == -1: 81 if type(objectNumbers) == int:
93 cursor.execute('SELECT * from positions order by trajectory_id, frame_number') 82 if objectNumbers == -1:
83 cursor.execute('SELECT * from positions order by trajectory_id, frame_number')
94 else: 84 else:
95 cursor.execute('SELECT * from positions where trajectory_id between 0 and {0} order by trajectory_id, frame_number'.format(objectNumbers)) 85 cursor.execute('SELECT * from positions where trajectory_id between 0 and {0} order by trajectory_id, frame_number'.format(objectNumbers))
96 elif type(objectNumbers) == list: 86 elif type(objectNumbers) == list:
97 cursor.execute('SELECT * from positions where trajectory_id in ('+', '.join([str(n) for n in objectNumbers])+') order by trajectory_id, frame_number') 87 cursor.execute('SELECT * from positions where trajectory_id in ('+', '.join([str(n) for n in objectNumbers])+') order by trajectory_id, frame_number')
98 elif trajectoryType == 'object': 88 elif trajectoryType == 'object':
99 if type(objectNumbers) == int: 89 if type(objectNumbers) == int:
100 if objectNumbers == -1: 90 if objectNumbers == -1:
101 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id group by object_id, frame_number') 91 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id group by object_id, frame_number')
102 else: 92 else:
103 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id and trajectory_id between 0 and {0} group by object_id, frame_number'.format(objectNumbers)) 93 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id and trajectory_id between 0 and {0} group by object_id, frame_number'.format(objectNumbers))
104 elif type(objectNumbers) == list: 94 elif type(objectNumbers) == list:
105 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id and trajectory_id in ('+', '.join([str(n) for n in objectNumbers])+') group by object_id, frame_number') 95 cursor.execute('SELECT OF.object_id, P.frame_number, avg(P.x_coordinate), avg(P.y_coordinate) from positions P, objects_features OF where P.trajectory_id = OF.trajectory_id and trajectory_id in ('+', '.join([str(n) for n in objectNumbers])+') group by object_id, frame_number')
106 else: 96 else:
107 print('no trajectory type was chosen') 97 print('no trajectory type was chosen')
108 98 except sqlite3.OperationalError as err:
99 print('DB Error: {0}'.format(err))
100 return []
101
109 objId = -1 102 objId = -1
110 obj = None 103 obj = None
111 objects = [] 104 objects = []
112 for row in cursor: 105 for row in cursor:
113 if row[0] != objId: 106 if row[0] != objId:
207 200
208 print nObjectsPerType 201 print nObjectsPerType
209 202
210 out.close() 203 out.close()
211 204
212 205 if __name__ == "__main__":
213 206 import doctest
214 207 import unittest
215 # if __name__ == "__main__": 208 suite = doctest.DocFileSuite('tests/storage.txt')
216 # import doctest 209 unittest.TextTestRunner().run(suite)
217 # import unittest
218 # suite = doctest.DocFileSuite('tests/ubc_utils.txt')
219 # unittest.TextTestRunner().run(suite)
220 # #doctest.testmod() 210 # #doctest.testmod()
221 # #doctest.testfile("example.txt") 211 # #doctest.testfile("example.txt")