Hands-on Tutorial
This tutorial contains information for those who are just starting out and builds up to show how complex test suites can be built.
Introduction
The New Observing Strategies Testbed (NOS-T) is a computational environment to develop, test, mature, and socialize new operating concepts and technology for NOS. NOS-T provides infrastructure to integrate and orchestrate user-contributed applications for system-of-systems test cases with true distributed control over constituent systems. The overall concept, illustrated below, interconnects individual user applications and a NOS-T manager application via common information system infrastructure to coordinate the execution of virtual Earth science missions. NOS-T enables principal investigators to conduct test runs in the same environment, systematically changing variables to assess the overall efficacy of the proposed new observing strategies. Recorded data and outcomes provide evidence to advance technology readiness level and improve or innovate upon existing Earth science measurement techniques.
Setup
This section will show you how to set up NOS-T assuming you are a beginner to both coding and the testbed.
Integrated Development Environment (IDE)
An IDE will make developing applications and interacting with the testbed much easier. The developers have mostly used Spyder and Microsoft’s Visual Studio Code. Going forward, this tutorial will assume that you are using one of these IDE’s or something similar.
NOS-T Tools Download and Installation
The best way to get the NOS-T tools library and example codes is to clone the NOS-T git repository and install the tools.
Cloning the Repository
There are several ways to clone a git repository. Here is a good description of some of them.
Then, you need to clone the repository from the following link:
https://github.com/code-lab-org/nost-tools
Installing NOS-T tools requires pip version 23 or greater. Install via
python -m pip install --upgrade pip
Then, from a command prompt, navigate to the root directory (the location where you cloned the library) and install by running the following command:
pip install -e .[examples]
Following the instructions above will automatically install the python packages that NOS-T depends on to run. These package dependencies can otherwise be found in the requirements file.
NOS-T System description
The NOS-T system architecture follows a loosely coupled event-driven architecture (EDA) where member applications communicate state changes through events that are embodied as notification messages sent over a network. These event/message payloads contain the relevant data for communicating these state changes. EDA provides enhanced scalability and reliability over other software architectures by replicating event handling functions across infrastructure instances while maintaining modularity between applications through a simple event-handling interface. NOS-T can also be described as a service-oriented architecture (SOA) as applications trigger services in response to events.
The NOS-T architecture relies on a centralized infrastructure component called an event broker (synonymous with message broker) to exchange event notifications between applications. A broker simplifies the communication structure because each member application (client) only directly connects to the broker, rather than requiring each application to directly connect to every other application.
The FireSat+ Test Suite
From here, the tutorial will explain important functions using FireSat+, an example NOS-T test suite based on FireSat, the common space systems engineering application case. The operational concept for FireSat+ is that one or several satellites are searching for fires. The fires are ignited following a historical dataset. When one of the satellites orbit above these locations, it will detect the fire. Finally, once that satellite is in range of a ground station, it will report the fire.
This is a graphical representation of the FireSat+ message flows and their payloads.
For more information on FireSat+, please see the following:
The Interface Control Document has a high-level description of FireSat+ here.
A deeper dive into the applications and code is here.
A paper describing this test suite is here.
NOS-T test suites are made up of applications communicating over the broker. Next, the tutorial will detail two of the FireSat+ apps to give you a better idea of how they work.
The Satellites application - main_constellation.py
A key component of the FireSat+ example case is Satellite application. This application enables the user to generation a satellite constellation using the nost-tools library, leveraging predefined templates to construct a model of a real-life constellation. You will be guided through the how each code block works, to help understand the purpose of different components in an application.
First, you’ll need to create a new file titled main_constellation.py inside your examples/firesat_tutorial/satellites folder. To progress through this section, copy and paste the code blocks into this file. NOTE: You must maintain the indentations you see in these code blocks when pasting them into the main_constellation.py file.
This first part of the code contains import statements allow you to install the necessary dependencies to construct the application. The group at the top are regular Python dependencies while the ones at the bottom draw from the NOS-T tools library.
import logging
from datetime import datetime, timezone, timedelta
from dotenv import dotenv_values
import numpy as np
import pandas as pd
import copy
from skyfield.api import load, wgs84, EarthSatellite
from nost_tools.application_utils import ConnectionConfig, ShutDownObserver
from nost_tools.entity import Entity
from nost_tools.observer import Observer
from nost_tools.managed_application import ManagedApplication
from nost_tools.publisher import WallclockTimeIntervalPublisher
This next set of import statements are customized for FireSat+ values from the constellation configuration files. The first set of imports draws in the message schema configuration, which defines the structure of how Satellites communicates data. The second set of imports pulls in values to define the constellation: the PREFIX the messages will be published on, the NAME of the satellite, the SCALE of the timed simulation,
the two-line element sets (TLEs) that define the satellites’ orbit, and the FIELD_OF_REGARD, which indicates the region visible on Earth by the satellite’s instrument.
from constellation_config_files.schemas import (
FireStarted,
FireDetected,
FireReported,
SatelliteStatus,
GroundLocation,
)
from constellation_config_files.config import (
PREFIX,
NAME,
SCALE,
TLES,
FIELD_OF_REGARD,
)
The first line in the code block below sets up a logger to help you track what is going on. More info on the various levels can be found
here. Next, the function, compute_min_elevation, returns the minimum elevation angle required for a satellite to observe a point from it’s current location. It accepts the parameters altitude and field_of_regard to
complete mathematical functions to return the degree on minimum elevation.
logging.basicConfig(level=logging.INFO)
def compute_min_elevation(altitude, field_of_regard):
"""
Computes the minimum elevation angle required for a satellite to observe a point from current location.
Args:
altitude (float): Altitude (meters) above surface of the observation
field_of_regard (float): Angular width (degrees) of observation
Returns:
float : min_elevation
The minimum elevation angle (degrees) for observation
"""
earth_equatorial_radius = 6378137.000000000
earth_polar_radius = 6356752.314245179
earth_mean_radius = (2 * earth_equatorial_radius + earth_polar_radius) / 3
# eta is the angular radius of the region viewable by the satellite
sin_eta = np.sin(np.radians(field_of_regard / 2))
# rho is the angular radius of the earth viewed by the satellite
sin_rho = earth_mean_radius / (earth_mean_radius + altitude)
# epsilon is the min satellite elevation for obs (grazing angle)
cos_epsilon = sin_eta / sin_rho
if cos_epsilon > 1:
return 0.0
return np.degrees(np.arccos(cos_epsilon))
Next, the compute_sensor_radius function pulls in the result of compute_min_elevation and the altitude value to return sensor_radius, which provides the radius of the nadir pointing sensor’s circular view projected onto Earth.
def compute_sensor_radius(altitude, min_elevation):
"""
Computes the sensor radius for a satellite at current altitude given minimum elevation constraints.
Args:
altitude (float): Altitude (meters) above surface of the observation
min_elevation (float): Minimum angle (degrees) with horizon for visibility
Returns:
float : sensor_radius
The radius (meters) of the nadir pointing sensors circular view of observation
"""
earth_equatorial_radius = 6378137.0
earth_polar_radius = 6356752.314245179
earth_mean_radius = (2 * earth_equatorial_radius + earth_polar_radius) / 3
# rho is the angular radius of the earth viewed by the satellite
sin_rho = earth_mean_radius / (earth_mean_radius + altitude)
# eta is the nadir angle between the sub-satellite direction and the target location on the surface
eta = np.degrees(np.arcsin(np.cos(np.radians(min_elevation)) * sin_rho))
# calculate swath width half angle from trigonometry
sw_HalfAngle = 90 - eta - min_elevation
if sw_HalfAngle < 0.0:
return 0.0
return earth_mean_radius * np.radians(sw_HalfAngle)
The get_elevation_angle is a function that uses the Skyfield library. It accepts the parameters t, sat, and loc. The first two, respectively, represent the Skyfield time object, the Skyfield EarthSat object. The third is the latitude/longitude of the spacecraft’s subpoint, along with the spacecraft altitude. It returns an elevation angle in respect to the topocentric horizon.
def get_elevation_angle(t, sat, loc):
"""
Returns the elevation angle (degrees) of satellite with respect to the topocentric horizon.
Args:
t (:obj:`Time`): Time object of skyfield.timelib module
sat (:obj:`EarthSatellite`): Skyview EarthSatellite object from skyfield.sgp4lib module
loc (:obj:`GeographicPosition`): Geographic location on surface specified by latitude-longitude from skyfield.toposlib module
Returns:
float : alt.degrees
Elevation angle (degrees) of satellite with respect to the topocentric horizon
"""
difference = sat - loc
topocentric = difference.at(t)
# NOTE: Topos uses term altitude for what we are referring to as elevation
alt, az, distance = topocentric.altaz()
return alt.degrees
These two functions, check_in_view and check_in_range, affirm if the elevation angle and immediate location of the satellite enable it to connect to a ground station and view regions on Earth.
def check_in_view(t, satellite, topos, min_elevation):
"""
Checks if the elevation angle of the satellite with respect to the ground location is greater than the minimum elevation angle constraint.
Args:
t (:obj:`Time`): Time object of skyfield.timelib module
satellite (:obj:`EarthSatellite`): Skyview EarthSatellite object from skyfield.sgp4lib module
topos (:obj:`GeographicPosition`): Geographic location on surface specified by latitude-longitude from skyfield.toposlib module
min_elevation (float): Minimum elevation angle (degrees) for ground to be in view of satellite, as calculated by compute_min_elevation
Returns:
bool : isInView
True/False indicating visibility of ground location to satellite
"""
isInView = False
elevationFromFire = get_elevation_angle(t, satellite, topos)
if elevationFromFire >= min_elevation:
isInView = True
return isInView
def check_in_range(t, satellite, grounds):
"""
Checks if the satellite is in range of any of the operational ground stations.
Args:
t (:obj:`Time`): Time object of skyfield.timelib module
satellite (:obj:`EarthSatellite`): Skyview EarthSatellite object from skyfield.sgp4lib module
grounds (:obj:`DataFrame`): Dataframe of ground station locations, minimum elevation angles for communication, and operational status (T/F)
Returns:
bool, int :
isInRange
True/False indicating visibility of satellite to any operational ground station
groundId
groundId of the ground station currently in comm range (NOTE: If in range of two ground stations simultaneously, will return first groundId)
"""
isInRange = False
groundId = None
for k, ground in grounds.iterrows():
if ground.operational:
groundLatLon = wgs84.latlon(ground.latitude, ground.longitude)
satelliteElevation = get_elevation_angle(t, satellite, groundLatLon)
if satelliteElevation >= ground.elevAngle:
isInRange = True
groundId = k
break
return isInRange, groundId
Constellation class
The next section of code blocks define the Constellation class. In object-oriented programming, a class is a replicable object that can be assigned unique parameters to generate a diverse collection of similar objects. The Constellation class leverages the NOS-T tools library ‘Entity’ object class to construct the constellation chain.
The first two functions in the Constellation class, init and initialize, prepare the test run for startup by initializing data.
# define an entity to manage satellite updates
class Constellation(Entity):
"""
*This object class inherits properties from the Entity object class in the NOS-T tools library*
Args:
cName (str): A string containing the name for the constellation application
app (:obj:`ManagedApplication`): An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions
id (list): List of unique *int* ids for each satellite in the constellation
names (list): List of unique *str* for each satellite in the constellation (must be same length as **id**)\n
ES (list): Optional list of :obj:`EarthSatellite` objects to be included in the constellation (NOTE: at least one of **ES** or **tles** MUST be specified, or an exception will be thrown)\n
tles (list): Optional list of Two-Line Element *str* to be converted into :obj:`EarthSatellite` objects and included in the simulation
Attributes:
fires (list): List of fires with unique fireId (*int*), ignition (:obj:`datetime`), and latitude-longitude location (:obj:`GeographicPosition`) - *NOTE:* initialized as [ ]
grounds (:obj:`DataFrame`): Dataframe containing information about ground stations with unique groundId (*int*), latitude-longitude location (:obj:`GeographicPosition`), min_elevation (*float*) angle constraints, and operational status (*bool*) - *NOTE:* initialized as **None**
satellites (list): List of :obj:`EarthSatellite` objects included in the constellation - *NOTE:* must be same length as **id**
detect (list): List of detected fires with unique fireId (*int*), detected :obj:`datetime`, and name (*str*) of detecting satellite - *NOTE:* initialized as [ ]
report (list): List of reported fires with unique fireId (*int*), reported :obj:`datetime`, name (*str*) of reporting satellite, and groundId (*int*) of ground station reported to - *NOTE:* initialized as [ ]
positions (list): List of current latitude-longitude-altitude locations (:obj:`GeographicPosition`) of each satellite in the constellation - *NOTE:* must be same length as **id**
next_positions (list): List of next latitude-longitude-altitude locations (:obj:`GeographicPosition`) of each satellite in the constellation - *NOTE:* must be same length as **id**
min_elevations_fire (list): List of *floats* indicating current elevation angle (degrees) constraint for visibility by each satellite - *NOTE:* must be same length as **id**, updates every time step
"""
ts = load.timescale()
PROPERTY_FIRE_REPORTED = "reported"
PROPERTY_FIRE_DETECTED = "detected"
PROPERTY_POSITION = "position"
def __init__(self, cName, app, id, names, ES=None, tles=None):
super().__init__(cName)
self.app = app
self.id = id
self.names = names
self.fires = []
self.grounds = None
self.satellites = []
if ES is not None:
for satellite in ES:
self.satellites.append(satellite)
if tles is not None:
for i, tle in enumerate(tles):
self.satellites.append(
EarthSatellite(tle[0], tle[1], self.names[i], self.ts)
)
self.detect = []
self.report = []
self.positions = self.next_positions = [None for satellite in self.satellites]
self.min_elevations_fire = [
compute_min_elevation(
wgs84.subpoint(satellite.at(satellite.epoch)).elevation.m,
FIELD_OF_REGARD[i],
)
for i, satellite in enumerate(self.satellites)
]
def initialize(self, init_time):
"""
Activates the :obj:`Constellation` at a specified initial scenario time
Args:
init_time (:obj:`datetime`): Initial scenario time for simulating propagation of satellites
"""
super().initialize(init_time)
self.grounds = pd.DataFrame(
{
"groundId": pd.Series([], dtype="int"),
"latitude": pd.Series([], dtype="float"),
"longitude": pd.Series([], dtype="float"),
"elevAngle": pd.Series([], dtype="float"),
"operational": pd.Series([], dtype="bool"),
}
)
self.positions = self.next_positions = [
wgs84.subpoint(satellite.at(self.ts.from_datetime(init_time)))
for satellite in self.satellites
]
The next two functions, tick and tock, are very important for executing time-managed test suites. Generally, the tick function computes the current state of an application. Any cumbersome functions like simulations should be performed here. The tock function commits the state changes. You want this done as quickly as possible to maintain consistent timing between applications.
def tick(self, time_step):
"""
Computes the next :obj:`Constellation` state after the specified scenario duration and the next simulation scenario time
Args:
time_step (:obj:`timedelta`): Duration between current and next simulation scenario time
"""
super().tick(time_step)
self.next_positions = [
wgs84.subpoint(
satellite.at(self.ts.from_datetime(self.get_time() + time_step))
)
for satellite in self.satellites
]
for i, satellite in enumerate(self.satellites):
then = self.ts.from_datetime(self.get_time() + time_step)
self.min_elevations_fire[i] = compute_min_elevation(
float(self.next_positions[i].elevation.m), FIELD_OF_REGARD[i]
)
for j, fire in enumerate(self.fires):
if self.detect[j][self.names[i]] is None:
topos = wgs84.latlon(fire["latitude"], fire["longitude"])
isInView = check_in_view(
then, satellite, topos, self.min_elevations_fire[i]
)
if isInView:
self.detect[j][self.names[i]] = (
self.get_time() + time_step
) # TODO could use event times
if self.detect[j]["firstDetector"] is None:
self.detect[j]["firstDetect"] = True
self.detect[j]["firstDetector"] = self.names[i]
if (self.detect[j][self.names[i]] is not None) and (
self.report[j][self.names[i]] is None
):
isInRange, groundId = check_in_range(then, satellite, self.grounds)
if isInRange:
self.report[j][self.names[i]] = self.get_time() + time_step
if self.report[j]["firstReporter"] is None:
self.report[j]["firstReport"] = True
self.report[j]["firstReporter"] = self.names[i]
self.report[j]["firstReportedTo"] = groundId
def tock(self):
"""
Commits the next :obj:`Constellation` state and advances simulation scenario time
"""
self.positions = self.next_positions
for i, newly_detected_fire in enumerate(self.detect):
if newly_detected_fire["firstDetect"]:
detector = newly_detected_fire["firstDetector"]
self.notify_observers(
self.PROPERTY_FIRE_DETECTED,
None,
{
"fireId": newly_detected_fire["fireId"],
"detected": newly_detected_fire[detector],
"detected_by": detector,
},
)
self.detect[i]["firstDetect"] = False
for i, newly_reported_fire in enumerate(self.report):
if newly_reported_fire["firstReport"]:
reporter = newly_reported_fire["firstReporter"]
self.notify_observers(
self.PROPERTY_FIRE_REPORTED,
None,
{
"fireId": newly_reported_fire["fireId"],
"reported": newly_reported_fire[reporter],
"reported_by": reporter,
"reported_to": newly_reported_fire["firstReportedTo"],
},
)
self.report[i]["firstReport"] = False
super().tock()
The next function, on_fire, checks the current simulation time vs. a database of actual fires detected by an space-based infrared sensor. This function then publishes a message containing information about the fire. It also maintains an internal database for when fires are detected and reported, and which satellite did the detecting/reporting.
def on_fire(self, client, userdata, message):
"""
Callback function appends a dictionary of information for a new fire to fires :obj:`list` when message detected on the *PREFIX/fires/location* topic
Args:
client (:obj:`MQTT Client`): Client that connects application to the event broker using the MQTT protocol. Includes user credentials, tls certificates, and host server-port information.
userdata: User defined data of any type (not currently used)
message (:obj:`message`): Contains *topic* the client subscribed to and *payload* message content as attributes
"""
started = FireStarted.parse_raw(message.payload)
self.fires.append(
{
"fireId": started.fireId,
"start": started.start,
"latitude": started.latitude,
"longitude": started.longitude,
},
)
satelliteDictionary = dict.fromkeys(
self.names
) # Creates dictionary where keys are satellite names and values are defaulted to NoneType
satelliteDictionary[
"fireId"
] = (
started.fireId
) # Adds fireId to dictionary, which will coordinate with position of dictionary in list of dictionaries
detectDictionary = copy.deepcopy(satelliteDictionary)
detectDictionary["firstDetect"] = False
detectDictionary["firstDetector"] = None
self.detect.append(detectDictionary)
reportDictionary = copy.deepcopy(satelliteDictionary)
reportDictionary["firstReport"] = False
reportDictionary["firstReporter"] = None
reportDictionary["firstReportedTo"] = None
self.report.append(reportDictionary)
The final block of the Constellation class is next. It contains the on_ground function which is used to collect information on ground station locations and elevation angles when those messages are published.
def on_ground(self, client, userdata, message):
"""
Callback function appends a dictionary of information for a new ground station to grounds :obj:`list` when message detected on the *PREFIX/ground/location* topic. Ground station information is published at beginning of simulation, and the :obj:`list` is converted to a :obj:`DataFrame` when the Constellation is initialized.
Args:
client (:obj:`MQTT Client`): Client that connects application to the event broker using the MQTT protocol. Includes user credentials, tls certificates, and host server-port information.
userdata: User defined data of any type (not currently used)
message (:obj:`message`): Contains *topic* the client subscribed to and *payload* message content as attributes
"""
location = GroundLocation.parse_raw(message.payload)
if location.groundId in self.grounds.groundId:
self.grounds[
self.grounds.groundId == location.groundId
].latitude = location.latitude
self.grounds[
self.grounds.groundId == location.groundId
].longitude = location.longitude
self.grounds[
self.grounds.groundId == location.groundId
].elevAngle = location.elevAngle
self.grounds[
self.grounds.groundId == location.groundId
].operational = location.operational
print(f"Station {location.groundId} updated at time {self.get_time()}.")
else:
self.grounds = self.grounds.append(
{
"groundId": location.groundId,
"latitude": location.latitude,
"longitude": location.longitude,
"elevAngle": location.elevAngle,
"operational": location.operational,
},
ignore_index=True,
)
print(f"Station {location.groundId} registered at time {self.get_time()}.")
Position Publisher Class
The next class in the Satellites application is the Position Publisher. This class takes the satellite location information from the Constellation class and publishes it over the NOS-T infrastructre. These messages are used for the Scoreboard application, which is a geospatial visualization tool.
# define a publisher to report satellite status
class PositionPublisher(WallclockTimeIntervalPublisher):
"""
*This object class inherits properties from the WallclockTimeIntervalPublisher object class from the publisher template in the NOS-T tools library*
The user can optionally specify the wallclock :obj:`timedelta` between message publications and the scenario :obj:`datetime` when the first of these messages should be published.
Args:
app (:obj:`ManagedApplication`): An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions
constellation (:obj:`Constellation`): Constellation :obj:`Entity` object class
time_status_step (:obj:`timedelta`): Optional duration between time status 'heartbeat' messages
time_status_init (:obj:`datetime`): Optional scenario :obj:`datetime` for publishing the first time status 'heartbeat' message
"""
def __init__(
self, app, constellation, time_status_step=None, time_status_init=None
):
super().__init__(app, time_status_step, time_status_init)
self.constellation = constellation
self.isInRange = [
False for i, satellite in enumerate(self.constellation.satellites)
]
def publish_message(self):
"""
*Abstract publish_message method inherited from the WallclockTimeIntervalPublisher object class from the publisher template in the NOS-T tools library*
This method sends a :obj:`SatelliteStatus` message to the *PREFIX/constellation/location* topic for each satellite in the constellation (:obj:`Constellation`).
"""
for i, satellite in enumerate(self.constellation.satellites):
next_time = constellation.ts.from_datetime(
constellation.get_time() + 60 * self.time_status_step
)
satSpaceTime = satellite.at(next_time)
subpoint = wgs84.subpoint(satSpaceTime)
sensorRadius = compute_sensor_radius(
subpoint.elevation.m, constellation.min_elevations_fire[i]
)
self.isInRange[i], groundId = check_in_range(
next_time, satellite, constellation.grounds
)
self.app.send_message(
"location",
SatelliteStatus(
id=i,
name=satellite.name,
latitude=subpoint.latitude.degrees,
longitude=subpoint.longitude.degrees,
altitude=subpoint.elevation.m,
radius=sensorRadius,
commRange=self.isInRange[i],
time=constellation.get_time(),
).json(),
)
# define an observer to send fire detection events
class FireDetectedObserver(Observer):
"""
*This object class inherits properties from the Observer object class from the observer template in the NOS-T tools library*
Args:
app (:obj:`ManagedApplication`): An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions
Fire Observer Classes
The next code block contains two different fire observation classes. The first of these is for detecting fires and the second is for reporting fires. The concept of operations for FireSat+ is that fires are first ignited, then detected when a satellite passes over them. Finally, the fires are reported when the detecting satellite is in range of a ground station for the data downlink. The Fire Observer classes publish this over the testbed for postprocessing of results, and for Scoreboard visualization.
def __init__(self, app):
self.app = app
def on_change(self, source, property_name, old_value, new_value):
"""
*Standard on_change callback function format inherited from Observer object class in NOS-T tools library*
In this instance, the callback function checks for notification of the "detected" property and publishes :obj:`FireDetected` message to *PREFIX/constellation/detected* topic:
"""
if property_name == Constellation.PROPERTY_FIRE_DETECTED:
self.app.send_message(
"detected",
FireDetected(
fireId=new_value["fireId"],
detected=new_value["detected"],
detected_by=new_value["detected_by"],
).json(),
)
# define an observer to send fire reporting events
class FireReportedObserver(Observer):
"""
*This object class inherits properties from the Observer object class from the observer template in the NOS-T tools library*
Args:
app (:obj:`ManagedApplication`): An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions
"""
def __init__(self, app):
self.app = app
def on_change(self, source, property_name, old_value, new_value):
"""
*Standard on_change callback function format inherited from Observer object class in NOS-T tools library*
In this instance, the callback function checks for notification of the "reported" property and publishes :obj:`FireReported` message to *PREFIX/constellation/reported* topic:
"""
if property_name == Constellation.PROPERTY_FIRE_REPORTED:
self.app.send_message(
"reported",
FireReported(
fireId=new_value["fireId"],
reported=new_value["reported"],
reported_by=new_value["reported_by"],
reported_to=new_value["reported_to"],
).json(),
)
# name guard used to ensure script only executes if it is run as the __main__
if __name__ == "__main__":
# Note that these are loaded from a .env file in current working directory
credentials = dotenv_values(".env")
HOST, PORT = credentials["HOST"], int(credentials["PORT"])
USERNAME, PASSWORD = credentials["USERNAME"], credentials["PASSWORD"]
# set the client credentials
The final block of code in the Satellites app is for initializing data and adding the functions and classes.
# create the managed application
app = ManagedApplication(NAME)
# load current TLEs for active satellites from Celestrak (NOTE: User has option to specify their own TLE instead)
activesats_url = "https://celestrak.com/NORAD/elements/active.txt"
activesats = load.tle_file(activesats_url, reload=True)
by_name = {sat.name: sat for sat in activesats}
names = ["AQUA", "TERRA", "SUOMI NPP", "NOAA 20", "SENTINEL-2A", "SENTINEL-2B"]
ES = []
indices = []
for name_i, name in enumerate(names):
ES.append(by_name[name])
indices.append(name_i)
# initialize the Constellation object class (in this example from EarthSatellite type)
constellation = Constellation("constellation", app, indices, names, ES)
# add observer classes to the Constellation object class
constellation.add_observer(FireDetectedObserver(app))
constellation.add_observer(FireReportedObserver(app))
# add the Constellation entity to the application's simulator
app.simulator.add_entity(constellation)
# add a shutdown observer to shut down after a single test case
app.simulator.add_observer(ShutDownObserver(app))
# add a position publisher to update satellite state every 5 seconds of wallclock time
app.simulator.add_observer(
PositionPublisher(app, constellation, timedelta(seconds=1))
)
# start up the application on PREFIX, publish time status every 10 seconds of wallclock time
app.start_up(
PREFIX,
config,
True,
time_status_step=timedelta(seconds=10) * SCALE,
time_status_init=datetime(2020, 1, 1, 7, 20, tzinfo=timezone.utc),
time_step=timedelta(seconds=1) * SCALE,
)
# add message callbacks
app.add_message_callback("fire", "location", constellation.on_fire)
app.add_message_callback("ground", "location", constellation.on_ground)
while True:
pass
The Manager application - main_manager.py
Maintaining a consistent simulation clock is important for many NOS-T use cases. For test suites that need to run faster than real time, it is an absolute necessity. The NOS-T Manager application is a good way to orchestrate all of the pieces for these types of tests. The manager is included in the NOS-T Tools library and will ensure that compliant applications start at the same time, and use a consistent simulation clock throughout the test run.
As above with the Satellites application, you should create a blank main_manager.py file in the examples/firesat_tutorial/manager folder. NOTE: You must maintain the indentations you see in these code blocks when pasting them into the main_manager.py file.
Next, we will go through the Manager code block-by-block to understand what it is doing. First, we have all of the import statements that the Manager relies on. The first of these three are general Python dependencies, and the second two are drawn from the NOS-T tools library. The last imports come from a config file that you should adjust for any specific test suites. In that config file you will need to set your desired event message prefix, the time scale, and any time scale updates.
import logging
from datetime import datetime, timedelta, timezone
from dotenv import dotenv_values
from nost_tools.application_utils import ConnectionConfig, ShutDownObserver
from nost_tools.manager import Manager
# The test suite event prefix, time scale, and any updated time scales go in the config.py file
from manager_config_files.config import (
PREFIX,
SCALE,
UPDATE,
)
logging.basicConfig(level=logging.INFO)
The time scale is a simple multiplier, i.e. if SCALE = 60 then the time will be sped up by 60x – meaning that each second of real time
will be one minute of simulation time. The time scale updates are used when you want to change the time scale at any point during
the simulation. As for the updates, They take a form like this:
UPDATE = [TimeScaleUpdate(120.0, datetime(2020, 1, 1, 8, 20, 0, tzinfo=timezone.utc))]
The above command would change the time scale to 120x at the given datetime object in simulation time. If you do not wish to update the time scale during a test case, then you can set
UPDATE = []
Finally, the last line in the above code block sets up a logger to help you track what is going on. More info on the various levels can be found here.
The next block of code starts with a name guard and credentials like the Satellites app above. These credentials will be drawn from an environment file described below. The next four lines of code follow their preceding comments. Using the various NOS-T tools from the library the connection is set, the manager application is created, it is set to shut down after the test case, and is commanded to start up.
if __name__ == "__main__":
# Note that these are loaded from a .env file in current working directory
credentials = dotenv_values(".env")
HOST, PORT = credentials["HOST"], int(credentials["PORT"])
USERNAME, PASSWORD = credentials["USERNAME"], credentials["PASSWORD"]
# set the client credentials from the config file
config = ConnectionConfig(USERNAME, PASSWORD, HOST, PORT, True)
# create the manager application from the template in the tools library
manager = Manager()
# add a shutdown observer to shut down after a single test case
manager.simulator.add_observer(ShutDownObserver(manager))
# start up the manager on PREFIX from config file
manager.start_up(PREFIX, config, True)
This final section of code contains the vital information for executing your test plan. The comments on the right side give a good explanation of
what each line means. It is important to note that the SCALE and UPDATE values should be set in the config file as explained
above.
manager.execute_test_plan(
datetime(2020, 1, 1, 7, 20, 0, tzinfo=timezone.utc), # scenario start datetime
datetime(2020, 1, 1, 10, 20, 0, tzinfo=timezone.utc), # scenario stop datetime
start_time=None, # optionally specify a wallclock start datetime for synchronization
time_step=timedelta(seconds=1), # wallclock time resolution for simulation
time_scale_factor=SCALE, # initial scale between wallclock and scenario clock (e.g. if SCALE = 60.0 then 1 wallclock second = 1 scenario minute)
time_scale_updates=UPDATE, # optionally schedule changes to the time_scale_factor at a specified scenario time
time_status_step=timedelta(seconds=1)* SCALE, # optional duration between time status 'heartbeat' messages
time_status_init=datetime(2020, 1, 1, 7, 21, 0, tzinfo=timezone.utc), # optional initial scenario datetime to start publishing time status 'heartbeat' messages
command_lead=timedelta(seconds=5), # lead time before a scheduled update or stop command
)
Test Suite Wrap-Up
Next, we’ll go through the next steps to actually executing FireSat+.
File Tree Checkup
If you have done everything correctly up to this point, you should see a file tree like the image below. Most importantly, you should have the five folders in the firesat folder which contain the constituent FireSat+ applications. These applications are described in the next section.
Remaining Applications
There are a total of five files you will need to run for FireSat+, four user applications, the NOS-T manager application, and the Scoreboard, a geospatial data visualization tool. Managing an NOS-T Test Run
Executing the FireSat+ Test Suite
There are a few more steps necessary to run FireSat+. You need to create a Cesium token to run the Scoreboard and set up environment files for each application.
Cesium Access Token and Assets
The FireSat+ Scoreboard application uses the Cesium geospatial visualization tool which requires getting an access token and an 3D Earth map asset. You will get an access token by signing in at the following link:
https://cesium.com/ion/signin/tokens
After creating an account, you must add the Asset “Blue Marble Next Generation July, 2004” from the Asset Depot (ID 3845) to your account assets to enable visualization.
Setting Up Environment Files
In order to protect your (and our) information, these applications all use environment files for usernames, passwords, event broker host site URLs, and port numbers. You will need to create an environment file in the each FireSat+ folders.
Note that you can use most text editors to make these files but be sure that you are not saving them as a .txt file type. For instance, if you save a .evn file as a .txt file type using Windows Notepad, it will actually save as .evn.txt which will not work. If you’re using Windows Notepad choose the file type “All Files (.)”.
For the applications coded in python (.py files) you will need to create a text file with the name “.env” containing the following text:
HOST="your event broker host URL"
PORT=8883 - your connection port
USERNAME="your event broker username"
PASSWORD="your event broker password"
The Scoreboard application is in .html, and pulls in credentials from a JavaScript file. To do this create a text file with the name “env.js” containing the following information:
var HOST="your event broker host URL"
var PORT=8883 - your connection port
var USERNAME="your event broker username"
var PASSWORD="your event broker password"
var TOKEN="your Cesium token (see Cesium installation instructions)"
Executing FireSat+
Finally, you need to run the five applications together in order to execute the FireSat+ test suite. These applications need to be logically separated when running. For the python scripts, this can be done by running them on separate computers, by using separate consoles in Spyder, or separate terminals with VSCode. The Scoreboard is an .html file and can be run in a web browser, double-clicking the file should work. Each folder in the FireSat+ test suite has a code you need to run, they are:
main_fire.py - The Fires app publishes historical fire data.
main_ground.py - The Ground app models a ground station in Svalbard, Norway.
main_constellation.py - The Satellites app models the constellation of spacecraft observing and reporting the fires.
scoreboard.html - The aforementioned Scoreboard gives a view of what’s happening during a test run.
main_manager.py - The NOS-T Manager app orchestrates each test run by starting the other apps at the same time, maintaining a consistent time throughout, and shutting down the apps at the end.
You must run the main_manager.py application last, otherwise it does not matter in which order you start the other applications. All of the .py applications will give an output that they are waiting for the test case to start up.
If everything is running correctly, the Scoreboard app should show an image similar to below.
Conclusion
This hands-on tutorial was developed to help users get started with NOS-T from a basic level. It begins with downloading an IDE for running scripts to interface with NOS-T and finishes with executing the FireSat+ example code. Some good next steps for learning other NOS-T functions and developing your own test suites can be found at the following links: