This is a temporary section that contains the latest working simulation code covering the topic in the section. This code does run without know major issues. However, the detailed description in the preceding sections most likely will not cover all the functionality in the working code.
When the documentation complete this section will be removed. Each section end with a working example covering the features introduced to that point.
The following code snapshot is file
kese_gw_proto1_v7.py as of Feb 15, 2007
Features in Snapshoot. The v7 snapshot is feature complete for the Sun Up Chapter objectives. This includes the following:
Creation of a Geography. A hexagon Geo map of the simulation world is generated and stored in a data structure.
Character visualization of Geo Map. The Geo map is displayed with simple ascii characters. This includes a display showing Geo coordinates and a Geo map display showing the number of families in each Geo..
Modeling Arrival of New Families. Simulation let user set the number of new family units that arrive each day. The value in the code is 3 new families.
control pace of simulation. The real time pace of each simulation day is controlled. Default in 2 real time seconds per simulation day.
Geo capacity for support family units. Each Geo has a capacity to support a limited number if family units. This is modeled as a Process with limited capacity. The family units in a Geo compete for "a living." Those that do not find a living move on.
Family unit Behaviors. Two family behaviors are model. First, a family unit tries to make a living each day. If they succeed they stay in the "make living" behavior. If no living is available the family unit switches to the "Move" behavior and start the search for a Geo where they can make a living.
Figure 1. Code Snapshot as of Feb 15, 2008
# Knowledge and Economics Simulation Engine
# RealTime driver "Go West" scenario
from SimPy.SimulationRT import *
from random import *
import time
keseWorldWidth = 5
keseWorldHight = 8
simDays = 20
simHours = simDays * 24
averageGeoFamilyLivings = 3
externalFamiliesPerDay = 8
# Print feedback control
printPreBannerStuff = False
printNewGeos = False
printGeoLivings = False # This takes True to print all or (z,q) for a specific Geo or False for non
printGeoSunupEvent = False # Same as above
printDaylightCycle = False
printNewFamily = False
printFamilyMovein = False
printFamilyMovingout = False
printFamilySunupEvent = False
printFamilyGotLiving = False
printFamilyNoLining = False
printFamilyStartsMoving = False
printFamilyMovingTo = False
# -------------------------------------------------------
# World - the Master container for KESE
# -------------------------------------------------------
class World(Process):
def __init__(self, width, height, realTime):
Process.__init__(self, name='kese_world')
# store world map size
self.geosWide = width
self.geosHigh = height
# store world map range
self.geoXrange = (0, self.geosWide-1)
self.geoYrange = (0, self.geosHigh-1 )
self.rt = realTime
print '********* test', self.rt.timeClock(), 'ticks', self.rt.timeTicks()
# show ranges stored -- debugging
if printPreBannerStuff == True:
print 'X Wide', self.geosWide, \
'X range', self.geoXrange, \
', X min', self.geoXrange[0], \
', X max', self.geoXrange[1]
print 'Y High', self.geosHigh, \
'Y range', self.geoYrange, \
', Y min', self.geoYrange[0], \
', Y max', self.geoYrange[1]
# display sim to user
self.Banner()
# generate geo map
self.worldGeos = WorldGeosMap(world = self)
# store of next Family ID
self.nextFamilyId = 1
#create and activate associated Processes
externalitiesToWorld = Externality(self)
activate(externalitiesToWorld, externalitiesToWorld.keseCycle())
#####################################################################################
# main PEM for simulation - this controls the knowledge and economics cycle
#####################################################################################
def economicCycle(self):
while True:
# --- ecomonic cycle messages go here ---
yield waitevent, self, newDay
# fire externalities messge in hour 2 of day
yield hold, self, 1
externalityPhase.signal()
yield waitevent, self, sunUp # families will try to get living resource
yield hold, self, 1 # let all livings request queue with priorities
allocateLivings.signal() # message to renew livings
yield hold, self, 1
livingsAllocated.signal() # let families know living allocations for day are completed
yield waitevent, self, dayEnd
# display population distribution in world
self.popMap()
print 'end of day %d -- hour %d' %(now()/24, now()), '\n'
print 'Knowledge and Economics Simulation ended with run time of ', self.rt.timeTicks()
# ------------ World helper
def Banner(self):
print '\n\n\n'
print ' KESE \n'
print ' Go West Scenario'
print ' Knowledge and Economics Simulation Engine'
print ' OpenSimulations a division of SQI Inc.'
print '\n\n\n'
def worldGeosWide(self):
return self.geosWide
def worldGeosHigh(self):
return self.geosHigh
def getGeo(self, coordinates): # returns handle to geo object given x y geo center
return self.worldGeos.getGeo(coordinates)
def getGeoIds(self):
return self.worldGeos.getGeoIds() # returns a list of ids
def worldFamilyPopulation(self):
totalFamilies = 0
for statusGeoId in self.getGeoIds():
geoToStatus = self.getGeo(statusGeoId)
totalFamilies += geoToStatus.getGeoTotalFamies()
return totalFamilies
# -----------------------------------------------
# Helper methods for economic cycle
# -----------------------------------------------
def newFamily(self, geoToLiveIn):
# create family and place in geo
newFamily = Family(self.nextFamilyId, geoToLiveIn, self)
activate(newFamily, newFamily.familyActions())
geo = self.getGeo(geoToLiveIn)
geo.familyMoveIn(self.nextFamilyId, newFamily)
# increment for next family
self.nextFamilyId = self.nextFamilyId + 1
def familiesByGeo(self):
for statusGeoId in self.getGeoIds():
geoToStatus = self.getGeo(statusGeoId)
print 'Geo', statusGeoId, 'has families', geoToStatus.getFamilyIds()
def getFamiliesIn(self,geoCoordinates):
geoToStatus = self.getGeo(geoCoordinates)
return len(geoToStatus.getFamilyIds())
def popMap(self):
print '\n\nGeo Center Map'
for q in range(0, -(self.geosHigh), -1):
# print 'q is -------------------- ', q
mapRowTopOffset = []
mapRowBottomLeft = []
#for qRow in range(0, self.geosWide+1, 1):
for qRow in range(0, self.geosWide, 1):
# print 'qRow is ', qRow
zRow = qRow - self.geosHigh - q + 1
# print 'zRow is ', zRow
x = zRow
y = q + qRow + 1
geoCoord = (x, y)
# print geoCoord
mapRowTopOffset.append( self.getFamiliesIn((x,y)) )
x = zRow
y = q + qRow
geoCoord = (x, y)
# print geoCoord
mapRowBottomLeft.append( self.getFamiliesIn((x,y)) )
# print mapRowTopOffset
# print mapRowBottomLeft
display = ' '
for coord in mapRowTopOffset:
display += str(coord).center(12)
print display
display = ''
for coord in mapRowBottomLeft:
display += str(coord).center(12)
print display
def geoCoordMap(self):
print '\n\nGeo Center Map'
for q in range(0, -(self.geosHigh), -1):
# print 'q is -------------------- ', q
mapRowTopOffset = []
mapRowBottomLeft = []
#for qRow in range(0, self.geosWide+1, 1):
for qRow in range(0, self.geosWide, 1):
# print 'qRow is ', qRow
zRow = qRow - self.geosHigh - q + 1
# print 'zRow is ', zRow
x = zRow
y = q + qRow + 1
geoCoord = (x, y)
# print geoCoord
mapRowTopOffset.append(geoCoord)
x = zRow
y = q + qRow
geoCoord = (x, y)
# print geoCoord
mapRowBottomLeft.append(geoCoord)
# print mapRowTopOffset
display = ' '
for coord in mapRowTopOffset:
display += str(coord).center(12)
print display
# print mapRowBottomLeft
display = ''
for coord in mapRowBottomLeft:
display += str(coord).center(12)
print display
print '\n\n'
# --------------------------------------------------------
# GeoMapping for the World
# --------------------------------------------------------
class WorldGeosMap:
def __init__(self, world):
myWorld = world
geos = {}
# create geo set for map (in a dictionalry structure)
self.geos = {}
print 'world is', myWorld.worldGeosWide(), 'geos wide'
print 'world is', myWorld.worldGeosHigh(), 'geos high'
# create the Goes for requested map size
#for z in range(0, myWorld.worldGeosWide()+1, 1):
for z in range(0, myWorld.worldGeosWide(), 1):
for zCol in range(-myWorld.worldGeosHigh()+1, 1, 1):
qCol = -(myWorld.worldGeosHigh() - 1 + zCol)
# first column
x = zCol+z
y = z+qCol
if printNewGeos == True:
print 'I am a new Geo with coordinates are %d - %d' %(x, y)
self.geos[(x,y)] = Geo(geosX = x, geosY = y, world = self)
activate(self.geos[(x,y)], self.geos[(x,y)].day())
# second offset column
x = zCol+z
y = z+qCol+1
if printNewGeos == True:
print 'I am a new Geo with coordinates are %d - %d' %(x, y)
self.geos[(x,y)] = Geo(geosX = x, geosY = y, world = self)
activate(self.geos[(x,y)], self.geos[(x,y)].day())
# Summary date for world map
print '** Total number of geo locations greated:', len(self.geos)
# Show Geo co-ordinate map
myWorld.geoCoordMap()
def getGeo(self, coordinates): # returns handle to geo object given x y geo center
return self.geos[coordinates]
def getGeoIds(self):
return self.geos.keys()
# --------------------------------------------------------
# Geo - the core economic region
# --------------------------------------------------------
class Geo(Process):
def __init__(self, geosX, geosY, world):
Process.__init__(self, name='Geo')
self.myWorld = world
# create geo object at specific co-ordinates
self.geoLocation = (geosX, geosY)
# print 'I am a new Geo at (%d,%d)' %self.geoLocation
# create the resource level for family support and set default
self.cycleResourceUnits = averageGeoFamilyLivings
# livings is a general measure of how many families a geo can support
self.livings = Level(name = 'family_livings',
unitName = 'units',
initialBuffered = 0,
putQType = FIFO,
getQType = PriorityQ)
# Common Knowledge in Geo
self.bestNeighor = (0, 0) # commmon knowledge of where the best Geo to find living
# create the dictionary object to hold the faimly units
self.familiesInGeo = {}
def day(self):
# allocate livings
while True:
'''
In this simulation newDay is an economic event. At dawn each family awakes and
tries to secure a living for the day by making a request to the livings resource
for the geo.
'''
yield waitevent, self, newDay
self.updateCommonKnowledge()
yield waitevent, self, sunUp
if ((printGeoSunupEvent == True) or (printGeoSunupEvent == self.geoLocation)):
print 'Geo', self.geoLocation, 'got sunUp event at', now() # instructional
'''
The
'''
yield waitevent, self, allocateLivings
if ((printGeoLivings == True) or (printGeoLivings == self.geoLocation)):
print 'Geo', self.geoLocation, 'got allocateLivings event at', now(), \
'with %d livings from prior day' %self.livings.amount
'''
We can model the case where an unsettled geo is to as having more resourses that will
be available at the sustaining rate. This is done by starting the resource with
and initial buffer greater that the renewable rate. The test below does not update
the resource level until below sustaining levels.
'''
if self.livings.amount < self.cycleResourceUnits:
yield put, self, self.livings, (self.cycleResourceUnits - self.livings.amount)
yield hold, self, 1
# print 'Available Livings is Geo', self.geoLocation, 'at', now(), 'are %d' %self.livings.amount
def familyMoveIn(self, familyId, family):
self.familiesInGeo[familyId] = family
if printFamilyMovein == True:
print 'Family', family.getFamilyId(), 'just moved into Geo',self.geoLocation
return self.geoLocation
def familyMoveOut(self, familyId, family):
del self.familiesInGeo[familyId]
if printFamilyMovingout == True:
print 'Family', family.getFamilyId(), 'just moved out of geo', self.geoLocation
def getFamilyIds(self):
return self.familiesInGeo.keys()
def getGeoLocation(self):
return self.geoLocation
def getGeoTotalFamies(self):
return len(self.familiesInGeo.keys())
def geoState(self):
print 'first test'
# information on resource levels
def geoLivings(self):
return self.livings # provide livings resourse level object
def geoResourceLevel(self):
return self.livings.amount
def geoResourcesAvailable(self):
return self.livings.amount - getGeoTotalFamies()
def updateCommonKnowledge(self):
# find lowest popultion Geo - Has to be next to this one
bestGeo = (0,0)
lowestPop = 999999
for offSet in ( (1,-1),(0,-1),(-1,0),(-1,1),(1,0),(0,1) ):
# test to see that Geo is on World map
if (self.geoLocation[0]+offSet[0], self.geoLocation[1]+offSet[1]) in self.myWorld.getGeoIds():
# look aroung this Geo for a better opportunity
tryThisGeo = self.myWorld.getGeo( (self.geoLocation[0]+offSet[0], self.geoLocation[1]+offSet[1]) )
# print 'From', self.geoLocation, 'try this Geo is ',(self.geoLocation[0]+offSet[0], self.geoLocation[1]+offSet[1]), 'with families', tryThisGeo.getGeoTotalFamies()
if tryThisGeo.getGeoTotalFamies() < lowestPop:
bestGeo = (self.geoLocation[0]+offSet[0], self.geoLocation[1]+offSet[1])
# print 'families inside loop', tryThisGeo.getGeoTotalFamies()
lowestPop = tryThisGeo.getGeoTotalFamies()
# print 'From Geo', self.geoLocation, 'the best move is to', bestGeo
# lowest populated neighbor has been found so update "common knowledge"
self.bestNeighor = bestGeo
def bestNeighorForLiving(self):
return self.bestNeighor
# ---------------------------------------------------------------------------
# Externalities - Economic forces that are outside the behavior of simulation
# ---------------------------------------------------------------------------
class Externality(Process):
def __init__(self, world):
Process.__init__(self, name='externality')
self.geoWorld = world
def keseCycle(self): # the PEM for Externality
while True:
yield waitevent, self, (externalityPhase)
for x in range(0, externalFamiliesPerDay, 1):
# create a new family and place them in geo2.2
self.geoWorld.newFamily((0,0))
# -------------------------------------------------------
# Family -
# -------------------------------------------------------
class Family(Process):
maxPriority = 1000 # will be reduced by day appeared on map
def __init__(self, familyId, liveIn, world):
Process.__init__(self, name='Family')
# create a new family and place them a in geo
self.familyId = familyId
self.myWorld = world
self.liveIn = liveIn # coordinates of Geo that we live in
self.geoLivingIn = world.getGeo(self.liveIn) # Geo object ref - for peformance
self.appearDay = int(now()/24)
self.geoMoveInDay = int(now()/24)
self.behaviorCase = 'makeLiving'
if printNewFamily == True:
print 'We are the new family unit', self.familyId, 'living in geo', self.liveIn, \
'created at', now()
# print 'Geo living in', self.geoLivingIn -------------
def familyActions(self): # PEM for Family
while True:
yield waitevent, self, sunUp
'''
The bahavior of the Family unit is determined by the following cases:
makeLiving - the family is in a Geo and trying to settle and make a living. If they can't
find the resources they switch to a travel behavior.
'''
if printGeoSunupEvent == True:
print 'sunUp for Family %d with priority %d' \
%(self.familyId, self.maxPriority - self.appearDay)
# ------------ make living behavior ------------------------------------------
if self.behaviorCase == 'makeLiving':
# try to make a living
geoLivingIn = self.myWorld.getGeo(self.liveIn)
livingsResourseForGeo = geoLivingIn.geoLivings()
yield (get, self, livingsResourseForGeo, 1, (self.maxPriority - self.appearDay)),\
(waitevent, self, livingsAllocated)
# print 'Family', self.familyId, 'released from livings resource yield at', now()
if self.acquired(livingsResourseForGeo):
if printFamilyGotLiving == True:
print 'Family', self.familyId, 'in Geo', self.liveIn, \
'at game hour', now(), \
'got living for day', int(now()/24)
else:
if printFamilyNoLining == True:
print 'Family', self.familyId, 'in Geo', self.liveIn, \
'at game hour', now(), \
'no living available for day', int(now()/24)
print ' -- We must now move the family'
# place family in travel behavior
self.behaviorCase = 'travelToNewGeo'
# --------------- travel behavior -------------------------------------------
elif self.behaviorCase == 'travelToNewGeo':
# select Geo to move to using the Common Knowledge in this Geo
geoLivingIn = self.myWorld.getGeo(self.liveIn)
targetGeoCoord = geoLivingIn.bestNeighorForLiving()
# get Geo object of move target
targetGeo = self.myWorld.getGeo(targetGeoCoord)
# print 'The moving target Geo is', targetGeoCoord
if printFamilyMovingTo == True:
print 'Family %d living in' %self.familyId, self.liveIn,\
' --> moving to', targetGeo.getGeoLocation()
# wait for sunUp of next day
yield waitevent, self, sunUp
if printFamilyStartsMoving == True:
print 'Family', self.familyId, 'starts moving to new Geo at', now()
# remove fom Geo
self.geoLivingIn.familyMoveOut(self.familyId, self)
# wait for sunDown
yield waitevent, self, sunDown
# add to new Geo
geoMovingTo = self.myWorld.getGeo(targetGeoCoord)
# add family and have family update where they live
self.liveIn = geoMovingTo.familyMoveIn(self.familyId, self)
# now update Geo object reference
self.geoLivingIn = self.myWorld.getGeo(self.liveIn)
# change family behavior to looking for a living
self.behaviorCase = 'makeLiving'
else:
print 'WE HAVE BIG PROBLEM. FAMILY BEHAVIOR STATE DOES NOT MATCH ANY KNOWN BEHAVIORS'
def getFamilyId(self):
return self.familyId
##################################################################################
# daylight massaging Process
##################################################################################
'''
This is a simple 12 hour cycle. However the procuss could be extended to
simulate seasons and/or latitude.
Note: this process also fires a new day event
'''
class Daylight(Process):
def __init__(self):
Process.__init__(self, name='Daylight')
self.sunUpHour = 6
self.sunDownHour = 18
def sunEvents(self):
# start 24 hour loop
while True:
yield hold, self, 1 # tick internal clock to first hour of next day
# send start of new day event
newDay.signal()
if printDaylightCycle == True:
print 'From Daylight Process: Start of day at %d' %now()
# wait for daylight and them sed event
yield hold, self, self.sunUpHour
sunUp.signal()
if printDaylightCycle == True:
print 'From Daylight Process: sun up at %d' %now()
# wait for nightfall and then send event
yield hold, self, (self.sunDownHour-self.sunUpHour)-1 # sub 1 to set at start of sundown
sunDown.signal()
if printDaylightCycle == True:
print 'From Daylight Process: sundown at %d' %now()
# wait for midnight
yield hold, self, 24-self.sunDownHour
dayEnd.signal()
if printDaylightCycle == True:
print 'From Daylight Process: Day ended at %d' %now()
####################################################################################
# Real Time helper class
####################################################################################
class RealTime():
def __init__(self):
self.launchTime = time.clock()
# setup stuff for debugging
if printPreBannerStuff == True:
print 'simulation environment launched at %.1f' %self.launchTime
def timeTicks(self):
return time.clock() - self.launchTime
def timeClock(self):
return self.launchTime
def reset(self):
self.launchTime = time.clock()
return self.launchTime
##########################################################################
#
# Module level code that initializes simulation
#
##########################################################################
rt = RealTime()
initialize()
# create World object which prints banner
kese_world = World(width = keseWorldWidth, height = keseWorldHight, realTime = rt)
activate(kese_world, kese_world.economicCycle())
# create day and daylight event Process
daylightEvents = Daylight()
activate(daylightEvents, daylightEvents.sunEvents())
# Create Event Objects
externalityPhase = SimEvent(name='Externality Phase signal for Economic Cycle')
sunUp = SimEvent(name='Sun rise event')
sunDown = SimEvent(name='Sun set event')
newDay = SimEvent(name='Start of day event')
dayEnd = SimEvent(name='End of day event')
allocateLivings = SimEvent(name='Renew livings resource')
livingsAllocated = SimEvent(name='Livings allocations completed for day')
# Reset start time for events that trigger running sim
print 'Simulation setup time was %.1f' %rt.timeTicks()
print 'Simulaiton run started at %.1f' %rt.reset()
# simualtion starts here
simulate(real_time=True, rel_speed=12, until=simHours) ##unit sim time = 1 sec clock
print 'Knowledge and Economics Simulation ended', \
'with run time of %d.1 second per day' %(rt.timeTicks()/simDays)