OSVE Callback Datapacks
The CALLBACK datapacks are intended to notify and provide to the OSVE caller program the steps, the data and events risen during the simulation.
Note that in the example, the filePath properties were set, but the logic says that
this filePath will not be required because no any file is going to be written.
But OSVE will use this “filePath” string as an identifier of the datapack, to be able to
notify to the OsveSubscriber linked to it with the specified data in the datapack definition.
An OsveSubscriber is a Python object with some callback methods invoked from the OSVE execute/simulation
process. The available callback methods to register to are onSimulationStart(), onSimulationTimeStep(),
onSimulationEnd(), onEventStateChanged(). For fully detailed comments about the OsveSubscribers,
read osve/subscribers/*.py documentation.
In the example above, the CALLBACK datapacks have the first column as the time in UTC format. The callbacksDatapack1 will then will provide the EPS_SA_POWER (Available solar arrays power) column and the callbacksDatapack2 will provide the SC quaternions.
So for each CALLBACK dataPack, the user shall indicate at least the filePath containing a string identifier, the ‘“type”: “CALLBACK”’ and at least one datapack field, usually the ‘time’ because is computationally straight forward.
How to implement an OsveSubscriber from Python scripting
To properly understand how these OSVE callbacks needs to be implemented, first we are going to explain the most general and flexible way to handle them, that is implementing a sub class of the OsveSubscriberAbstract class.
Inheriting the OsveSubscriberAbstract class
In the example below we are going to show how to register a general OSVE Subscriber in order to validate during the simulation the Spacecraft total body rates. In case that during the simulation the total body rate exceeds 0.02 degrees/sec2 the simulation shall be aborted. For doing this only overriding the function onSimulationTimeStep() would be necessary, but we implemented all of them just to show how they shall be implemented:
from osve import osve
from osve.subscribers.osve_subscriber_abstract import OsveSubscriberAbstract
class OsveSubscriber(OsveSubscriberAbstract):
def __init__(self):
super().__init__("dataPack1.csv")
def onSimulationStart(self, data) -> int:
# Notifies that simulation started, data contains the selected overlay name's and unit's,
# if returned int is smaller than 0 then simulation is aborted.
print ("onSimulationStart: " + str(data))
return 0
def onSimulationTimeStep(self, data) -> int:
# Notifies that a time step has been simulated, data contains the selected overlay name's and values,
# if returned int is smaller than 0 then simulation is aborted.
print ("onSimulationTimeStep: " + str(data))
if data["TOTAL_BODY_RATE"] > 0.02:
print ("TOTAL_BODY_RATE CONSTRAINT BREAK!! TOTAL_BODY_RATE: " + str(data["TOTAL_BODY_RATE"]))
return -1
return 0
def onSimulationEnd(self, data) -> int:
# Notifies that simulation finished, data contains the simulation end time,
# if returned int is smaller than 0 then simulation is aborted.
print ("onSimulationEnd: " + str(data))
return 0
def onEventStateChanged(self, data) -> int:
# Notifies that an event has changed its state, data contains the event properties,
# if returned int is smaller than 0 then simulation is aborted.
print ("onEventStateChanged: " + str(data))
return 0
test_input_path = "SOME_PATH_POINTING_TO_SCENARIO"
test_input_config_path = "SOME_PATH_POINTING_TO_SESSION_FILE"
# First, instantiate the OSVE Simulator
sim = osve.osve()
# Second, register the defined OsveSubscriber
sim.register_subscriber(OsveSubscriber())
# Finally, run simulation
sim.execute(test_input_path, test_input_config_path)
And this is how this dataPack shall be defined in the session file:
...
"outputFiles": {
...
"dataPacks": [
{
"filePath": "callbacksDatapack1",
"type": "CALLBACK",
"fields": [
{
"type": "time",
"format": "utc"
},
{
"type": "MAPPS",
"overlayId": "TOTAL_BODY_RATE"
}
]
}
]
},
...
Inheriting the OsveEventSubscriber class
In case that you only need to register to the risen OSVE Events, we suggest you to take advantage of this helper class intended to minimise the development effort:
from osve import osve
from osve.subscribers.osve_event_subscriber import OsveEventSubscriber
def onEventStateChanged(data) -> int:
# Notifies that an event has changed its state
# if returned int is smaller than 0 then simulation is aborted.
print ("onEventStateChanged: " + str(data))
return 0
test_input_path = "SOME_PATH_POINTING_TO_SCENARIO"
test_input_config_path = "SOME_PATH_POINTING_TO_SESSION_FILE"
# First, register required OsveSubscribers
theOsveSubscriber = OsveEventSubscriber("callbacksDatapack1", onEventStateChanged)
# Second, instantiate the OSVE Simulator
sim = osve.osve()
# Finally, run simulation
the_osve.execute(test_input_path, test_input_config_path)
And this is how this dataPack shall be defined in the session file:
...
"outputFiles": {
...
"dataPacks": [
{
"filePath": "callbacksDatapack1",
"type": "CALLBACK",
"fields": [
{
"type": "time",
"format": "utc"
}
]
}
]
},
...
Inheriting the OsvePtrAbstract class
The OsvePtrAbstract is a helper OSVE Subscriber class that allows to process the generated data per each executed PTR block. So classes inheriting it, shall implement a method “onPtrBlockEnd()” that will receive as an argument a dictionary with all the block details and generated data during the simulation of the block time period.
In this example we are just going to show the PTR Block details and the number of datapack steps data, the number of events risen and the generated logs since the PTR block start until its end:
from osve import osve
from osve.subscribers.osve_ptr_abstract import OsvePtrAbstract
class OsvePtrSubscriber(OsvePtrAbstract):
def __init__(self):
super().__init__("theOsvePtrSubscriber")
def onPtrBlockEnd(self, blockData) -> int:
print("onPtrBlockEnd -> " + str(blockData["block_start"]) + " - " + str(blockData["block_end"])
+ " T: " + str(blockData["block_type"])
+ " M: " + str(blockData["block_mode"])
+ " S-> " + str(len(blockData["block_steps"]))
+ " E-> " + str(len(blockData["block_events"]))
+ " L-> " + str(len(blockData["block_logs"])))
return 0
test_input_path = "SOME_PATH_POINTING_TO_SCENARIO"
test_input_config_path = "SOME_PATH_POINTING_TO_SESSION_FILE"
# First, instantiate the OSVE Simulator
sim = osve.osve()
# Second, register the OsvePtrSubscriber
theOsvePtrSubscriber = OsvePtrSubscriber()
sim.register_subscriber(theOsvePtrSubscriber)
sim.register_logger(theOsvePtrSubscriber)
# Finally, run simulation
sim.execute(test_input_path, test_input_config_path)
And this is how this dataPack shall be defined in the session file:
...
"outputFiles": {
...
"dataPacks": [
{
"filePath": "theOsvePtrSubscriber",
"type": "CALLBACK",
"fields": [
{
"type": "time",
"format": "utc"
},
{
"type": "MAPPS",
"overlayId": "TOTAL_BODY_RATE"
}
]
}
]
},
...
Grouping logs per PTR Block using the OsvePtrAbstract class
Here is a basic example of how to make a report per PTR Block of the log messages generated per PTR Block time span:
from osve import osve
from osve.subscribers.osve_ptr_abstract import OsvePtrAbstract
class OsvePtrLogger(OsvePtrAbstract):
blocks_data = []
def __init__(self):
super().__init__("theOsvePtrLogger")
def onPtrBlockEnd(self, blockData) -> int:
self.blocks_data.append(blockData)
return 0
def report(self):
print ("")
print ("----------------------------------------------")
print (" theOsvePtrLogger:")
print ("----------------------------------------------")
print ("")
idx = 0
for blockData in self.blocks_data:
print("BLOCK " + str(idx) + ", From: " + str(blockData["block_start"]) + " , To: " + str(blockData["block_end"]))
for log_data in blockData["block_logs"]:
print(" -> " + str(log_data["severity"]) + " , "
+ str(log_data["module"]) + " , "
+ str(log_data["time"]) + " , "
+ str(log_data["text"]))
idx += 1
test_input_path = "SOME_PATH_POINTING_TO_SCENARIO"
test_input_config_path = "SOME_PATH_POINTING_TO_SESSION_FILE"
# First, instantiate the OSVE Simulator
sim = osve.osve()
# Second, register the OsvePtrLogger
theOsvePtrLogger = OsvePtrLogger()
sim.register_subscriber(theOsvePtrLogger)
sim.register_logger(theOsvePtrLogger)
# Finally, run simulation and generate the report
sim.execute(test_input_path, test_input_config_path)
theOsvePtrLogger.report()
And this is how this dataPack shall be defined in the session file:
...
"outputFiles": {
...
"dataPacks": [
{
"filePath": "theOsvePtrLogger",
"type": "CALLBACK",
"fields": [
{
"type": "time",
"format": "utc"
}
]
}
]
},
...
Note that even we are not using this datapack “time” field data in our code, this field is required to allow OSVE to invoke “onPtrBlockEnd()” callback function, nevertheless, other callback functions needs to be invoked internally during the simulation.
More helper classes would be implemented in the future, such as a class to perform averages or the total sum of a series or profile. But all these will be intended only to minimise the development effort at Python side, because all these features could be implemented already using a subclass implementation of the OsveSubscriberAbstract class.
The OsveDatapack class
The OsveDatapack class intended to provide the requested overlays data obtained from the OSVE simulation.
Usage example:
theOsveDatapack = OsveDatapack()
theOsveDatapack.add_overlays(eps_utils.OSVE_OVERLAYS["POWER"])
theOsveDatapack.commit_changes(local_session_file)
the_osve.register_subscriber(theOsveDatapack)
if the_osve.execute(test_input_path, local_session_file) == 0:
data = theOsveDatapack.toNumpyArray()
start_time = data["time"][0]
end_time = data["time"][-1]
print("Simulation interval from " + str(data["time"][0]) + " to " + str(data["time"][-1]) +
" , duration: " + str((data["time"][-1] - data["time"][0]) / np.timedelta64(1, 's')))
print("Average power consumption (Watts): " + str(np.mean(data["TOTAL_POWER"])))
print("Maximum power consumption (Watts): " + str(np.max(data["TOTAL_POWER"])))
else:
print("OSVE execution failed. See log file.")
The OsveFiles class
The OsveFiles class intended to provide access to the EPS Files Layer of an EPS Mass-Memory Datastore
Each file in the files layer will have the following attributes:
fileName: Filename provided as parameter in the OPEN_FILE action call.
fileSize: The file size in Bytes.
creationDate: File creation/opening date in format YYYY-MM-DDThh:mm:ss
closingDate: File closing date in format YYYY-MM-DDThh:mm:ss
downlinkStartDate: Date at which the file is prepared/ready for downlink in format YYYY-MM-DDThh:mm:ss
downlinkDate: Date when the file completely downloaded in format YYYY-MM-DDThh:mm:ss. Once downlinked the file is deleted also.
deleteDate: Date when the file is deleted in format YYYY-MM-DDThh:mm:ss. Could be deleted because just downlinked.
Usage example:
# Get local session file
session_file_local = write_local_session_file(test_input_config_path, "", "")
# Instantiate the OsveFiles class
the_osve_files = OsveFiles()
# Add data stores to look at
the_osve_files.add_datastore("SSMM_HIGH_RES", "SSMM_RS_BULK")
the_osve_files.add_datastore("SSMM_HIGH_RES", "SSMM_RS_SELECTED")
the_osve_files.add_datastore("SSMM_LOW_RES", "SSMM_RS_SELECTOR")
# Save changes on local session file
the_osve_files.commit_changes(session_file_local)
# Register the OSVE subscriber
the_osve.register_subscriber(the_osve_files)
# Run the simulation
the_osve.execute(test_input_path, session_file_local)
# Print all mentioned datastore files processed by OSVE/EPS
print(the_osve_files.getFilesMap())
Example of result:
{
"EPS_DS_FILES_SSMM_HIGH_RES_SSMM_RS_BULK":[
{
"closingDate":"",
"creationDate":"2032-06-01T04:45:40",
"deleteDate":"",
"downlinkDate":"",
"downlinkStartDate":"",
"fileName":"OPENED_FILE",
"fileSize":0
},
{
"closingDate":"2032-06-01T09:45:40",
"creationDate":"2032-06-01T08:45:40",
"deleteDate":"",
"downlinkDate":"",
"downlinkStartDate":"",
"fileName":"OPENED_FILE",
"fileSize":0
}
],
"EPS_DS_FILES_SSMM_HIGH_RES_SSMM_RS_SELECTED":[
{
"closingDate":"2032-06-01T04:15:40",
"creationDate":"2032-06-01T03:45:40",
"deleteDate":"2032-06-01T21:59:22",
"downlinkDate":"2032-06-01T21:59:22",
"downlinkStartDate":"2032-06-01T05:35:40",
"fileName":"DOWNLINKED_FILE",
"fileSize":1125000
},
{
"closingDate":"2032-06-01T03:15:40",
"creationDate":"2032-06-01T02:45:40",
"deleteDate":"2032-06-01T22:03:42",
"downlinkDate":"",
"downlinkStartDate":"2032-06-01T22:01:32",
"fileName":"DOWNLINKING_FILE",
"fileSize":1125000
},
{
"closingDate":"2032-06-01T06:15:40",
"creationDate":"2032-06-01T05:45:40",
"deleteDate":"2032-06-01T07:45:40",
"downlinkDate":"",
"downlinkStartDate":"",
"fileName":"DELETED_FILE",
"fileSize":1125000
}
],
"EPS_DS_FILES_SSMM_LOW_RES_SSMM_RS_SELECTOR":[
{
"closingDate":"2032-06-01T03:15:40",
"creationDate":"2032-06-01T02:45:40",
"deleteDate":"2032-06-01T03:15:42",
"downlinkDate":"2032-06-01T03:15:42",
"downlinkStartDate":"2032-06-01T03:15:40",
"fileName":"File_1_Thumb",
"fileSize":22499
}
...
]
}
The OsveFiles onFilesMapChangedFn callback function
The OsveFiles class init method support to pass a function pointer called as onFilesMapChangedFn. By default this pointer is not passed as an argument.
But the user can define a custom function following this scheme in order to have access to the updates performed at any Files Layer registered.
With this the user could abort simulation if one file reach it maximum size, or for example, if one file takes too much to be closed or down linked.
Example of avoiding files bigger than 5Kb:
# Define the function to handle and manage File Layers updates
def onFilesMapChanged(t, map):
# For each datastore in the map
for ds_key in map:
ds_files = map[ds_key]
# For each file in the datastore
for file in ds_files:
if file["fileSize"] > 5e+3:
# File size bigger than 5Kb, report error
print ("SIMULATION ABORTED!")
print ("File: " + str(file["fileName"]) + " is bigger than 5Kb, at time: " + str(t))
# Abort simulation by returning < 0
return -1
# Continue simulation
return 0
# Get local session file
session_file_local = write_local_session_file(test_input_config_path, "", "")
# Instantiate the OsveFiles class --WHILE PASSING FUNCTION POINTER--
the_osve_files = OsveFiles(onFilesMapChanged)
# Add data stores to look at
the_osve_files.add_datastore("SSMM_HIGH_RES", "SSMM_RS_BULK")
the_osve_files.add_datastore("SSMM_HIGH_RES", "SSMM_RS_SELECTED")
the_osve_files.add_datastore("SSMM_LOW_RES", "SSMM_RS_SELECTOR")
# Save changes on local session file
the_osve_files.commit_changes(session_file_local)
# Register the OSVE subscriber
the_osve.register_subscriber(the_osve_files)
# Run the simulation
the_osve.execute(test_input_path, session_file_local)