Mercurial Hosting > traffic-intelligence

comparison trafficintelligence/events.py @ 1146:b219d5a1bb55

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
added code to categorize interactions
author Nicolas Saunier <nicolas.saunier@polymtl.ca>
date Wed, 29 Apr 2020 01:09:55 -0400
parents 6a8fe3ed3bc6
children 8c0ec7e1eb8e
comparison
equal deleted inserted replaced
1145:66f063ca2d24 1146:b219d5a1bb55
92 's', 92 's',
93 's'] 93 's']
94 94
95 timeIndicators = ['Time to Collision', 'predicted Post Encroachment Time'] 95 timeIndicators = ['Time to Collision', 'predicted Post Encroachment Time']
96 96
97 def __init__(self, num = None, timeInterval = None, roaduserNum1 = None, roaduserNum2 = None, roadUser1 = None, roadUser2 = None, categoryNum = None): 97 def __init__(self, num = None, timeInterval = None, roaduserNum1 = None, roaduserNum2 = None, roadUser1 = None, roadUser2 = None):
98 moving.STObject.__init__(self, num, timeInterval) 98 moving.STObject.__init__(self, num, timeInterval)
99 if timeInterval is None and roadUser1 is not None and roadUser2 is not None: 99 if timeInterval is None and roadUser1 is not None and roadUser2 is not None:
100 self.timeInterval = roadUser1.commonTimeInterval(roadUser2) 100 self.timeInterval = roadUser1.commonTimeInterval(roadUser2)
101 self.roadUser1 = roadUser1 101 self.roadUser1 = roadUser1
102 self.roadUser2 = roadUser2 102 self.roadUser2 = roadUser2
104 self.roadUserNumbers = set([roaduserNum1, roaduserNum2]) 104 self.roadUserNumbers = set([roaduserNum1, roaduserNum2])
105 elif roadUser1 is not None and roadUser2 is not None: 105 elif roadUser1 is not None and roadUser2 is not None:
106 self.roadUserNumbers = set([roadUser1.getNum(), roadUser2.getNum()]) 106 self.roadUserNumbers = set([roadUser1.getNum(), roadUser2.getNum()])
107 else: 107 else:
108 self.roadUserNumbers = None 108 self.roadUserNumbers = None
109 self.categoryNum = categoryNum
110 self.indicators = {} 109 self.indicators = {}
111 self.interactionInterval = None 110 self.interactionInterval = None
112 # list for collison points and crossing zones 111 # list for collison points and crossing zones
113 self.collisionPoints = None 112 self.collisionPoints = None
114 self.crossingZones = None 113 self.crossingZones = None
155 values = {} 154 values = {}
156 for k, indicator in self.indicators.items(): 155 for k, indicator in self.indicators.items():
157 values[k] = indicator[instant] 156 values[k] = indicator[instant]
158 return values 157 return values
159 158
160 def plot(self, options = '', withOrigin = False, timeStep = 1, withFeatures = False, **kwargs): 159 def plot(self, options = '', withOrigin = False, timeStep = 1, withFeatures = False, restricted = True, **kwargs):
161 self.roadUser1.plot(options, withOrigin, timeStep, withFeatures, **kwargs) 160 if restricted:
162 self.roadUser2.plot(options, withOrigin, timeStep, withFeatures, **kwargs) 161 self.roadUser1.getObjectInTimeInterval(self.timeInterval).plot(options, withOrigin, timeStep, withFeatures, **kwargs)
162 self.roadUser2.getObjectInTimeInterval(self.timeInterval).plot(options, withOrigin, timeStep, withFeatures, **kwargs)
163 else:
164 self.roadUser1.plot(options, withOrigin, timeStep, withFeatures, **kwargs)
165 self.roadUser2.plot(options, withOrigin, timeStep, withFeatures, **kwargs)
163 166
164 def plotOnWorldImage(self, nPixelsPerUnitDistance, options = '', withOrigin = False, timeStep = 1, **kwargs): 167 def plotOnWorldImage(self, nPixelsPerUnitDistance, options = '', withOrigin = False, timeStep = 1, **kwargs):
165 self.roadUser1.plotOnWorldImage(nPixelsPerUnitDistance, options, withOrigin, timeStep, **kwargs) 168 self.roadUser1.plotOnWorldImage(nPixelsPerUnitDistance, options, withOrigin, timeStep, **kwargs)
166 self.roadUser2.plotOnWorldImage(nPixelsPerUnitDistance, options, withOrigin, timeStep, **kwargs) 169 self.roadUser2.plotOnWorldImage(nPixelsPerUnitDistance, options, withOrigin, timeStep, **kwargs)
167 170
177 else: 180 else:
178 print('Please set the interaction road user attributes roadUser1 and roadUser1 through the method setRoadUsers') 181 print('Please set the interaction road user attributes roadUser1 and roadUser1 through the method setRoadUsers')
179 182
180 def computeIndicators(self): 183 def computeIndicators(self):
181 '''Computes the collision course cosine only if the cosine is positive''' 184 '''Computes the collision course cosine only if the cosine is positive'''
182 collisionCourseDotProducts = {}#[0]*int(self.timeInterval.length()) 185 collisionCourseDotProducts = {}
183 collisionCourseAngles = {} 186 collisionCourseAngles = {}
184 velocityAngles = {} 187 velocityAngles = {}
185 distances = {}#[0]*int(self.timeInterval.length()) 188 distances = {}
186 speedDifferentials = {} 189 speedDifferentials = {}
187 interactionInstants = [] 190 interactionInstants = []
188 for instant in self.timeInterval: 191 for instant in self.timeInterval:
189 deltap = self.roadUser1.getPositionAtInstant(instant)-self.roadUser2.getPositionAtInstant(instant) 192 deltap = self.roadUser1.getPositionAtInstant(instant)-self.roadUser2.getPositionAtInstant(instant)
190 v1 = self.roadUser1.getVelocityAtInstant(instant) 193 v1 = self.roadUser1.getVelocityAtInstant(instant)
214 minDistances={} 217 minDistances={}
215 for instant in self.timeInterval: 218 for instant in self.timeInterval:
216 minDistances[instant] = moving.MovingObject.minDistance(self.roadUser1, self.roadUser2, instant) 219 minDistances[instant] = moving.MovingObject.minDistance(self.roadUser1, self.roadUser2, instant)
217 self.addIndicator(indicators.SeverityIndicator(Interaction.indicatorNames[3], minDistances, mostSevereIsMax = False)) 220 self.addIndicator(indicators.SeverityIndicator(Interaction.indicatorNames[3], minDistances, mostSevereIsMax = False))
218 221
219 def categorize(self): 222 # categories = {'headon': 0,
220 '''Computes the interaction category by instant ''' 223 # 'rearend': 1,
224 # 'side': 2,
225 # 'parallel': 3}
226
227 def categorize(self, velocityAngleTolerance, parallelAngleTolerance):
228 '''Computes the interaction category by instant
229 velocityAngleTolerance and parallelAngleTolerance in radian'''
230 parallelAngleToleranceCosine = np.cos(parallelAngleTolerance)
231 self.categories = {}
232 collisionCourseDotProducts = self.getIndicator(Interaction.indicatorNames[0])
233 velocityAngles = self.getIndicator(Interaction.indicatorNames[4])
234 for instant in self.timeInterval:
235 if velocityAngles[instant] < velocityAngleTolerance: # parallel or rear end
236 midVelocity = self.roadUser1.getVelocityAtInstant(instant) + self.roadUser2.getVelocityAtInstant(instant)
237 deltap = self.roadUser1.getPositionAtInstant(instant)-self.roadUser2.getPositionAtInstant(instant)
238 if moving.Point.dot(midVelocity, deltap) < parallelAngleToleranceCosine:
239 self.categories[instant] = Interaction.categories["parallel"]
240 else:
241 self.categories[instant] = Interaction.categories["rearend"]
242 elif velocityAngles[instant] > np.pi - velocityAngleTolerance: # head on
243 if collisionCourseDotProducts[instant] > 0:
244 self.categories[instant] = Interaction.categories["headon"]
245 else:
246 if collisionCourseDotProducts[instant] > 0:
247 self.categories[instant] = Interaction.categories["side"]
221 248
222 def computeCrossingsCollisions(self, predictionParameters, collisionDistanceThreshold, timeHorizon, computeCZ = False, debug = False, timeInterval = None): 249 def computeCrossingsCollisions(self, predictionParameters, collisionDistanceThreshold, timeHorizon, computeCZ = False, debug = False, timeInterval = None):
223 '''Computes all crossing and collision points at each common instant for two road users. ''' 250 '''Computes all crossing and collision points at each common instant for two road users. '''
224 TTCs = {} 251 TTCs = {}
225 collisionProbabilities = {} 252 collisionProbabilities = {}