Mercurial Hosting > traffic-intelligence
comparison python/storage.py @ 215:5e2983b05d4e
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") |