Mercurial > hg > nsaunier > traffic-intelligence

comparison trafficintelligence/events.py @ 1267:ad60e5adf084

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
cleaned interaction categorization and added stationary category
author Nicolas Saunier <nicolas.saunier@polymtl.ca>
date Wed, 29 May 2024 09:52:42 -0400
parents ebb18043616e
children ca70a79688ae
comparison
equal deleted inserted replaced
1266:ebb18043616e 1267:ad60e5adf084
194 cvutils.displayTrajectories(videoFilename, [self.roadUser1, self.roadUser2], homography = homography, firstFrameNum = firstFrameNum, lastFrameNumArg = lastFrameNum, undistort = undistort, intrinsicCameraMatrix = intrinsicCameraMatrix, distortionCoefficients = distortionCoefficients, undistortedImageMultiplication = undistortedImageMultiplication) 194 cvutils.displayTrajectories(videoFilename, [self.roadUser1, self.roadUser2], homography = homography, firstFrameNum = firstFrameNum, lastFrameNumArg = lastFrameNum, undistort = undistort, intrinsicCameraMatrix = intrinsicCameraMatrix, distortionCoefficients = distortionCoefficients, undistortedImageMultiplication = undistortedImageMultiplication)
195 else: 195 else:
196 print('Please set the interaction road user attributes roadUser1 and roadUser1 through the method setRoadUsers') 196 print('Please set the interaction road user attributes roadUser1 and roadUser1 through the method setRoadUsers')
197 197
198 def computeIndicators(self): 198 def computeIndicators(self):
199 '''Computes the collision course cosine only if the cosine is positive''' 199 '''Computes all cinematic indicators but the expensive safety indicators (TTC, PET)'''
200 collisionCourseDotProducts = {} 200 collisionCourseDotProducts = {}
201 collisionCourseAngles = {} 201 collisionCourseAngles = {}
202 velocityAngles = {} 202 velocityAngles = {}
203 distances = {} 203 distances = {}
204 speedDifferentials = {} 204 speedDifferentials = {}
231 self.addIndicator(indicators.SeverityIndicator(Interaction.indicatorNames[3], minDistances, mostSevereIsMax = False)) 231 self.addIndicator(indicators.SeverityIndicator(Interaction.indicatorNames[3], minDistances, mostSevereIsMax = False))
232 232
233 def categorize(self, velocityAngleTolerance, parallelAngleTolerance, headonCollisionCourseAngleTolerance = None, speedThreshold = 0.): 233 def categorize(self, velocityAngleTolerance, parallelAngleTolerance, headonCollisionCourseAngleTolerance = None, speedThreshold = 0.):
234 '''Computes the interaction category by instant 234 '''Computes the interaction category by instant
235 all 3 angle arguments in radian 235 all 3 angle arguments in radian
236 velocityAngleTolerance: indicates the angle threshold for rear and head on (180-velocityAngleTolerance), as well as the maximum collision course angle for head on (if headonCollisionCourseAngleTolerance is None) 236 velocityAngleTolerance: indicates the angle threshold for rear and head on (180-velocityAngleTolerance),
237 parallelAngleTolerance: indicates the angle between velocity vector (average for parallel) and position vector 237 as well as the maximum collision course angle for head on (if headonCollisionCourseAngleTolerance is None)
238 parallelAngleTolerance: indicates the tolerance on the expected 90 deg angle
239 between velocity vector (average for parallel) and position vector for a parallel interaction
240 speedThreshold defines stationary users: a stationary interaction is between one moving and
241 one stationary user, and their distance decreases
238 242
239 an instant may not be categorized if it matches the side definition (angle) 243 an instant may not be categorized if it matches the side definition (angle)
240 but the distance is growing (at least one user is probably past the point of trajectory crossing)''' 244 but the distance is growing (at least one user is probably past the point of trajectory crossing)'''
241 parallelAngleToleranceCosine = np.cos(parallelAngleTolerance) 245 minParallelAngleCosine = np.cos(np.pi/2+parallelAngleTolerance)
246 maxParallelAngleCosine = np.cos(np.pi/2-parallelAngleTolerance)
242 if headonCollisionCourseAngleTolerance is None: 247 if headonCollisionCourseAngleTolerance is None:
243 headonCollisionCourseAngleTolerance = velocityAngleTolerance 248 headonCollisionCourseAngleTolerance = velocityAngleTolerance
244 speedThreshold2 = speedThreshold**2 249 speedThreshold2 = speedThreshold**2
245 250
246 self.categories = {} 251 self.categories = {}
247 collisionCourseDotProducts = self.getIndicator(Interaction.indicatorNames[0]) 252 collisionCourseDotProducts = self.getIndicator(Interaction.indicatorNames[0])
248 collisionCourseAngles = self.getIndicator(Interaction.indicatorNames[1]) 253 collisionCourseAngles = self.getIndicator(Interaction.indicatorNames[1])
249 distances = self.getIndicator(Interaction.indicatorNames[2]) 254 distances = self.getIndicator(Interaction.indicatorNames[2])
250 velocityAngles = self.getIndicator(Interaction.indicatorNames[4]) 255 velocityAngles = self.getIndicator(Interaction.indicatorNames[4])
251 for instant in self.timeInterval: 256 for instant in self.timeInterval:
252 if instant in velocityAngles: 257 stationaryUser1 = self.roadUser1.getVelocityAtInstant(instant).norm2Squared() <= speedThreshold2
258 stationaryUser2 = self.roadUser2.getVelocityAtInstant(instant).norm2Squared() <= speedThreshold2
259 if stationaryUser1 != stationaryUser2 and collisionCourseDotProducts[instant] > 0: # only one is not moving and is getting closer
260 self.categories[instant] = Interaction.categories["stationary"]
261 # true stationary would be all the times (parked, difficult without semantic knowledge of the scene
262 # alternatively, one could get the previous or next non zero velocity to identify user orientation
263 elif velocityAngles.existsAtInstant(instant):
253 if velocityAngles[instant] < velocityAngleTolerance: # parallel or rear end 264 if velocityAngles[instant] < velocityAngleTolerance: # parallel or rear end
254 midVelocity = self.roadUser1.getVelocityAtInstant(instant) + self.roadUser2.getVelocityAtInstant(instant) 265 midVelocity = self.roadUser1.getVelocityAtInstant(instant) + self.roadUser2.getVelocityAtInstant(instant)
255 deltap = self.roadUser1.getPositionAtInstant(instant)-self.roadUser2.getPositionAtInstant(instant) 266 deltap = self.roadUser1.getPositionAtInstant(instant)-self.roadUser2.getPositionAtInstant(instant)
256 if abs(moving.Point.dot(midVelocity, deltap)/(midVelocity.norm2()*distances[instant])) < parallelAngleToleranceCosine: 267 if minParallelAngleCosine < abs(moving.Point.dot(midVelocity, deltap)/(midVelocity.norm2()*distances[instant])) < maxParallelAngleCosine:
257 self.categories[instant] = Interaction.categories["parallel"] 268 self.categories[instant] = Interaction.categories["parallel"]
258 else: 269 else:
259 self.categories[instant] = Interaction.categories["rearend"] 270 self.categories[instant] = Interaction.categories["rearend"]
260 elif velocityAngles[instant] > np.pi - velocityAngleTolerance and collisionCourseAngles[instant] < headonCollisionCourseAngleTolerance: # head on 271 elif velocityAngles[instant] > np.pi - velocityAngleTolerance and collisionCourseAngles[instant] < headonCollisionCourseAngleTolerance: # head on
261 self.categories[instant] = Interaction.categories["headon"] 272 self.categories[instant] = Interaction.categories["headon"]
262 elif collisionCourseDotProducts[instant] > 0: 273 elif collisionCourseDotProducts[instant] > 0:
263 self.categories[instant] = Interaction.categories["side"] 274 self.categories[instant] = Interaction.categories["side"]
264 # true stationary is when object does not move for the whole period of the interaction, otherwise get last (or next) velocity vector for user orientation 275 # true stationary is when object does not move for the whole period of the interaction, otherwise get last (or next) velocity vector for user orientation
265 # if instant not in self.categories: # if it's none of the other categories (could be with almost stationary vehicle) and only one speed is 0 276 # leaving is not a good interaction category (issue in Etienne's 2022 paper):
266 # stationaryUser1 = self.roadUser1.getVelocityAtInstant(instant).norm2Squared() <= speedThreshold2 277 # means we are past the situation in which users are approaching
267 # stationaryUser2 = self.roadUser2.getVelocityAtInstant(instant).norm2Squared() <= speedThreshold2
268 # if stationaryUser1 != stationaryUser2 and collisionCourseDotProducts[instant] > 0: # only one is not moving
269 # self.categories[instant] = Interaction.categories["stationary"]
270 # leaving is not a good interaction category (issue in Etienne's 2022 paper): means we are past the situation in which users are approaching
271 # could try to predict what happened before, but it's not observed 278 # could try to predict what happened before, but it's not observed
272 279
273 280
274 def computeCrossingsCollisions(self, predictionParameters, collisionDistanceThreshold, timeHorizon, computeCZ = False, debug = False, timeInterval = None): 281 def computeCrossingsCollisions(self, predictionParameters, collisionDistanceThreshold, timeHorizon, computeCZ = False, debug = False, timeInterval = None):
275 '''Computes all crossing and collision points at each common instant for two road users. ''' 282 '''Computes all crossing and collision points at each common instant for two road users. '''