Reads control plan data from a file and creates the control plans in AIMSUN by Peter Hidas.

Disclaimer of Warranty: TO THE EXTENT PERMITTED BY APPLICABLE LAW THE AIMSUN SCRIPTS, INCLUDING BUT NOT LIMITED TO ALL DATA, TOOLS AND CALCULATIONS THEREIN ARE PROVIDED "AS IS" AND WITHOUT EXPRESS OR IMPLIED WARRANTY OF ANY KIND BY EITHER TSS OR ANYONE ELSE WHO HAS BEEN INVOLVED IN THE CREATION, PRODUCTION OR DELIVERY OF THE AIMSUN SCRIPTS, INCLUDING BUT NOT LIMITED TO ANY EXPRESS OR IMPLIED WARRANTY OF MERCHANTABILITY, ACCURACY, QUIET ENJOYMENT, NONINFRINGEMENT OR FITNESS FOR A PARTICULAR PURPOSE. NO COVENANTS, WARRANTIES OR INDEMNITIES OF ANY KIND ARE GRANTED BY TSS TO YOU THE USER. SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING, REPAIR OR CORRECTION.

# A script that reads control plan data from a file # and creates the control plans # Written by Peter Hidas 26-May-2006 # based on a script from # (c) 2005 TSS-Transport Simulation Systems # Modified 15-Jun-2006 # Modified PH 20-Oct-2006: added offset to cSignal and input # Modified PH 6-Dec-2006: adding Actuated control parameters # Modified PH 11-Dec-2006: reading Control Plan and Node names as words # =============================================================== # Control Plan input data structure # first char: line code, one of: c, n, t, p, i, a # c: create new Control Plan, followed by name of control plan (may be several words!) # n: add Node to control plan, followed by Node name (TCS) (may be several words!) # t: timing: cycle time, yellow (amber) time, offset time[Optional], Actuated[Optional] # p: phase: green time, signal group1, signal group 2, ... # a: Actuated phase: Recall (see note 6), MinGreen, SecondsActuation, PassageTime, Hold (see Note 7), Det1, Det2 (Note 8)... # i: interphase: duration, signal group1, signal group2, ... # # Notes # 1. data items must be separated spaces # 2. The first character of each line is used as a code # 3. Only 6 codes are 'active': (c, n, t, p, i, a), lines starting with any other # character are ignored # 5. The script makes use of some general functions: # - findByName # - findByExternalId # which are implemented in the "initial.py" script - this script is always # executed by Aimsun before any other scripts # # 6. Recall can be one of: No, Min, Max, Coord # 7. Hold is boolean: 1 (True) or 0 (False) # 8. Detectors must be identified by their Name # ============================================ # Sample input data (remove the '# ' from the beginning of each line!): # c Control Plan 07:00 # n TCS271 # # York St & Market St # t 99 2 72 Actuated # p 19 1 5 # a Coord 19 2 2 1 TCS271-d12 TCS271-d3 # i 6 # p 58 2 4 6 # a Min 36 2 2 0 TCS271-d56 TCS271-d4 TCS271-d7 # i 6 4 # p 5 3 4 # a No 5 2 2 0 TCS271-d8 # i 5 # n TCS281 # # York St & King St # t 99 2 -1 # p 54 2 # i 6 # p 33 1 # i 6 # n TCS282 # # ... etc ... # # # # # # # # # # # # # # # # # # # # # # c Control Plan 07:15 # n TCS271 # # York St & Market St # # ... etc ... #============================================================= # global variables phaseStart = 0 # phase Start time # Reset the path & filename for the proper file in each case!!!!!!!! path = 'D:/My_Documents/York_St/Calibrated/' dataName = 'york_20061211_analysis.prn' fName = path + dataName logName = path + dataName[:dataName.index('.')] + '.log' logFile = open(logName, 'w') recallType = ['No', 'Min', 'Max', 'Coord'] numbers = '0123456789-' # A Phase class to hold signal phase data class cPhase: "Phase record: duration, isInterPhase, list of sigGroups" time = 0 isInterPhase = False sigGroups = [] def __init__(self, words, iValue): self.sigGroups = [] self.time = int(words[1]) # green time duration self.isInterPhase = iValue # True: interphase, False: normal phase sGroups = words[2:] for i in range(len(sGroups)): # signal groups assigned to phase self.sigGroups.append(int(sGroups[i])) def text(self): s = "phase time= %i interPhase: %s groups:" % (self.time, self.isInterPhase) for i in range(len(self.sigGroups)): s = s + ' ' + str(self.sigGroups[i]) s = s + '\n' return s # A Signal class to hold a signal control data class cSignal: "Signal record: name, Id, extId, cycleTime, yellowTime, list of phases" name = "Unknown" Id = 0 extId = "Unknown" cycleTime = 0 yellowTime = 0 offset = 0 phases = [] def text(self): s = "%s %i %s, cTime: %i yTime: %i \n" % (self.name, self.Id, self.extId, self.cycleTime, self.yellowTime) for i in range(len(self.phases)): s = s + ' ' + self.phases[i].text() s = s + '\n' return s #A ControlPlan class to hold all data of a signal control plan class cPlan: "Signal control Plan: name, list of Signals" name = "Unknown" signals = [] def text(self): s = "Plan %s\n" % (self.name) for i in range(len(self.signals)): s = s + self.signals[i].text() s = s + '\n' return s # Returns (and creates if needed) the folder for the control plan def getControlPlanFolder( model ): externalFolderName = "Control Plans" folderName = "GKModel::controlPlans" type = GKSystem.getSystem().getType( "GKControlPlan" ) folder = model.getCreateRootFolder().findFolder( folderName ) if folder == None: folder = model.getCreateRootFolder().createFolder( externalFolderName, folderName, type) # print "controlPlanFolder returned" return folder # Creates a control plan with the given name def createControlPlan( model, name ): global logFile folder = getControlPlanFolder( model ) plan = findByName ( model, name, "GKControlPlan" ) # plan = findControlPlan ( model, name ) if plan != None: print "Control Plan %s exists! It will be removed!" % (name) logFile.write("Control Plan %s exists! It will be removed!\n" % (name)) planDelCmd = plan.getDelCmd() model.getCommander().addCommand( planDelCmd ) # folder.remove( plan ) # deleteObject( plan ) plan = GKSystem.getSystem().newObject( "GKControlPlan", model ) plan.setName( name ) folder.append( plan ) print "Control Plan (%s) created" % (name) logFile.write("Control Plan (%s) created\n" % (name)) return plan def createSigGroups( node ): sigGroups = [] for sigGroup in node.getSignals(): sigGroups.append( sigGroup ) return sigGroups def createControlJunction( cp, node, cycleTime, yellowTime, offset, isActivated): global logFile cpNode = cp.createControlJunction( node.getId() ) cpNode.setCycle( cycleTime ) cpNode.setYellowTime( yellowTime ) cpNode.setOffset( offset ) if isActivated: cpNode.setControlJunctionType( GKControlJunction.eActuated ) else: cpNode.setControlJunctionType( GKControlJunction.eFixedControl ) print "%s control created Cycle: %i Yellow: %i Offset: %i" % ( node.getName(), cycleTime, yellowTime, offset ) logFile.write("%s control created Cycle: %i Yellow: %i Offset: %i\n" % ( node.getName(), cycleTime, yellowTime, offset ) ) return cpNode def addPhase( cpNode, fromTime, duration, isInterphase): global phaseStart, logFile phase = cpNode.createPhase() phase.setFrom( fromTime ) phase.setDuration( duration ) phase.setInterphase( isInterphase ) # print "Phase created start: %i length: %i" % (fromTime, duration) logFile.write("Phase created start: %i length: %i\n" % (fromTime, duration) ) phaseStart = phaseStart + duration return phase def setActuatedParams( phase, aPhase): rct = GKControlPhase.GKControlPhaseRecall( recallType.index( aPhase.recall ) ) phase.setRecall( rct ) phase.setMaxDuration( aPhase.time ) phase.setMinDuration( aPhase.minGreen ) phase.setMaximumInitial( aPhase.minGreen ) phase.setSecondsActuation( aPhase.secAct ) phase.setPassageTime( aPhase.passTime ) phase.setUsingGapReduction( False ) phase.setHold( bool(aPhase.hold) ) for i in range(len(aPhase.detectors)): dName = aPhase.detectors[i] det = findByName( model, dName, "GKDetector" ) if det == None: print "Detector %s not found!" % (dName) logFile.write("Detector %s not found!\n" % (dName) ) else: cdet = phase.createControlDetector( det ) phase.addControlDetector( cdet ) print "Detector %s added!" % (dName) logFile.write("Detector %s added!\n" % (dName) ) def createPlan(aPlan): global phaseStart, logFile # Create a new control plan controlPlan = createControlPlan( model, aPlan.name ) # create a JunctionPlan for each node in the plan for aNode in aPlan.signals: print "Searching for Node %s ..." % (aNode.name) logFile.write("Searching for Node %s ...\n" % (aNode.name) ) node = findByName( model, aNode.name, "GKNode" ) # node = findNode( model, aNode.Id) if node == None: print "Node %s not found!" % ( aNode.name ) logFile.write("Node %s not found!\n" % (aNode.name) ) else: # Create control plan for node cycleTime = aNode.cycleTime cpNode = createControlJunction( controlPlan, node, cycleTime, aNode.yellowTime, aNode.offset, aNode.isActuated ) # Create the Signal Groups sigGroups = createSigGroups( node ) print "signal groups appended:" logFile.write("signal groups appended:\n") for a in sigGroups: # print "group %i Id: %i" % (sigGroups.index(a)+1, a.getId() ) logFile.write(" group %i Id: %i\n" % (sigGroups.index(a)+1, a.getId() )) # create phases phaseStart = 0 for aPhase in aNode.phases: # print "adding phase:" # print aPhase.text() # add phase to cpNode from phaseStart, duration, intergreen phase = addPhase( cpNode, phaseStart, aPhase.time, aPhase.isInterPhase ) # print phases for i in range(len(aPhase.sigGroups)): g = aPhase.sigGroups[i]-1 if g in range(len(sigGroups)): phase.addSignal( sigGroups[ g ].getId() ) # print "added group %i Id: %i" % (aPhase.sigGroups[i], sigGroups[g].getId() ) logFile.write("added group %i Id: %i\n" % (aPhase.sigGroups[i], sigGroups[g].getId() )) else: print "Group %i does not exist at %s!" % (g, aNode.name ) logFile.write("Group %i does not exist at %s!\n" % (g, aNode.name ) ) if aNode.isActuated and not aPhase.isInterPhase: setActuatedParams( phase, aPhase) # if the Node is Actuated, calculate ActuatedForceOffs if aNode.isActuated: cpNode.calcActuatedForceOff () # check if phaseTimes add up to cycleTime if phaseStart != cycleTime: print "Total phase time (%i) different from Cycle time (%i)!" % (phaseStart, cycleTime) logFile.write("Total phase time (%i) different from Cycle time (%i)!\n" % (phaseStart, cycleTime)) # ========= Entry code, the script starts here ======================== # Read the control plan data from file try: aPlan = cPlan() # aPlan stores all data for one Control Plan f = open(fName, 'r') for line in f: # print line # logFile.write(line) line = line.strip() # remove leading/trailing spaces and EOL char words = line.split() # split(line) if line[0] == 'c': # --- Control Plan --- if aPlan.name != "Unknown": #print aPlan.text() createPlan(aPlan) aPlan = cPlan() # create new plan aPlan.name = words[1] for i in range(2, len(words)): aPlan.name = aPlan.name + ' ' + words[i] aPlan.signals = [] elif line[0] == 'n': # --- Node ----------- aNode = cSignal() aNode.phases = [] aNode.name = words[1] for i in range(2, len(words)): aNode.name = aNode.name + ' ' + words[i] #aNode.Id = int(words[2]) #aNode.extId = words[3] aPlan.signals.append(aNode) elif line[0] == 't': # ----- Time ---------- aNode.cycleTime = int(words[1]) aNode.yellowTime = int(words[2]) if len(words) > 3 and words[3][0] in numbers: # optional: offset aNode.offset = int(words[3]) if words[ len(words)-1 ][0] in ('aA'): # optional: Activated aNode.isActuated = True else: aNode.isActuated = False elif line[0] == 'p': # ----- Phase -------- phase = cPhase(words, 0 ) # False: NOT interphase aNode.phases.append(phase) elif line[0] == 'a': # ----- Actuated Phase data ----- phase.recall = words[1] phase.minGreen = int(words[2]) phase.secAct = int(words[3]) phase.passTime = int(words[4]) phase.hold = int(words[5]) # 0: False, 1: True phase.detectors = [] for i in range(6, len(words)): phase.detectors.append(words[i]) elif line[0] == 'i': # ------ Interphase ------------ phase = cPhase(words, 1 ) # True: interphase aNode.phases.append(phase) else: pass # now create the last plan from the file: if aPlan.name != "Unknown": createPlan(aPlan) print "Done!!" finally: f.close() logFile.close() # Be sure that we reset the UNDO buffer after a non undoable modification model.getCommander().addCommand( None )