# ISC License
#
# Copyright (c) 2023, Autonomous Vehicle Systems Lab, University of Colorado at Boulder
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
import array
import inspect
# Import some architectural stuff that we will probably always use
import os
import sys
import warnings
import xml.etree.ElementTree as ET
from collections import OrderedDict
import matplotlib.pyplot as plt
import numpy as np
from Basilisk.architecture import alg_contain, bskLogging, sim_model
from Basilisk.utilities import deprecated, simulationArchTypes
from Basilisk.utilities.pythonVariableLogger import PythonVariableLogger
from Basilisk.utilities.simulationProgessBar import SimulationProgressBar
# Point the path to the module storage area
# define ASCI color codes
processColor = '\u001b[32m'
taskColor = '\u001b[33m'
moduleColor = '\u001b[36m'
endColor = '\u001b[0m'
[docs]
class EventHandlerClass:
"""Event Handler Class"""
def __init__(self, eventName, eventRate=int(1E9), eventActive=False,
conditionList=[], actionList=[], terminal=False):
self.eventName = eventName
self.eventActive = eventActive
self.eventRate = eventRate
self.conditionList = conditionList
self.actionList = actionList
self.occurCounter = 0
self.prevTime = -1
self.checkCall = None
self.operateCall = None
self.terminal = terminal
def methodizeEvent(self):
if self.checkCall != None:
return
funcString = 'def EVENT_check_' + self.eventName + '(self):\n'
funcString += ' if('
for condValue in self.conditionList:
funcString += ' ' + condValue + ' and'
funcString = funcString[:-3] + '):\n'
funcString += ' return 1\n'
funcString += ' return 0'
exec (funcString)
self.checkCall = eval('EVENT_check_' + self.eventName)
funcString = 'def EVENT_operate_' + self.eventName + '(self):\n'
for actionValue in self.actionList:
funcString += ' '
funcString += actionValue + '\n'
funcString += ' return 0'
exec (funcString)
self.operateCall = eval('EVENT_operate_' + self.eventName)
def checkEvent(self, parentSim):
nextTime = int(-1)
if self.eventActive == False:
return(nextTime)
nextTime = self.prevTime + self.eventRate - (self.prevTime%self.eventRate)
if self.prevTime < 0 or (parentSim.TotalSim.CurrentNanos%self.eventRate == 0):
nextTime = parentSim.TotalSim.CurrentNanos + self.eventRate
eventCount = self.checkCall(parentSim)
self.prevTime = parentSim.TotalSim.CurrentNanos
if eventCount > 0:
self.eventActive = False
self.operateCall(parentSim)
self.occurCounter += 1
if self.terminal:
parentSim.terminate = True
return(nextTime)
[docs]
class StructDocData:
"""Structure data documentation class"""
class StructElementDef:
def __init__(self, type, name, argstring, desc=''):
self.type = type
self.name = name
self.argstring = argstring
self.desc = desc
def __init__(self, strName):
self.strName = strName
self.structPopulated = False
self.structElements = {}
def clearItem(self):
self.structPopulated = False
self.structElements = {}
def populateElem(self, xmlSearchPath):
if self.structPopulated == True:
return
xmlFileUse = xmlSearchPath + '/' + self.strName + '.xml'
try:
xmlData = ET.parse(xmlFileUse)
except:
print("Failed to parse the XML structure for: " + self.strName)
print("This file does not exist most likely: " + xmlFileUse)
return
root = xmlData.getroot()
validElement = root.find("./compounddef[@id='" + self.strName + "']")
for newVariable in validElement.findall(".//memberdef[@kind='variable']"):
typeUse = newVariable.find('type').text if newVariable.find('type') is not None else \
None
nameUse = newVariable.find('name').text if newVariable.find('type') is not None else \
None
argstringUse = newVariable.find('argsstring').text if newVariable.find('argsstring') is not None else \
None
descUse = newVariable.find('./detaileddescription/para').text if newVariable.find(
'./detaileddescription/para') is not None else \
None
if descUse == None:
descUse = newVariable.find('./briefdescription/para').text if newVariable.find(
'./briefdescription/para') is not None else \
None
newElement = StructDocData.StructElementDef(typeUse, nameUse, argstringUse, descUse)
self.structElements.update({nameUse: newElement})
self.structPopulated = True
def printElem(self):
print(" " + self.strName + " Structure Elements:")
for key, value in self.structElements.items():
outputString = ''
outputString += value.type + " " + value.name
outputString += value.argstring if value.argstring is not None else ''
outputString += ': ' + value.desc if value.desc is not None else ''
print(" " + outputString)
class DataPairClass:
def __init__(self):
self.outputMessages = set([])
self.inputMessages = set([])
self.name = ""
self.outputDict = {}
[docs]
class SimBaseClass:
"""Simulation Base Class"""
def __init__(self):
self.TotalSim = sim_model.SimModel()
self.TaskList = []
self.procList = []
self.StopTime = 0
self.nextEventTime = 0
self.terminate = False
self.eventMap = {}
self.simBasePath = os.path.dirname(os.path.realpath(__file__)) + '/../'
self.dataStructIndex = self.simBasePath + '/xml/index.xml'
self.indexParsed = False
self.simulationInitialized = False
self.simulationFinished = False
self.bskLogger = bskLogging.BSKLogger()
self.showProgressBar = False
self.allModules = set()
[docs]
def SetProgressBar(self, value):
"""
Shows a dynamic progress in the terminal while the simulation is executing.
"""
self.showProgressBar = value
[docs]
def ShowExecutionOrder(self):
"""
Shows in what order the Basilisk processes, task lists and modules are executed
"""
for processData in self. TotalSim.processList:
print(f"{processColor}Process Name: {endColor}" + processData.processName +
" , " + processColor + "priority: " + endColor + str(processData.processPriority))
for task in processData.processTasks:
print(f"{taskColor}Task Name: {endColor}" + task.TaskPtr.TaskName +
", " + taskColor + "priority: " + endColor + str(task.taskPriority) +
", " + taskColor + "TaskPeriod: " + endColor + str(task.TaskPtr.TaskPeriod/1.0e9) + "s")
for module in task.TaskPtr.TaskModels:
print(moduleColor + "ModuleTag: " + endColor + module.ModelPtr.ModelTag +
", " + moduleColor + "priority: " + endColor + str(module.CurrentModelPriority))
print("")
[docs]
def AddModelToTask(self, TaskName, NewModel, ModelData=None, ModelPriority=-1):
"""
This function is responsible for passing on the logger to a module instance (model), adding the
model to a particular task, and defining
the order/priority that the model gets updated within the task.
:param TaskName (str): Name of the task
:param NewModel (obj): Model to add to the task
:param ModelData: None or struct containing, only used for C BSK modules
:param ModelPriority (int): Priority that determines when the model gets updated. (Higher number = Higher priority)
:return:
"""
# Supports calling AddModelToTask(TaskName, NewModel, ModelPriority)
if isinstance(ModelData, int):
ModelPriority = ModelData
ModelData = None
for Task in self.TaskList:
if Task.Name == TaskName:
Task.TaskData.AddNewObject(NewModel, ModelPriority)
if ModelData is not None:
try:
ModelData.bskLogger = self.bskLogger
except:
pass
Task.TaskModels.append(ModelData)
else:
try:
NewModel.bskLogger = self.bskLogger
except:
pass
Task.TaskModels.append(NewModel)
return
raise ValueError(f"Could not find a Task with name: {TaskName}")
[docs]
def CreateNewProcess(self, procName, priority = -1):
"""
Creates a process and adds it to the sim
:param procName (str): Name of process
:param priority (int): Priority that determines when the model gets updated. (Higher number = Higher priority)
:return: simulationArchTypes.ProcessBaseClass object
"""
proc = simulationArchTypes.ProcessBaseClass(procName, priority)
self.procList.append(proc)
self.TotalSim.addNewProcess(proc.processData)
return proc
[docs]
def CreateNewTask(self, TaskName, TaskRate, InputDelay=None, FirstStart=0):
"""
Creates a simulation task on the C-level with a specific update-frequency (TaskRate), an optional delay, and
an optional start time.
Args:
TaskName (str): Name of Task
TaskRate (int): Number of nanoseconds to elapse before update() is called
InputDelay (int): (depreciated, unimplemented) Number of nanoseconds simulating a lag of the particular task
FirstStart (int): Number of nanoseconds to elapse before task is officially enabled
Returns:
simulationArchTypes.TaskBaseClass object
"""
if InputDelay is not self.CreateNewTask.__defaults__[0]:
deprecated.deprecationWarn("InputDelay", "2024/12/13",
"This input variable is non-functional and now depreciated.")
Task = simulationArchTypes.TaskBaseClass(TaskName, TaskRate, FirstStart)
self.TaskList.append(Task)
return Task
def ResetTask(self, taskName):
for Task in self.TaskList:
if Task.Name == taskName:
Task.resetTask(self.TotalSim.CurrentNanos)
[docs]
def InitializeSimulation(self):
"""
Initialize the BSK simulation. This runs the SelfInit() and Reset() methods on each module.
"""
if(self.simulationInitialized):
self.TotalSim.resetThreads(self.TotalSim.getThreadCount())
self.TotalSim.assignRemainingProcs()
self.TotalSim.ResetSimulation()
self.TotalSim.selfInitSimulation()
self.TotalSim.resetInitSimulation()
self.simulationInitialized = True
[docs]
def ExecuteSimulation(self):
"""
run the simulation until the prescribed stop time or termination.
"""
self.initializeEventChecks()
nextStopTime = self.TotalSim.NextTaskTime
nextPriority = -1
progressBar = SimulationProgressBar(self.StopTime, self.showProgressBar)
while self.TotalSim.NextTaskTime <= self.StopTime and not self.terminate:
if self.TotalSim.CurrentNanos >= self.nextEventTime >= 0:
self.nextEventTime = self.checkEvents()
self.nextEventTime = self.nextEventTime if self.nextEventTime >= self.TotalSim.NextTaskTime else self.TotalSim.NextTaskTime
if 0 <= self.nextEventTime < nextStopTime:
nextStopTime = self.nextEventTime
nextPriority = -1
if self.terminate:
break
self.TotalSim.StepUntilStop(nextStopTime, nextPriority)
progressBar.update(self.TotalSim.NextTaskTime)
nextPriority = -1
nextStopTime = self.StopTime
nextStopTime = nextStopTime if nextStopTime >= self.TotalSim.NextTaskTime else self.TotalSim.NextTaskTime
self.terminate = False
progressBar.markComplete()
progressBar.close()
[docs]
def disableTask(self, TaskName):
"""
Disable this particular task from being executed.
"""
for Task in self.TaskList:
if Task.Name == TaskName:
Task.disable()
[docs]
def enableTask(self, TaskName):
"""
Enable this particular task to be executed.
"""
for Task in self.TaskList:
if Task.Name == TaskName:
Task.enable()
def parseDataIndex(self):
self.dataStructureDictionary = {}
try:
xmlData = ET.parse(self.dataStructIndex)
except:
print("Failed to parse the XML index. Likely that it isn't present")
return
root = xmlData.getroot()
for child in root:
newStruct = StructDocData(child.attrib['refid'])
self.dataStructureDictionary.update({child.find('name').text:
newStruct})
self.indexParsed = True
[docs]
def createNewEvent(self, eventName, eventRate=int(1E9), eventActive=False,
conditionList=[], actionList=[], terminal=False):
"""
Create an event sequence that contains a series of tasks to be executed.
"""
if (eventName in list(self.eventMap.keys())):
return
newEvent = EventHandlerClass(eventName, eventRate, eventActive,
conditionList, actionList, terminal)
self.eventMap.update({eventName: newEvent})
def initializeEventChecks(self):
self.eventList = []
for key, value in self.eventMap.items():
value.methodizeEvent()
self.eventList.append(value)
self.nextEventTime = 0
def checkEvents(self):
nextTime = -1
for localEvent in self.eventList:
localNextTime = localEvent.checkEvent(self)
if(localNextTime >= 0 and (localNextTime < nextTime or nextTime <0)):
nextTime = localNextTime
return nextTime
def setEventActivity(self, eventName, activityCommand):
if eventName not in list(self.eventMap.keys()):
print("You asked me to set the status of an event that I don't have.")
return
self.eventMap[eventName].eventActive = activityCommand
[docs]
def setAllButCurrentEventActivity(self, currentEventName, activityCommand, useIndex=False):
"""Set all event activity variables except for the currentEventName event. The ``useIndex`` flag can be used to
prevent enabling or disabling every task, and instead only alter the ones that belong to the same group (for
example, the same spacecraft). The distinction is made through an index set after the ``_`` symbol in the event
name. All events of the same group must have the same index."""
if useIndex:
index = currentEventName.partition('_')[2] # save the current event's index
for eventName in list(self.eventMap.keys()):
if currentEventName != eventName:
if useIndex:
if eventName.partition('_')[2] == index:
self.eventMap[eventName].eventActive = activityCommand
else:
self.eventMap[eventName].eventActive = activityCommand
def SetCArray(InputList, VarType, ArrayPointer):
if(isinstance(ArrayPointer, (list, tuple))):
raise TypeError('Cannot set a C array if it is actually a python list. Just assign the variable to the list directly.')
CmdString = 'sim_model.' + VarType + 'Array_setitem(ArrayPointer, CurrIndex, CurrElem)'
CurrIndex = 0
for CurrElem in InputList:
exec (CmdString)
CurrIndex += 1
def getCArray(varType, arrayPointer, arraySize):
CmdString = 'outputList.append(sim_model.' + varType + 'Array_getitem(arrayPointer, currIndex))'
outputList = []
currIndex = 0
for currIndex in range(arraySize):
exec (CmdString)
currIndex += 1
return outputList
def synchronizeTimeHistories(arrayList):
returnArrayList = arrayList
timeCounter = 0
for i in range(len(returnArrayList)):
while returnArrayList[i][0,0] > returnArrayList[0][timeCounter,0]:
timeCounter += 1
for i in range(len(returnArrayList)):
while(returnArrayList[i][1,0] < returnArrayList[0][timeCounter,0]):
returnArrayList[i] = np.delete(returnArrayList[i], 0, 0)
timeCounter = -1
for i in range(len(returnArrayList)):
while returnArrayList[i][-1,0] < returnArrayList[0][timeCounter,0]:
timeCounter -= 1
for i in range(len(returnArrayList)):
while(returnArrayList[i][-2,0] > returnArrayList[0][timeCounter,0]):
returnArrayList[i] = np.delete(returnArrayList[i], -1, 0)
timeNow = returnArrayList[0][0,0] #Desirement is to have synched arrays match primary time
outputArrayList = []
indexPrev = [0]*len(returnArrayList)
outputArrayList = [[]]*len(returnArrayList)
timeNow = returnArrayList[0][0,0]
outputArrayList[0] = returnArrayList[0][0:-2, :]
for i in range(1, returnArrayList[0].shape[0]-1):
for j in range(1, len(returnArrayList)):
while(returnArrayList[j][indexPrev[j]+1,0] < returnArrayList[0][i,0]):
indexPrev[j] += 1
dataProp = returnArrayList[j][indexPrev[j]+1,1:] - returnArrayList[j][indexPrev[j],1:]
dataProp *= (timeNow - returnArrayList[j][indexPrev[j],0])/(returnArrayList[j][indexPrev[j]+1,0] - returnArrayList[j][indexPrev[j],0])
dataProp += returnArrayList[j][indexPrev[j],1:]
dataRow = [timeNow]
dataRow.extend(dataProp.tolist())
outputArrayList[j].append(dataRow)
timePrevious = timeNow
timeNow = returnArrayList[0][i,0]
for j in range(1, len(returnArrayList)):
outputArrayList[j] = np.array(outputArrayList[j])
return outputArrayList