Constellations

This application demonstrates a constellation of satellites for monitoring fires propagated from Two-Line Elements (TLEs).

The application contains one Constellation (Entity) object class, one PositionPublisher (WallclockTimeIntervalPublisher) class, and two Observer object classes to monitor for FireDetected and FireReported events, respectively. The application also adds several methods outside of these classes containing standardized calculations sourced from Ch. 5 of Space Mission Analysis and Design by Wertz and Larson. The example also includes a startup script that demonstrates how to add each of these objects to the simulator before executing the test case, as well as how to map callback functions to the newly started client.

Methods

main_constellation.compute_min_elevation(field_of_regard)

Computes the minimum elevation angle required for a satellite to observe a point from current location.

Parameters:

altitude (float) – Altitude (meters) above surface of the observation
field_of_regard (float) – Angular width (degrees) of observation

Returns:

min_elevation: The minimum elevation angle (degrees) for observation

Return type:

float

main_constellation.compute_sensor_radius(min_elevation)

Computes the sensor radius for a satellite at current altitude given minimum elevation constraints.

Parameters:

altitude (float) – Altitude (meters) above surface of the observation
min_elevation (float) – Minimum angle (degrees) with horizon for visibility

Returns:

sensor_radius: The radius (meters) of the nadir pointing sensors circular view of observation

Return type:

float

main_constellation.get_elevation_angle(sat, loc)

Returns the elevation angle (degrees) of satellite with respect to the topocentric horizon.

Parameters:

t (Time) – Time object of skyfield.timelib module
sat (EarthSatellite) – Skyview EarthSatellite object from skyfield.sgp4lib module
loc (GeographicPosition) – Geographic location on surface specified by latitude-longitude from skyfield.toposlib module

Returns:

alt.degrees: Elevation angle (degrees) of satellite with respect to the topocentric horizon

Return type:

float

main_constellation.check_in_view(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.

Parameters:

t (Time) – Time object of skyfield.timelib module
satellite (EarthSatellite) – Skyview EarthSatellite object from skyfield.sgp4lib module
topos (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:

isInView: True/False indicating visibility of ground location to satellite

Return type:

bool

main_constellation.check_in_range(satellite, grounds)

Checks if the satellite is in range of any of the operational ground stations.

Parameters:

t (Time) – Time object of skyfield.timelib module
satellite (EarthSatellite) – Skyview EarthSatellite object from skyfield.sgp4lib module
grounds (DataFrame) – Dataframe of ground station locations, minimum elevation angles for communication, and operational status (T/F)

Returns:

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)

Return type:

bool, int

Classes

While the latter methods are globally defined, the following classes have built-in methods that are unique to each object. The Constellation class tracks state transitions for the satellites in the constellation, while the PositionPublisher standardizes the message structure and frequency of published messages updating these states. The FireDetectedObserver and FireReportedObserver objects demonstrate using the Observer class from the NOS-T Tools Library as a triggered message in an event-driven architecture. In this case, messages are triggered by transitions in states of each fire.

Constellation

class Constellation(cName, app, id, names, ES=None, tles=None)

Bases: Entity

This object class inherits properties from the Entity object class in the NOS-T tools library

Parameters:

cName (str) – A string containing the name for the constellation application
app (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)
ES (list) – Optional list of 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)
tles (list) – Optional list of Two-Line Element str to be converted into EarthSatellite objects and included in the simulation

fires

List of fires with unique fireId (int), ignition (datetime), and latitude-longitude location (GeographicPosition) - NOTE: initialized as [ ]

Type:: list

grounds

Dataframe containing information about ground stations with unique groundId (int), latitude-longitude location (GeographicPosition), min_elevation (float) angle constraints, and operational status (bool) - NOTE: initialized as None

Type:: DataFrame

satellites

List of EarthSatellite objects included in the constellation - NOTE: must be same length as id

Type:: list

detect

List of detected fires with unique fireId (int), detected datetime, and name (str) of detecting satellite - NOTE: initialized as [ ]

Type:: list

report

List of reported fires with unique fireId (int), reported datetime, name (str) of reporting satellite, and groundId (int) of ground station reported to - NOTE: initialized as [ ]

Type:: list

positions

List of current latitude-longitude-altitude locations (GeographicPosition) of each satellite in the constellation - NOTE: must be same length as id

Type:: list

next_positions

List of next latitude-longitude-altitude locations (GeographicPosition) of each satellite in the constellation - NOTE: must be same length as id

Type:: list

min_elevations_fire

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

Type:: list

Constellation.initialize(init_time)

Activates the Constellation at a specified initial scenario time

Parameters:: init_time (datetime) – Initial scenario time for simulating propagation of satellites

Constellation.tick(time_step)

Computes the next Constellation state after the specified scenario duration and the next simulation scenario time

Parameters:: time_step (timedelta) – Duration between current and next simulation scenario time

Constellation.tock(): Commits the next Constellation state and advances simulation scenario time

Constellation.on_fire(ch, method, properties, body)

Callback function appends a dictionary of information for a new fire to fires list when message detected on the PREFIX.fires.location topic

Parameters:

ch (Channel) – Channel object used to communicate with the event broker
method (Method) – Method object containing information about the message delivery, such as delivery tag and exchange
properties (Properties) – Properties object containing metadata about the message, such as content type and message ID
body (bytes) – Contains the message content as a byte string, which is decoded to a UTF-8 string and then parsed into a FireStarted object

Constellation.on_ground(ch, method, properties, body)

Callback function appends a dictionary of information for a new ground station to grounds list when message detected on the PREFIX/ground/location topic. Ground station information is published at beginning of simulation, and the list is converted to a DataFrame when the Constellation is initialized.

Parameters:

ch (Channel) – Channel object used to communicate with the event broker
method (Method) – Method object containing information about the message delivery, such as delivery tag and exchange
properties (Properties) – Properties object containing metadata about the message, such as content type and message ID
body (bytes) – Contains the message content as a byte string, which is decoded to a UTF-8 string and then parsed into a FireStarted object

PositionPublisher

class PositionPublisher(app, constellation, time_status_step=None, time_status_init=None)

Bases: 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 timedelta between message publications and the scenario datetime when the first of these messages should be published.

Parameters:

app (ManagedApplication) – An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions
constellation (Constellation) – Constellation Entity object class
time_status_step (timedelta) – Optional duration between time status ‘heartbeat’ messages
time_status_init (datetime) – Optional scenario datetime for publishing the first time status ‘heartbeat’ message

PositionPublisher.publish_message()

Abstract publish_message method inherited from the WallclockTimeIntervalPublisher object class from the publisher template in the NOS-T tools library

This method sends a SatelliteStatus message to the PREFIX/constellation/location topic for each satellite in the constellation (Constellation).

FireDetectedObserver

class FireDetectedObserver(app)

Bases: Observer

This object class inherits properties from the Observer object class from the observer template in the NOS-T tools library

Parameters:: app (ManagedApplication) – An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions

        if property_name == Constellation.PROPERTY_FIRE_DETECTED:
            self.app.send_message(
                self.app.app_name,
                "detected",
                FireDetected(
                    fireId=new_value["fireId"],
                    detected=new_value["detected"],
                    detected_by=new_value["detected_by"],
                ).model_dump_json(),
            )

FireDetectedObserver.on_change(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 FireDetected message to PREFIX/constellation/detected topic:

FireReportedObserver

class FireReportedObserver(app)

Bases: Observer

This object class inherits properties from the Observer object class from the observer template in the NOS-T tools library

Parameters:: app (ManagedApplication) – An application containing a test-run namespace, a name and description for the app, client credentials, and simulation timing instructions

        if property_name == Constellation.PROPERTY_FIRE_REPORTED:
            self.app.send_message(
                self.app.app_name,
                "reported",
                FireReported(
                    fireId=new_value["fireId"],
                    reported=new_value["reported"],
                    reported_by=new_value["reported_by"],
                    reported_to=new_value["reported_to"],
                ).model_dump_json(),
            )

FireReportedObserver.on_change(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 FireReported message to PREFIX/constellation/reported topic:

Schema

Schema are implemented using the pydantic library. The following schema define consistent message structures between this application and other observer applications:

pydantic settings SatelliteStatus

Bases: BaseModel

Message schema object class with properties inherited from the pydantic library’s BaseModel

Standardized satellite status message includes:

Show JSON schema

{
   "title": "SatelliteStatus",
   "description": "*Message schema object class with properties inherited from the pydantic library's BaseModel*\n\nStandardized satellite status message includes:",
   "type": "object",
   "properties": {
      "id": {
         "description": "Unique satellite identifier",
         "title": "Id",
         "type": "integer"
      },
      "name": {
         "description": "Satellite name for labeling.",
         "title": "Name",
         "type": "string"
      },
      "latitude": {
         "description": "Latitude (deg) of satellite subpoint location.",
         "maximum": 90,
         "minimum": -90,
         "title": "Latitude",
         "type": "number"
      },
      "longitude": {
         "description": "Longitude (deg) of satellite subpoint location.",
         "maximum": 180,
         "minimum": -180,
         "title": "Longitude",
         "type": "number"
      },
      "altitude": {
         "description": "Altitude (meters) of satellite above sea level",
         "title": "Altitude",
         "type": "number"
      },
      "radius": {
         "description": "Radius of nadir pointing cone of vision",
         "title": "Radius",
         "type": "number"
      },
      "commRange": {
         "default": false,
         "description": "Boolean for if satellite is in ground stations view",
         "title": "Commrange",
         "type": "boolean"
      },
      "time": {
         "description": "Time in satellite reference frame",
         "format": "date-time",
         "title": "Time",
         "type": "string"
      }
   },
   "required": [
      "id",
      "name",
      "latitude",
      "longitude",
      "altitude",
      "radius",
      "time"
   ]
}

Fields:

id (int)
name (str)
latitude (float)
longitude (float)
altitude (float)
radius (float)
commRange (bool)
time (datetime.datetime)

field id: int [Required]: Unique satellite identifier

field name: str [Required]: Satellite name for labeling.

field latitude: float [Required]

Latitude (deg) of satellite subpoint location.

Constraints:

ge = -90
le = 90

field longitude: float [Required]

Longitude (deg) of satellite subpoint location.

Constraints:

ge = -180
le = 180

field altitude: float [Required]: Altitude (meters) of satellite above sea level

field radius: float [Required]: Radius of nadir pointing cone of vision

field commRange: bool = False: Boolean for if satellite is in ground stations view

field time: datetime [Required]: Time in satellite reference frame

pydantic settings GroundLocation

Bases: BaseModel

Message schema object class with properties inherited from the pydantic library’s BaseModel

Standardized message for ground station information includes:

Show JSON schema

{
   "title": "GroundLocation",
   "description": "*Message schema object class with properties inherited from the pydantic library's BaseModel*\n\nStandardized message for ground station information includes:",
   "type": "object",
   "properties": {
      "groundId": {
         "description": "Unique ground station identifier.",
         "title": "Groundid",
         "type": "integer"
      },
      "latitude": {
         "description": "Latitude (deg) of ground station.",
         "maximum": 90,
         "minimum": -90,
         "title": "Latitude",
         "type": "number"
      },
      "longitude": {
         "description": "Longitude (deg) of ground station.",
         "maximum": 180,
         "minimum": -180,
         "title": "Longitude",
         "type": "number"
      },
      "elevAngle": {
         "description": "Minimum elevation angle (deg) for satellite-ground communications",
         "title": "Elevangle",
         "type": "number"
      },
      "operational": {
         "default": true,
         "description": "True, if this ground station is operational.",
         "title": "Operational",
         "type": "boolean"
      }
   },
   "required": [
      "groundId",
      "latitude",
      "longitude",
      "elevAngle"
   ]
}

Fields:

groundId (int)
latitude (float)
longitude (float)
elevAngle (float)
operational (bool)

field groundId: int [Required]: Unique ground station identifier.

field latitude: float [Required]

Latitude (deg) of ground station.

Constraints:

ge = -90
le = 90

field longitude: float [Required]

Longitude (deg) of ground station.

Constraints:

ge = -180
le = 180

field elevAngle: float [Required]: Minimum elevation angle (deg) for satellite-ground communications

field operational: bool = True: True, if this ground station is operational.

pydantic settings FireStarted

Bases: BaseModel

Message schema object class with properties inherited from the pydantic library’s BaseModel

Standardized message for fire ignition includes:

Show JSON schema

{
   "title": "FireStarted",
   "description": "*Message schema object class with properties inherited from the pydantic library's BaseModel*\n\nStandardized message for fire ignition includes:",
   "type": "object",
   "properties": {
      "fireId": {
         "description": "Unique fire identifier.",
         "title": "Fireid",
         "type": "integer"
      },
      "start": {
         "anyOf": [
            {
               "format": "date-time",
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "description": "Time fire started.",
         "title": "Start"
      },
      "latitude": {
         "anyOf": [
            {
               "maximum": 90,
               "minimum": -90,
               "type": "number"
            },
            {
               "type": "null"
            }
         ],
         "description": "Latitude (deg) of fire location.",
         "title": "Latitude"
      },
      "longitude": {
         "anyOf": [
            {
               "maximum": 180,
               "minimum": -180,
               "type": "number"
            },
            {
               "type": "null"
            }
         ],
         "description": "Longitude (deg) of fire location.",
         "title": "Longitude"
      }
   },
   "required": [
      "fireId",
      "start",
      "latitude",
      "longitude"
   ]
}

Fields:

fireId (int)
start (datetime.datetime | None)
latitude (float | None)
longitude (float | None)

field fireId: int [Required]: Unique fire identifier.

field start: datetime | None [Required]: Time fire started.

field latitude: float | None [Required]

Latitude (deg) of fire location.

Constraints:

ge = -90
le = 90

field longitude: float | None [Required]

Longitude (deg) of fire location.

Constraints:

ge = -180
le = 180

pydantic settings FireDetected

Bases: BaseModel

Message schema object class with properties inherited from the pydantic library’s BaseModel

Standardized message for fire detection includes:

Show JSON schema

{
   "title": "FireDetected",
   "description": "*Message schema object class with properties inherited from the pydantic library's BaseModel*\n\nStandardized message for fire detection includes:",
   "type": "object",
   "properties": {
      "fireId": {
         "description": "Unique fire identifier.",
         "title": "Fireid",
         "type": "integer"
      },
      "detected": {
         "description": "Time fire detected.",
         "format": "date-time",
         "title": "Detected",
         "type": "string"
      },
      "detected_by": {
         "description": "Satellite name that detected the fire.",
         "title": "Detected By",
         "type": "string"
      }
   },
   "required": [
      "fireId",
      "detected",
      "detected_by"
   ]
}

Fields:

fireId (int)
detected (datetime.datetime)
detected_by (str)

field fireId: int [Required]: Unique fire identifier.

field detected: datetime [Required]: Time fire detected.

field detected_by: str [Required]: Satellite name that detected the fire.

pydantic settings FireReported

Bases: BaseModel

Message schema object class with properties inherited from the pydantic library’s BaseModel

Standardized message for fire reporting includes:

Show JSON schema

{
   "title": "FireReported",
   "description": "*Message schema object class with properties inherited from the pydantic library's BaseModel*\n\nStandardized message for fire reporting includes:",
   "type": "object",
   "properties": {
      "fireId": {
         "description": "Unique fire identifier.",
         "title": "Fireid",
         "type": "integer"
      },
      "reported": {
         "description": "Time fire reported.",
         "format": "date-time",
         "title": "Reported",
         "type": "string"
      },
      "reported_by": {
         "description": "Satellite name that sent the fire report.",
         "title": "Reported By",
         "type": "string"
      },
      "reported_to": {
         "description": "Station id that received the fire report.",
         "title": "Reported To",
         "type": "integer"
      }
   },
   "required": [
      "fireId",
      "reported",
      "reported_by",
      "reported_to"
   ]
}

Fields:

fireId (int)
reported (datetime.datetime)
reported_by (str)
reported_to (int)

field fireId: int [Required]: Unique fire identifier.

field reported: datetime [Required]: Time fire reported.

field reported_by: str [Required]: Satellite name that sent the fire report.

field reported_to: int [Required]: Station id that received the fire report.

Startup Script

The following code demonstrates how the Constellation Entity object class is initialized and added to the simulator, how the application is started up, and how callback functions are assigned to the application:

    # Define application name
    NAME = "constellation"

    # Load config
    config = ConnectionConfig(yaml_file="firesat.yaml", app_name=NAME)

    # Create the managed application
    app = ManagedApplication(app_name=NAME)

    # Load current TLEs for active satellites from Celestrak
    activesats_url = (
        "https://celestrak.org/NORAD/elements/gp.php?GROUP=active&FORMAT=tle"
    )
    activesats = load.tle_file(
        activesats_url, filename="satellites/active.txt", reload=True
    )

    by_name = {sat.name: sat for sat in activesats}
    names = [
        "AQUA",
        "TERRA",
        "SUOMI NPP",
        "NOAA 20 (JPSS-1)",  # "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
    app.start_up(
        config.rc.simulation_configuration.execution_parameters.general.prefix,
        config,
    )

    # Add message callbacks
    app.add_message_callback("fire", "location", constellation.on_fire)
    app.add_message_callback("ground", "location", constellation.on_ground)

    while True:
        pass

In this example, six satellites (AQUA, TERRA, SUOMI NPP, NOAA 20, SENTINEL 2A, SENTINEL 2B) are included in the simulation. CelesTrak is queried for current active TLEs, which returns this information as list of EarthSatellite objects. A subset list of EarthSatellite objects is constructed containing the six satellites of interest. This subset is used for initializing the Constellation Entity before adding to the simulator.