Source code for pypdevs.tracer
# Copyright 2014 Modelling, Simulation and Design Lab (MSDL) at
# McGill University and the University of Antwerp (http://msdl.cs.mcgill.ca/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
[docs]class Tracers(object):
"""
Interface for all tracers
"""
[docs] def __init__(self):
"""
Constructor
"""
self.tracers = []
self.tracers_init = []
self.uid = 0
[docs] def registerTracer(self, tracer, server, recover):
"""
Register a tracer, so that it will also receive all transitions.
:param tracer: tuple of the form (file, classname, [args])
:param server: the server object to be able to make remote calls
:param recover: whether or not this is a recovered registration (used during checkpointing)
"""
loc = {}
try:
exec("from pypdevs.tracers.%s import %s" % (tracer[0], tracer[1]), {}, loc)
except:
exec("from %s import %s" % (tracer[0], tracer[1]), {}, loc)
self.tracers.append(loc[tracer[1]](self.uid, server, *tracer[2]))
# self.tracers.append(eval("%s(%i, server, *%s)" % (tracer[1],
# self.uid,
# tracer[2])))
self.tracers_init.append(tracer)
self.uid += 1
self.tracers[-1].startTracer(recover)
[docs] def hasTracers(self):
"""
Checks whether or not there are any registered tracers
:returns: bool
"""
return len(self.tracers) > 0
[docs] def getByID(self, uid):
"""
Gets a tracer by its UID
:param uid: the UID of the tracer to return
:returns: tracer
"""
return self.tracers[uid]
[docs] def stopTracers(self):
"""
Stop all registered tracers
"""
for tracer in self.tracers:
tracer.stopTracer()
[docs] def tracesUser(self, time, aDEVS, variable, value):
"""
Perform all tracing actions for a user imposed modification. This is NOT supported by default DEVS, so we don't require tracers to handle this either.
:param time: the time at which the modification happend; this will be the termination time of the previous simulation run and **not** the time at which the timeAdvance was recomputed!
:param aDEVS: the atomic DEVS model that was altered
:param variable: the variable that was altered (as a string)
:param value: the new value of the variable
"""
for tracer in self.tracers:
try:
tracer.traceUser(time, aDEVS, variable, value)
except AttributeError:
# Some tracers choose to ignore this event
pass
[docs] def tracesInit(self, aDEVS, t):
"""
Perform all tracing actions for an initialisation
:param aDEVS: the model that was initialised
:param t: the time at which the initialisation should be logged
"""
if aDEVS.full_name is None:
return
for tracer in self.tracers:
tracer.traceInit(aDEVS, t)
[docs] def tracesInternal(self, aDEVS):
"""
Perform all tracing actions for an internal transition
:param aDEVS: the model that transitioned
"""
if aDEVS.full_name is None:
return
for tracer in self.tracers:
tracer.traceInternal(aDEVS)
[docs] def tracesExternal(self, aDEVS):
"""
Perform all tracing actions for an external transition
:param aDEVS: the model that transitioned
"""
if aDEVS.full_name is None:
return
for tracer in self.tracers:
tracer.traceExternal(aDEVS)
[docs] def tracesConfluent(self, aDEVS):
"""
Perform all tracing actions for a confluent transition
:param aDEVS: the model that transitioned
"""
if aDEVS.full_name is None:
return
for tracer in self.tracers:
tracer.traceConfluent(aDEVS)