from numbers import Integral
import numpy as np
from Basilisk.utilities import deprecated, simHelpers
_DIRECT_VARIABLE_ERROR = (
"addVariableLog supports direct module variables only; create a "
"PythonVariableLogger during simulation setup for nested or computed time "
"histories and retain its output with addRetentionFunction()"
)
[docs]
class VariableRetentionParameters:
"""Store the configuration and runtime logger for a retained variable.
:param str varName: Variable identifier in ``<ModelTag>.<variableName>`` format.
:param int varRate: Minimum variable recording period in nanoseconds.
:param int startIndex: Deprecated first retained component index.
:param int stopIndex: Deprecated last retained component index, inclusive.
:param str varType: Deprecated legacy C-array type name.
"""
def __init__(
self,
varName,
varRate,
startIndex=None,
stopIndex=None,
varType=None,
):
self.varName = varName
self.varRate = varRate
self.startIndex = startIndex
self.stopIndex = stopIndex
self.varType = varType
self.logger = None
[docs]
class MessageRetentionParameters:
"""Store the configuration for retained message fields.
:param str name: Name of the message recorder.
:param list retainedVars: Message payload fields to retain.
"""
def __init__(self, name, retainedVars):
self.msgRecName = name
self.retainedVars = retainedVars
[docs]
class RetentionPolicy:
"""Control which simulation data is retained and how it is stored.
The first column of each retained message or variable array contains the
corresponding simulation time in nanoseconds.
"""
def __init__(self, rate=10_000_000_000): # [ns]
self.logRate = rate
self.messageLogList = []
self.varLogList = []
self.dataCallback = None
self.retentionFunctions = []
def addMessageLog(self, name, retainedVars):
self.messageLogList.append(MessageRetentionParameters(name, retainedVars))
[docs]
def addVariableLog(
self,
variableName,
startIndex=None,
stopIndex=None,
varType=None,
logRate=None,
):
"""Add a module variable to the retained Monte Carlo data.
The variable identifier must use ``<ModelTag>.<variableName>`` format
and refer to a direct public or getter-backed variable on a uniquely
tagged module that supports the standard ``logger()`` API. The final
period separates the model tag from the variable name, so model tags may
contain periods. The complete variable is retained by default;
multidimensional samples are flattened in row-major order.
``startIndex``, ``stopIndex``, and ``varType`` are compatibility
arguments for the removed legacy variable-logging API and will be
removed after 2027-08-26. New code should select component columns from
the retained NumPy array instead.
:param str variableName: Model tag and direct module variable name.
:param int startIndex: Deprecated first retained component index.
:param int stopIndex: Deprecated last retained component index, inclusive.
:param str varType: Deprecated legacy C-array type name; ignored.
:param int logRate: Minimum recording period in nanoseconds. The policy
default is used when this argument is ``None``.
"""
self._splitVariableName(variableName)
startIndex, stopIndex = self._normalizeLegacyIndices(
startIndex,
stopIndex,
)
if startIndex is not None or stopIndex is not None or varType is not None:
deprecated.deprecationWarn(
"RetentionPolicy.addVariableLog legacy arguments",
"2027/08/26",
"Variable loggers now retain complete values directly. Select "
"components from the retained NumPy array instead.",
)
if logRate is None:
logRate = self.logRate
logRate = self._normalizeLogRate(logRate)
variableParameters = VariableRetentionParameters(
variableName,
logRate,
startIndex,
stopIndex,
varType,
)
self.varLogList.append(variableParameters)
@staticmethod
def _normalizeLegacyIndices(startIndex, stopIndex):
"""Normalize and validate a deprecated inclusive component range."""
if startIndex is None and stopIndex is None:
return None, None
if startIndex is None:
startIndex = 0
if stopIndex is None:
stopIndex = startIndex
if any(
isinstance(index, bool) or not isinstance(index, Integral)
for index in (startIndex, stopIndex)
):
raise TypeError("startIndex and stopIndex must be integers")
if startIndex < 0 or stopIndex < startIndex:
raise ValueError(
"startIndex and stopIndex must define a nonnegative inclusive range"
)
return int(startIndex), int(stopIndex)
@staticmethod
def _normalizeLogRate(logRate):
"""Validate and normalize a variable recording period in nanoseconds."""
if isinstance(logRate, bool) or not isinstance(logRate, Integral):
raise TypeError("logRate must be an integer number of nanoseconds")
if logRate < 0:
raise ValueError("logRate must be nonnegative")
return int(logRate)
@staticmethod
def _splitVariableName(variableName):
"""Split and validate a retained module variable identifier."""
if not isinstance(variableName, str) or variableName == "":
raise ValueError("variableName must be a non-empty string")
modelTag, separator, moduleVariableName = variableName.rpartition(".")
if separator == "" or modelTag == "" or moduleVariableName == "":
raise ValueError(
"variableName must use '<ModelTag>.<variableName>' format"
)
if any(token in moduleVariableName for token in ("[", "]", "(", ")")):
raise ValueError(_DIRECT_VARIABLE_ERROR)
return modelTag, moduleVariableName
@staticmethod
def _findModelAndTask(simInstance, modelTag, retainedVariableName):
"""Return the uniquely tagged model and its owning simulation task."""
matches = []
prefixModelTags = set()
for task in simInstance.TaskList:
for model in task.TaskModels:
candidateTag = getattr(model, "ModelTag", None)
if candidateTag == modelTag:
matches.append((model, task))
elif (
isinstance(candidateTag, str)
and candidateTag != ""
and retainedVariableName.startswith(candidateTag + ".")
):
prefixModelTags.add(candidateTag)
if not matches:
if prefixModelTags:
matchingTags = ", ".join(sorted(prefixModelTags))
raise ValueError(
f"Could not find a model with ModelTag '{modelTag}'. The "
f"identifier extends existing ModelTag(s): {matchingTags}. "
f"{_DIRECT_VARIABLE_ERROR}."
)
raise ValueError(f"Could not find a model with ModelTag '{modelTag}'")
if len(matches) > 1:
raise ValueError(
f"ModelTag '{modelTag}' is not unique across simulation tasks"
)
return matches[0]
@staticmethod
def _getLoggerPriority(task):
"""Return a priority that schedules a logger after existing task models."""
priorities = getattr(task, "TaskModelPriorities", None)
if priorities is None or len(priorities) != len(task.TaskModels):
raise RuntimeError(
f"Task '{task.Name}' has inconsistent model-priority metadata"
)
if not priorities:
raise RuntimeError(f"Task '{task.Name}' does not contain any models")
# SysModelTask preserves insertion order for models with equal priority.
# Reusing the lowest priority therefore places the new logger last.
return min(priorities)
[docs]
def addLogsToSim(self, simInstance):
"""Create and schedule the variable loggers required by this policy."""
RetentionPolicy.addRetentionPoliciesToSim(simInstance, [self])
def _addLogsToSim(self, simInstance, loggerCache):
"""Create variable loggers, sharing compatible duplicate requests."""
for variable in self.varLogList:
modelTag, moduleVariableName = self._splitVariableName(variable.varName)
cachedLogger = loggerCache.get(variable.varName)
if cachedLogger is not None:
variable.logger = cachedLogger
continue
model, task = self._findModelAndTask(
simInstance,
modelTag,
variable.varName,
)
loggerFactory = getattr(model, "logger", None)
if not callable(loggerFactory):
raise TypeError(
f"Model '{modelTag}' does not provide the standard logger() "
"API; create a PythonVariableLogger during simulation setup "
"and retain its output with addRetentionFunction()"
)
variableLogger = loggerFactory(moduleVariableName, variable.varRate)
variableLogger.ModelTag = f"RetentionLogger:{variable.varName}"
loggerPriority = self._getLoggerPriority(task)
simInstance.AddModelToTask(
task.Name,
variableLogger,
ModelPriority=loggerPriority,
)
variable.logger = variableLogger
loggerCache[variable.varName] = variableLogger
[docs]
def addRetentionFunction(self, function):
"""Add a callback that returns custom data after simulation execution.
The callback receives the completed simulation instance and must return
a dictionary. Its entries are merged into the retained ``custom`` data.
:param callable function: Post-simulation data extraction callback.
"""
self.retentionFunctions.append(function)
def setDataCallback(self, dataCallback):
self.dataCallback = dataCallback
def executeCallback(self, data):
if self.dataCallback is not None:
self.dataCallback(data, self)
[docs]
@staticmethod
def addRetentionPoliciesToSim(simInstance, retentionPolicies):
"""Add the variable loggers from a list of policies to a simulation.
Compatible duplicate requests share one logger. Requests for the same
retained-data key with different rates or component ranges are rejected.
:param simInstance: Simulation instance receiving the loggers.
:param list retentionPolicies: Retention policies defining data to log.
"""
retentionPolicies = list(retentionPolicies)
requestedSignatures = {}
for retentionPolicy in retentionPolicies:
for variable in retentionPolicy.varLogList:
loggerSignature = (
variable.varRate,
variable.startIndex,
variable.stopIndex,
)
existingSignature = requestedSignatures.get(variable.varName)
if (
existingSignature is not None
and existingSignature != loggerSignature
):
raise ValueError(
f"Variable '{variable.varName}' has conflicting retention "
"settings"
)
requestedSignatures[variable.varName] = loggerSignature
loggerCache = {}
for retentionPolicy in retentionPolicies:
retentionPolicy._addLogsToSim(simInstance, loggerCache)
@staticmethod
def _getRetainedVariableData(variable):
"""Return one runtime variable logger as a time-column data array."""
if variable.logger is None:
raise RuntimeError(
f"Variable logger '{variable.varName}' was not added to the simulation"
)
_, moduleVariableName = RetentionPolicy._splitVariableName(variable.varName)
logTimes = variable.logger.times()
variableData = np.asarray(variable.logger[moduleVariableName])
sampleCount = len(logTimes)
if sampleCount == 0:
componentCount = (
variable.stopIndex-variable.startIndex+1
if variable.startIndex is not None
else 0
)
emptyData = np.empty((0, componentCount))
return simHelpers.addTimeColumn(logTimes, emptyData)
if variableData.ndim == 0 or variableData.shape[0] != sampleCount:
raise RuntimeError(
f"Variable logger '{variable.varName}' returned inconsistent sample data"
)
componentCount = variableData.size//sampleCount
componentData = variableData.reshape(sampleCount, componentCount)
if variable.startIndex is not None:
if variable.stopIndex >= componentData.shape[1]:
raise IndexError(
f"Legacy component range [{variable.startIndex}, "
f"{variable.stopIndex}] exceeds the width of '{variable.varName}'"
)
componentData = componentData[
:, variable.startIndex:variable.stopIndex+1
]
return simHelpers.addTimeColumn(logTimes, componentData)
[docs]
@staticmethod
def getDataForRetention(simInstance, retentionPolicies):
"""Return the data selected by a list of retention policies.
The returned dictionary contains ``messages``, ``variables``, and
``custom`` sub-dictionaries. Message and variable arrays have simulation
time prepended as their first column. Multidimensional variable samples
are flattened in row-major order.
:param simInstance: Simulation instance containing completed recorders.
:param list retentionPolicies: Policies defining the data to retain.
:return: Retained simulation data grouped by source.
:rtype: dict
"""
data = {"messages": {}, "variables": {}, "custom": {}}
for retentionPolicy in retentionPolicies:
for msgParam in retentionPolicy.messageLogList:
recorder = simInstance.msgRecList[msgParam.msgRecName]
msgTimes = recorder.times()
for varName in msgParam.retainedVars:
msgData = getattr(recorder, varName)
msgData = simHelpers.addTimeColumn(msgTimes, msgData)
messageKey = f"{msgParam.msgRecName}.{varName}"
data["messages"][messageKey] = msgData
for variable in retentionPolicy.varLogList:
retainedVariableData = RetentionPolicy._getRetainedVariableData(
variable
)
data["variables"][variable.varName] = retainedVariableData
for retentionFunction in retentionPolicy.retentionFunctions:
customData = retentionFunction(simInstance)
data["custom"].update(customData)
return data