import sys from abc import ABC, abstractmethod import random, itertools, math, numpy from contextlib import suppress from copy import deepcopy from operator import attrgetter #=====# import Logger import MinorFunctions as mf #----- Debug -----# RandomCount = 0 def RandomChoices(choices, probabilities): global RandomCount RandomCount += 1 return random.choices(choices, probabilities) def RandomUniform(firstValue, secondValue): global RandomCount RandomCount += 1 return random.uniform(firstValue, secondValue) #----- Variable abbreviations -----# # - opp = opponent # - ass = assumed #----- Global settings -----# numpy.set_printoptions(legacy='1.25') #----- Global Dictionaries -----# ATTRIBUTE_WEIGHTS_PERSONALITY = { # [FullRange, Low, Medium, High, Autonomous] 'Unknown': { 'General': [1, 0,0,0], 'condition': [1, 0,0,0,0] }, 'Normal': { 'General': [0, 15,70,15], 'condition': [0, 15,50,30,5] }, 'Cautious': { 'General': [0, 60,20,20], 'dc': [0, 30,40,30], 'condition': [0, 35,45,15,5] }, 'Dangerous': { 'General': [0, 20,20,60], 'condition': [0, 20,25,30,25] }, 'Autonomous': { 'General': [0, 60,30,10], 'condition': [0, 50,40,9,1] #'DT': [0, 0,0,0, 1] } } ATTRIBUTE_WEIGHTS_CONDITION = { 'Unknown': { 'General': [1, 0,0,0] }, 'Spotless': { 'General': [0, 65,25,10] }, 'Average': { 'General': [0, 15,60,25] }, 'Dirty': { 'General': [0, 65,25,10] }, 'Damaged': { 'General': [0, 65,25,10] } } ATTRIBUTE_BOUNDS = { #All of these will need verifying with evidence (and citations) # For now, I am using the same value ranges as VEHITS 2024 'ac': { # Comfortable acceleration 'FullRange': [0.20, 2.00], # VEHITS: [0.20, 2.00] 'Low': [1.25, 2.00], 'Medium': [1.75, 2.50], 'High': [2.25, 3.25] }, 'amax|condition': { # Maximum acceleration, based on Vehicle.condition 'FullRange': [2.50, 3.50], # VEHITS: [2.50, 3.50] 'Low': [1.70, 3.30], 'Medium': [2.80, 3.25], 'High': [3.40, 5.00] }, 'dc': { #Comfortable deceleration 'FullRange': [-0.50, -1.50], 'Low': [-0.50, -0.95], 'Medium': [-0.80, -1.25], 'High': [-1.10, -1.50] }, 'dmax|condition': { # Maximum deceleration, based on Vehicle.condition 'FullRange': [-4.50, -4.50], # VEHITS: [-4.50, -4.50] 'Low': [-3.00, -4.35], 'Medium': [-3.90, -5.25], 'High': [-4.80, -6.00] }, 'HWmin': { # Minimum acceptable headway 'FullRange': [0.50, 3.50], 'Low': [0.50, 1.80], 'Medium': [1.40, 2.70], 'High': [2.30, 3.50] }, 'DT': { #Decision/reaction time 'FullRange': [0.50, 1.50], 'Low': [0.50, 0.95], 'Medium': [0.80, 1.25], 'High': [1.10, 1.50] #'Autonomous': [0.10, 0.30] }, 'v0': { # Initial speed 'FullRange': [8, 18], # VEHITS: [8, 18] 'Low': [8, 12], 'Medium': [10, 16], 'High': [14, 20] }, # Only applicable to VEHITS 2024 'v0|J': { 'FullRange': [4, 10] # VEHITS: [4, 10] }, 'angryAlpha': { # Punishment severity 'FullRange': [0.35, 0.15], # VEHITS: [0.35, 0.15] 'Low': [0.65, 0.45], 'Medium': [0.50, 0.30], 'High': [0.35, 0.20] }, 'waitFactor': { # Wait penalty sensitivity 'FullRange': [0.10, 0.20], # VEHITS: [0.10, 0.20] 'Low': [0.05, 0.20], 'Medium': [0.15, 0.30], 'High': [0.25, 0.40] } } PAYOFF_WEIGHTS = {'a': 1, 'hw': 1, 'v': 1, 't': 1} POPULATION_PRIORS = {"Attentive": 0.75, "Cooperative": 0.60, "Distracted": 0.25, "Punitive": 0.40} SCENARIOS = { 0: 'Default', 1: 'Implicit Only', 2: 'Eye-contact Only', 3: 'Co-op Only', 4: 'Full Communication', 5: 'Feign Distraction', 6: 'Two-Way Communication', 7: 'Two-Way Force Only', 8: 'Two-Way No Force' } def generateAttribute(vehicle, attribute, personality=None): if not personality: personality = vehicle.personality if '|condition' in attribute: WEIGHTS = ATTRIBUTE_WEIGHTS_CONDITION attributeSelector = vehicle.condition else: WEIGHTS = ATTRIBUTE_WEIGHTS_PERSONALITY attributeSelector = personality if attribute in ATTRIBUTE_BOUNDS.keys(): bounds = ATTRIBUTE_BOUNDS.get( attribute + ('|J' if vehicle.position == 'Joining' and attribute == 'v0' else '') ).get( RandomChoices( list(ATTRIBUTE_BOUNDS.get(attribute).keys()), WEIGHTS.get(attributeSelector).get(attribute) or \ WEIGHTS.get(attributeSelector).get('General') )[0] ); attributeValue = RandomUniform(bounds[0], bounds[1]) if attribute == 'condition': return RandomChoices( list(ATTRIBUTE_WEIGHTS_CONDITION.keys()), WEIGHTS.get(personality).get(attribute) or \ WEIGHTS.get(personality).get('General') )[0] if attribute == 'DT' and personality == 'Autonomous': return RandomUniform(0.10, 0.30) if attribute == 'v0' and vehicle.position=='Joining': #VEHITS 2024 behaviour: pass #return 0.5 * attributeValue if attribute == 's0': return 0 if vehicle.position=='Mainline' else (RandomUniform(5+9, 80+9)) # + (7.5 if vehicle.scenario >= 6 else 0)) # was 5 - 90 if attribute == 'attentive': return True if personality == 'Autonomous' else RandomChoices([True,False],[0.75, 0.25])[0] if attribute == 'cooperative': # JUSTIFY return True if personality == 'Autonomous' else RandomChoices([True,False],[0.60, 0.40])[0] if attribute == 'vD': return RandomUniform(0.75, 1.50) * vehicle.v0 if attribute == 'ac' or attribute == 'dc': # For VEHITS 2024 use pass pass #return RandomUniform(0.5, 0.8) * getattr(vehicle, attribute[0]+'max') return attributeValue #----- Global Constants -----# ABS_MAX_DEC = -4.50 #https://hypertextbook.com/facts/2001/MeredithBarricella.shtml / https://copradar.com/chapts/references/acceleration.html - Maximum average driver deceleration LANECHANGE_DURATION = 5.00 #FIND SOURCE CRASH_PENALTY = -250 #TURN_RADIUS = 11.8 #https://www.standardsforhighways.co.uk/search/html/962a81c1-abda-4424-96c9-fe4c2287308c?standard=DMRB - NOT IN USE class Vehicle(ABC): #Any subclasses will revolve around species, NOT position. newID = itertools.count().__next__ def __init__( self, tester=False, species='HDV', personality='Unknown', intelligence='Random', position='Mainline', spatial=False, communicative=False, scenario=0 ): self.ID = Vehicle.newID() self.scenario = scenario self.tester = tester self.species = species #[ARV, HDV] - This does not currently do anything. self.personality = personality #[Unknown, Normal, Cautious, Dangerous, Autonomous] self.intelligence = intelligence #[Random, Transparent, Mirror, Learner] self.position = position #[Mainline, Joining] self.spatial = spatial #[True, False] self.communicative = communicative #[True, False] # Incremented in self.interact() if self.scenario >= 6 and self.position == 'Joining': self.stage = 1; self.maxStage = 2 else: self.stage = 0; self.maxStage = 0 self.setAttributes() def __del__(self): #print(f"{__class__.__name__} {self.ID} has been destroyed.") pass def verbalise(self): #Verbose output of vehicle attributes print(f"{self.species} {self.ID} is a {__class__.__name__} {__class__.__base__.__name__}.\n \ It approaches at {round(self.v0,2)} m/s from {round(self.s0,2)} m away. It would prefer to keep a time headway of {round(self.HWmin,2)} seconds.") def setAttributes(self): self.opponentProperties = {} self.possible_accelerations = {} self.observable_accelerations = {} self.signals = {} # acceleration, eyes, signal self.taken_actions = [] # Used to collect actions taken by the vehicle through an interaction self.alpha = 1.00; self.m = 1.00; self.l = 1.00 #Might customise later. self.a0 = 0; self.a1 = 0 if self.position == 'Mainline': self.a2 = 0 # attributes are iterated in the order they are declared below. The order is important for attributes which depend on others. attribute_list = ['condition', 'amax|condition', 'dmax|condition', 'v0', 's0', 'ac', 'dc', 'HWmin', 'DT'] if self.position == 'Mainline': attribute_list.extend(['attentive', 'cooperative', 'angryAlpha']) elif self.position == 'Joining': attribute_list.extend(['vD', 'waitFactor']) if self.tester: for attribute in attribute_list: setattr(self, attribute.split('|')[0], TESTVALUES.get(self.position[0]+'-'+attribute.split('|')[0]) or 0) #The "or 0" is to avoid having to explicitly specify M.s0 in the TESTER_VALUES as it is always 0. else: #Sources: https://s18798.pcdn.co/perception_action_lab/wp-content/uploads/sites/6482/2017/03/AyresLi2001IEEE.pdf #https://onlinepubs.trb.org/Onlinepubs/trr/1991/1303/1303-011.pdf for attribute in attribute_list: setattr(self, attribute.split('|')[0], generateAttribute(self, attribute)) def setAssumptions(self, opponent=None, predefined=False): self.opponentProperties['alpha'] = 1.00 assumption_list = ['ac', 'amax|condition', 'dc', 'dmax|condition', 'HWmin', 'DT'] Mainlines_assumptions = ['vD', 'waitFactor']; Joinings_assumptions = ['angryAlpha'] assumption_list.extend(Mainlines_assumptions if self.position == 'Mainline' else Joinings_assumptions) if self.tester: for assumption in assumption_list: self.opponentProperties[assumption.split('|')[0]] = \ TESTVALUES.get(self.position[0]+'-'+assumption.split('|')[0]+opponent.position[0]) else: if predefined: match self.intelligence: case 'Random': for assumption in assumption_list: self.opponentProperties[assumption.split('|')[0]] = \ generateAttribute(opponent, assumption, 'Unknown') #.get() defaults to 0, which is useful here. case 'Mirror': for assumption in (Mainlines_assumptions if self.position == 'Mainline' else Joinings_assumptions): self.opponentProperties[assumption.split('|')[0]] = generateAttribute(opponent, assumption, self.personality) else: match self.intelligence: #[Random, Transparent, Mirror, Learner] case 'Transparent': for assumption in assumption_list: self.opponentProperties[assumption.split('|')[0]] = \ getattr(opponent, assumption.split('|')[0]) case 'Mirror': for assumption in assumption_list: if hasattr(self, assumption.split('|')[0]): self.opponentProperties[assumption.split('|')[0]] = getattr(self, assumption.split('|')[0]) #else: -- Managed in predefined #self.opponentProperties[assumption.split('|')[0]] = generateAttribute(opponent, assumption, self.personality) case 'Learner': pass def communicate(self, include=['acceleration', 'eyes', 'signal']): #Probabiliies should be lifted out of here into a dictionary. match self.position: case 'Mainline': if 'acceleration' in include: # Below conditional statement is useful when a vehicle is "communicating" implicitly before moving a = 0 if len(self.observable_accelerations.keys()) == 0 else self.observable_accelerations.get(self.action) self.signals['acceleration'] = -int(math.copysign(a!=0,a)) if 'eyes' in include: self.signals['eyes'] = RandomChoices([1,0], [0.9, 0.1])[0] if self.attentive else RandomChoices([1,0], [0.05, 0.95])[0] else: # We use this to keep the random seed constant between different scenarios RandomChoices([0,0], [0.5,0.5]) if 'signal' in include: match [self.action, self.attentive, self.cooperative]: case ['Allow', 1, 1]: self.signals['signal'] = RandomChoices([1,0], [0.8,0.2])[0] case ['Allow', 1, 0]: self.signals['signal'] = RandomChoices([1,0], [0.2,0.8])[0] case ['Allow', 0, 1]: self.signals['signal'] = RandomChoices([1,0], [0.1,0.9])[0] case ['Allow', 0, 0]: self.signals['signal'] = RandomChoices([1,0], [0.05,0.95])[0] case ['Block', 1, 1]: self.signals['signal'] = RandomChoices([0,-1], [0.9,0.1])[0] case ['Block', 1, 0]: self.signals['signal'] = RandomChoices([0,-1], [0.2,0.8])[0] case ['Block', 0, 1]: self.signals['signal'] = RandomChoices([0,-1], [0.95,0.05])[0] case ['Block', 0, 0]: self.signals['signal'] = RandomChoices([0,-1], [0.9,0.1])[0] else: # We use this to keep the random seed constant between different scenarios RandomChoices([0,0], [0.5,0.5]) case 'Joining': pass # self.signals['any'] = 1 if self.action == 'Ask' else 0 # Not sure if we need this at all. def observe(self, opponent): self.opponentProperties['v0'] = opponent.v0 self.opponentProperties['s0'] = opponent.s0 self.opponentProperties['a0'] = opponent.a0 self.opponentProperties['a1'] = opponent.a1 if opponent.position == 'Mainline': self.opponentProperties['a2'] = opponent.a2 self.opponentProperties['__real_DT__'] = opponent.DT self.opponentProperties['stage'] = opponent.stage self.opponentProperties['signals'] = opponent.signals if opponent.position == 'Joining': self.opponentProperties['action'] = opponent.action if opponent.stage == 2 else 0 def evaluateStrategies(self, verbose=False): payoff_pairs = [] match self.position: case 'Mainline': action_pairs = [('Allow', 'Go'), ('Allow', 'Wait'), ('Block', 'Go'), ('Block', 'Wait')] for pair in action_pairs: payoff_pairs.append(moveVehicles(M=self, actionPair=pair, verbose=verbose)) exp_payoff_Allow = payoff_pairs[0][0] if payoff_pairs[0][1] >= payoff_pairs[1][1] else payoff_pairs[1][0] exp_payoff_Block = payoff_pairs[2][0] if payoff_pairs[2][1] >= payoff_pairs[3][1] else payoff_pairs[3][0] self.action = 'Allow' if exp_payoff_Allow >= exp_payoff_Block else 'Block' self.taken_actions.append(self.action) self.a1 = RandomChoices([0, self.possible_accelerations.get(self.action)], [(not self.attentive)*0.50, 0.50])[0] self.a2 = 0 if self.a1 == 0 and self.possible_accelerations.get(self.action) != self.a1 else self.possible_accelerations.get(self.action+'2') if self.a1 == 0 and self.possible_accelerations.get(self.action) != self.a1: self.observable_accelerations = {} # Resets observable accelerations if M is distracted. case 'Joining': action_pairs = [('Follow', 'Go'), ('Punish', 'Go'), ('FreeFlow|Follow', 'Go'), ('FreeFlow|Punish', 'Go'), ('FreeFlow', 'Wait')] for pair in action_pairs: payoff_pairs.append(moveVehicles(J=self, actionPair=pair, verbose=verbose)) exp_payoff_Go = payoff_pairs[0][1] + payoff_pairs[1][1] + payoff_pairs[2][1] + payoff_pairs[3][1] exp_payoff_Wait = payoff_pairs[4][1] if verbose: print("Expected Go Payoff: ", round(exp_payoff_Go, 2)) print("Expected Wait Payoff: ", round(exp_payoff_Wait, 2)) print("") self.action = 'Go' if exp_payoff_Go >= exp_payoff_Wait else 'Wait' self.taken_actions.append(self.action) if self.scenario >= 6: self.a0 = self.possible_accelerations.get(self.action) self.a1 = self.possible_accelerations.get(self.action+'1') # Stage 0 is default (no stages), i.e. scenario with only one stage will use stage 0. # For staged interactions, use stage=1 as the first stage. def interact(self, opponent, verbose=False): # Pre-interaction communication if self.communicative and self.position == 'Joining' and self.stage == 1: opponent.communicate(['eyes']) else: # We use this to keep the random seed constant between different scenarios opponent.communicate(['nothing']) self.observe(opponent) if self.stage < 2: self.setAssumptions(opponent) self.evaluateStrategies(verbose) # Post-interaction communication if self.communicative: match self.scenario: case 0: self.communicate(['nothing']) case 1: self.communicate(['acceleration']) case 2: self.communicate(['acceleration', 'eyes']) case _ if self.scenario <= 5: self.communicate() case 6: if opponent.stage == 2: self.communicate(['acceleration', 'signal']) else: # We use this to keep the random seed constant between different scenarios self.communicate(['nothing']) if self.stage < self.maxStage: self.stage +=1 #def __init__(self, tester, species, personality, intelligence, position, spatial, communicative): # self.personality = personality #[Unknown, Normal, Cautious, Dangerous, Autonomous] # self.intelligence = intelligence #[Random, Transparent, Mirror, Learner] def generateVehicles( count, tester=False, species='HDV', personality='Unknown', intelligence='Random', position='Mainline', spatial=False, communicative=False, scenario=0 ): if scenario not in SCENARIOS.keys(): sys.exit("Invalid scenario!") list_of_vehicles = [] for i in range(count): new_vehicle = Vehicle(tester, species, personality, intelligence, position, spatial, communicative, scenario) #new_vehicle.verbalise() list_of_vehicles.append(new_vehicle) return list_of_vehicles def moveVehicles(M=None, J=None, actionPair=(), verbose=False): # Drop one of the two vehicles to rely on the other's assumptions instead. def get_aM(mode): alpha = M.angryAlpha if mode == 'Punish' else M.alpha target_velocity = M.v0 if mode == 'FreeFlow' else min(vJ[i], M.v0) delta_s = (4*vM[i]) if mode == 'FreeFlow' else (sJ[i] - sM[i]) dmax = M.dmax #dmax = M.dc if mode == 'Allow' or mode == 'Block' else M.dmax # That's all you need. if i == 0: return M.a0 if J.species != 'fake' and J.stage != 1: # This is also valid during actual movement if i == 1: return M.a1 elif i == 2: return M.a2 if vM[i] == 0: return M.ac # TEST # if i == 2: # return aM[i-1] # Remove the statement to the right of the 'and' to allow M to Allow/Block during a Force if mode == 'Block' and M.opponentProperties['action'] != 'Go': # This explicitly uses assumptions because this action always relies on assumptions. a = J.amax + ( #M.opponentProperties['amax'] + ( # (2*(M.opponentProperties['s0']+(M.opponentProperties['v0']-M.v0)*LANECHANGE_DURATION))/LANECHANGE_DURATION**2 (2*((sJ[i]-sM[i])+(vJ[i]-vM[i])*LANECHANGE_DURATION))/LANECHANGE_DURATION**2 ) a = max(0, min(a, M.amax)) else: a = ((alpha*vM[i]**M.m)/(delta_s**M.l)) * (target_velocity-vM[i]) a = min(max(dmax, a), 0 if i <= 2 else M.ac) return min(M.dc, a) if (mode == 'Follow' or mode == 'Punish') and HWM[i] < M.HWmin else a def get_aJ(): if i == 0 and (J.stage == 0 or J.stage == 2): return J.a0 if i == 1 and J.stage == 2: return J.a1 # Remove the elif below if you want the model to behave like VEHITS 2024 elif i == 1 and J.stage == 0: return J.a0 if J.action == 'Wait': return J.dc if vJ[i] > J.v0 else 0 aB = ((J.alpha*vJ[i]**J.m)/((sJ[i]-sM[i])**J.l)) * (max(vM[i],J.vD)-vJ[i]) aB = min(max(0, aB), J.amax) aF = ((J.alpha*vJ[i]**J.m)/((4*vJ[i])**J.l)) * (J.vD-vJ[i]) aF = min(max(J.dmax, aF), J.ac) return aF if vJ[i] >= J.vD else aB def calculatePayoffs(): payoff_M = {'a': 0, 'hw': 0, 'v': 0, 't': 0} payoff_J = {'a': 0, 'hw': 0, 'v': 0, 't': 0} if M.species == 'fake': J.Bayesian_probabilities = mf.GetProbabilities(J.opponentProperties['signals'], POPULATION_PRIORS) if verbose: print("Signals: ", J.opponentProperties['signals']) print("Bayes: ", J.Bayesian_probabilities) match M.action: case 'Follow': bayes_prob = J.Bayesian_probabilities[0] case 'Punish': bayes_prob = J.Bayesian_probabilities[1] case 'FreeFlow|Follow': bayes_prob = J.Bayesian_probabilities[2] case 'FreeFlow|Punish': bayes_prob = J.Bayesian_probabilities[3] case 'FreeFlow': bayes_prob = 1 else: bayes_prob = 1 payoff_M['a'] = PAYOFF_WEIGHTS['a'] * (numpy.average(UaM) - numpy.std(aM)) payoff_J['a'] = PAYOFF_WEIGHTS['a'] * (bayes_prob * (numpy.average(UaJ) - numpy.std(aJ))) match J.action: case 'Go': payoff_M['hw'] = PAYOFF_WEIGHTS['hw'] * (CRASH_PENALTY if max(min(HWM), 0) == 0 else min(1-(M.HWmin/max(min(HWM), 0)), 0)) payoff_J['hw'] = PAYOFF_WEIGHTS['hw'] * ((CRASH_PENALTY if max(min(HWM), 0) == 0 else min(1-(J.HWmin/max(min(HWM), 0)), 0)) * bayes_prob) payoff_M['v'] = PAYOFF_WEIGHTS['v'] * (1-(M.v0/min(M.v0,J.vD))) case 'Wait': payoff_J['v'] = PAYOFF_WEIGHTS['v'] * (1-(J.vD/min(M.v0,J.vD))) dt_sum = sum(x[0] for x in zip(dt, HWJ) if x[1] < J.HWmin) payoff_J['t'] = -100 if J.v0 >= M.v0 else \ 1 * -J.waitFactor * ( dt_sum + ( 0 if HWJ[-1] >= J.HWmin else \ (J.HWmin*J.v0 + sJ[-1]-sM[-1]) / (0.5*(vM[0]+max(vM[-1],vM[0]))-J.v0) ) ) * PAYOFF_WEIGHTS['t'] if verbose: print("Payoffs: ", round(sum(payoff_M.values()), 2), round(sum(payoff_J.values()), 2)) print("") return (sum(payoff_M.values()), sum(payoff_J.values())) dt, vM, sM, vJ, sJ, HWM, HWJ, aM, UaM, aJ, UaJ = [[],[],[],[],[],[],[],[],[],[],[]] if not M: M = deepcopy(J); M.species = 'fake' M.action = actionPair[0] M.attentive = M.action.split('|')[0] != 'FreeFlow' M.cooperative = M.action.split('|')[-1] != 'Punish' for property in [key for key in J.opponentProperties.keys() if '__' not in key]: setattr(M, property, J.opponentProperties[property]) #if J.stage == 2: # In the second stage, J has observed M's true DT. # M.DT = J.opponentProperties['__real_DT__'] elif not J: J = deepcopy(M); J.species = 'fake' for property in [key for key in M.opponentProperties.keys() if '__' not in key]: setattr(J, property, M.opponentProperties[property]) if actionPair: M.action, J.action = actionPair # Important note: # If either vehicle is fake, then dt[1] and dt[2] are purely based on assumptions. dt.extend([0]) # ----- Use the if statement below if you want Scenarios <= 4 to behave as they did in VEHITS 2024 ----- # #if J.scenario >= 6: # J Comms On dt.extend([M.DT if M.species != 'fake' or J.stage == 1 else J.opponentProperties['__real_DT__']]) # J.stage being 1 automatically means it's doing comms if (M.species == 'fake' or J.species == 'fake') and (J.scenario < 6 or J.stage == 2): dt.extend([J.DT, M.DT]) elif M.species != 'fake' and J.species != 'fake': #real movement dt.extend([ (M.DT+M.opponentProperties['DT']) if J.DT > (M.DT+M.opponentProperties['DT']) else J.DT, abs(M.DT+M.opponentProperties['DT']-J.DT) if J.DT > M.opponentProperties['DT'] else M.DT ]) reseved_dt_length = len(dt) # Movement Begins i = -1 while True: # ----- This section is not subject to scenario or stage changes ----- # i+=1 if i >= reseved_dt_length: # (>=) because dt_length is always higher than its last index by 1 dt.append(0.50) if i == 0: vM.append(M.v0); sM.append(M.s0) vJ.append(J.v0); sJ.append(J.s0) UaM.append(0); UaJ.append(0) else: vM.append(max(0, vM[i-1]+aM[i-1]*dt[i])) sM.append(sM[i-1]+0.5*(vM[i-1]+vM[i])*dt[i]) vJ.append(max(0, vJ[i-1]+aJ[i-1]*dt[i])) sJ.append(sJ[i-1]+0.5*(vJ[i-1]+vJ[i])*dt[i]) UaM.append(min(dt[i]*(1-aM[i-1]/(M.ac if aM[i-1] >=0 else M.dc)), 0)) UaJ.append(min(dt[i]*(1-aJ[i-1]/(J.ac if aJ[i-1] >=0 else J.dc)), 0)) HWM.append(99 if vM[i] == 0 else (sJ[i]-sM[i])/vM[i]) HWJ.append(99 if vJ[i] == 0 else (sM[i]-sJ[i])/vJ[i]) # -------------------------------------------------------------------- # aJ.append(get_aJ()) if i <= 2: aM.append(get_aM(M.action.split('|')[0])) else: # The options are: before and after 12... punish, follow or freeflow # Remember that when J is fake, M is thinking, therefore its attention state is irrelevant. It will always assume attentive=1 for itself. if (not M.attentive and J.species != 'fake' and i < 10) or J.action== 'Wait': # Used to be i < 12 aM.append(get_aM('FreeFlow')) else: aM.append(get_aM('Follow' if M.cooperative or M.action == 'Allow' else 'Punish')) if i <= 2 and J.species == 'fake': M.possible_accelerations[M.action.split('|')[0] + ('2' if i == 2 else '')] = aM[i] if i <= 1 and M.species == 'fake': J.possible_accelerations[J.action + ('1' if i == 1 else '')] = aJ[i] if J.species == 'fake' and J.action == 'Wait': # We use J.action == 'Wait' because we want aM when J is waiting because when moving, M will behave as if J is waiting until it joins. if sum(dt[:i]) > M.opponentProperties['__real_DT__'] and not M.action.split('|')[0] in M.observable_accelerations.keys(): M.observable_accelerations[M.action.split('|')[0]] = aM[i-1] if ( i == 63 or ( i >= 23 and abs(aM[i]) <= 0.01 and abs(aJ[i]) <= 0.01 and (HWJ[i] >= J.HWmin if J.action=='Wait' else True) ) ) or (HWM[i] <= 0 and J.action=='Go'): break M.minimumHW = min(HWM) M.interactionDuration = sum(dt) if verbose: print((M.action, J.action), i+1) print("dt: ", [round(x, 2) for x in dt]) print("vM: ", [round(x, 2) for x in vM]) print("sM: ", [round(x, 2) for x in sM]) print("vJ: ", [round(x, 2) for x in vJ]) print("sJ: ", [round(x, 2) for x in sJ]) print("HW-M: ", [round(x, 2) for x in HWM]) print("HW-J: ", [round(x, 2) for x in HWJ]) print("aM: ", [round(x, 2) for x in aM]) print("UaM: ", [round(x, 2) for x in UaM]) print("aJ: ", [round(x, 2) for x in aJ]) print("UaJ: ", [round(x, 2) for x in UaJ]) print("") print("Minimum HW: ", round(M.minimumHW, 2)) print("Interaction duration: ", round(M.interactionDuration, 2)) if M.species == 'fake' or J.species == 'fake': print("Actions: ", M.action, J.action) else: print("Actions: ", M.taken_actions, J.taken_actions) return calculatePayoffs() def TestBench(): Mcar = generateVehicles(count=100, tester=False, personality='Autonomous', intelligence='Mirror', scenario=0, communicative=True) for car in Mcar: print(car.attentive) print(max(Mcar, key=attrgetter('DT')).DT) print(min(Mcar, key=attrgetter('DT')).DT) '''random.seed(15616516585651) activeScenario = 6 Mcar = generateVehicles(count=1, tester=True, intelligence='Mirror', scenario=activeScenario, communicative=True)[0] Jcar = generateVehicles(count=1, tester=True, intelligence='Mirror', position='Joining', scenario=activeScenario, communicative=True)[0] #Mcar = Vehicle(tester=False, intelligence='Mirror', scenario=activeScenario, communicative=True) #Jcar = Vehicle(tester=False, intelligence='Mirror', position='Joining', scenario=activeScenario, communicative=True) Mcar.setAssumptions(opponent=Jcar, predefined=True) Jcar.setAssumptions(opponent=Mcar, predefined=True) if activeScenario == 6: Jcar.interact(Mcar, verbose=True) Mcar.interact(Jcar, verbose=True) Jcar.interact(Mcar, verbose=True) Mcar.payoff, Jcar.payoff = moveVehicles(M=Mcar, J=Jcar, verbose=True) print(RandomCount) print(Jcar.opponentProperties['angryAlpha']) print('Done.')''' # TESTVALUES = { # Calcs.xlsx Tester # 'M-ac': 1.8, # 'M-amax': 2.2, # 'M-angryAlpha': 0.35, # 'M-attentive': True, # 'M-cooperative': True, # 'M-dc': -1.5, # 'M-dmax': -2.5, # 'M-DT': 0.8, # 'M-HWmin': 0.6, # 'M-s0': 0, # 'M-v0': 13.5, # 'M-acJ': 0.4, # 'M-amaxJ': 2.2, # 'M-dcJ': -1, # 'M-dmaxJ': -4, # 'M-DTJ': 1.15, # 'M-HWminJ': 2.2, # 'M-s0J': 24, # 'M-v0J': 9.5, # 'M-vDJ': 10, # 'M-waitFactorJ': 0.1, # 'J-ac': 1.2, # 'J-amax': 2, # 'J-dc': -1.5, # 'J-dmax': -3.8, # 'J-DT': 1.2, # 'J-HWmin': 1.2, # 'J-s0': 24, # 'J-v0': 9.5, # 'J-vD': 10.5, # 'J-waitFactor': 0.25, # 'J-acM': 1.4, # 'J-amaxM': 2.8, # 'J-angryAlphaM': 0.32, # 'J-dcM': -1.2, # 'J-dmaxM': -3.5, # 'J-DTM': 0.95, # 'J-HWminM': 1.5, # 'J-s0M': 0, # 'J-v0M': 13.5 # } # TESTVALUES = { # Original comparative test # 'M-ac': 0.701349711460026, # 'M-amax': 3.22246147755112, # 'M-angryAlpha': 0.317295453598947, # 'M-attentive': 0, # 'M-cooperative': True, # 'M-dc': -0.973042307388401, # 'M-dmax': -4.5, # 'M-DT': 0.637960764419036, # 'M-HWmin': 3.14598939235987, # 'M-s0': 0, # 'M-v0': 12.0312206826497, # 'M-acJ': 1.02741561224601, # 'M-amaxJ': 3.45218265092789, # 'M-dcJ': -1.05378334461993, # 'M-dmaxJ': -4.5, # 'M-DTJ': 1.44460006453038, # 'M-HWminJ': 0.868655742000369, # 'M-s0J': 50.2304145481586, # 'M-v0J': 8.9133701258175, # 'M-vDJ': 7.25254749033457, # 'M-waitFactorJ': 0.129948701071488, # 'J-ac': 1.02741561224601, # 'J-amax': 3.45218265092789, # 'J-dc': -1.05378334461993, # 'J-dmax': -4.5, # 'J-DT': 1.44460006453038, # 'J-HWmin': 0.868655742000369, # 'J-s0': 50.2304145481586, # 'J-v0': 8.9133701258175, # 'J-vD': 7.25254749033457, # 'J-waitFactor': 0.129948701071488, # 'J-acM': 0.701349711460026, # 'J-amaxM': 3.22246147755112, # 'J-angryAlphaM': 0.317295453598947, # 'J-dcM': -0.973042307388401, # 'J-dmaxM': -4.5, # 'J-DTM': 0.637960764419036, # 'J-HWminM': 3.14598939235987, # 'J-s0M': 0, # 'J-v0M': 12.0312206826497 # } TESTVALUES = { # Spot check test 'M-ac': 1.79194144259679, 'M-amax': 2.9444235056204, 'M-angryAlpha': 0.194512427540837, 'M-attentive': False, 'M-cooperative': True, 'M-dc': -1.18228285662356, 'M-dmax': -4.5, 'M-DT': 0.50703018783129, 'M-HWmin': 0.554548326676528, 'M-s0': 0, 'M-v0': 15.900854330085, 'M-acJ': 1.79194144259679, 'M-amaxJ': 2.9444235056204, 'M-dcJ': -1.18228285662356, 'M-dmaxJ': -4.5, 'M-DTJ': 0.50703018783129, 'M-HWminJ': 0.554548326676528, 'M-s0J': 41.0959874365638, 'M-v0J': 4.35136051258971, 'M-vDJ': 3.62440857309546, 'M-waitFactorJ': 0.189427458783609, 'J-ac': 0.907520783339566, 'J-amax': 3.41062640977725, 'J-dc': -1.08693049004818, 'J-dmax': -4.5, 'J-DT': 1.46850832824293, 'J-HWmin': 3.4531065975579, 'J-s0': 41.0959874365638, 'J-v0': 4.35136051258971, 'J-vD': 4.97582418986585, 'J-waitFactor': 0.18310646262762, 'J-acM': 0.907520783339566, 'J-amaxM': 3.41062640977725, 'J-angryAlphaM': 0.153796374327613, 'J-dcM': -1.08693049004818, 'J-dmaxM': -4.5, 'J-DTM': 1.46850832824293, 'J-HWminM': 3.4531065975579, 'J-s0M': 0, 'J-v0M': 15.900854330085 } # TESTVALUES = { # Nonsensical comms result # 'M-ac': 0.532793136660887, # 'M-amax': 3.29267658218082, # 'M-angryAlpha': 0.17324107972835, # 'M-attentive': 0, # 'M-cooperative': 0, # 'M-dc': -0.556999920204043, # 'M-dmax': -4.5, # 'M-DT': 1.29173636694617, # 'M-HWmin': 2.99847641441427, # 'M-s0': 0, # 'M-v0': 12.7040735767202, # 'M-acJ': 1.4562140378201, # 'M-amaxJ': 3.05412409163499, # 'M-dcJ': -0.665820767283445, # 'M-dmaxJ': -4.5, # 'M-DTJ': 0.868689445157431, # 'M-HWminJ': 1.53025565537846, # 'M-s0J': 78.6786811757065, # 'M-v0J': 4.48492240086774, # 'M-vDJ': 5.00764225815986, # 'M-waitFactorJ': 0.175629523707857, # 'J-ac': 1.4562140378201, # 'J-amax': 3.05412409163499, # 'J-dc': -0.665820767283445, # 'J-dmax': -4.5, # 'J-DT': 0.868689445157431, # 'J-HWmin': 1.53025565537846, # 'J-s0': 78.6786811757065, # 'J-v0': 4.48492240086774, # 'J-vD': 5.00764225815986, # 'J-waitFactor': 0.175629523707857, # 'J-acM': 0.532793136660887, # 'J-amaxM': 3.29267658218082, # 'J-angryAlphaM': 0.17324107972835, # 'J-dcM': -0.556999920204043, # 'J-dmaxM': -4.5, # 'J-DTM': 1.29173636694617, # 'J-HWminM': 2.99847641441427, # 'J-s0M': 0, # 'J-v0M': 12.7040735767202 # } if __name__ == '__main__': TestBench()