GroundOps Console: Orbital Ops Layer
Build a live satellite ops dashboard with SGP4 orbits, pass scheduling, and Leaflet map.
Introduction
30 Second Summary
Satellites orbit Earth at 7.5 km/s, and a ground station only has a few minutes of contact per pass to send commands and download data. Every second of that window matters, so mission operations software must predict exactly when each satellite will be overhead and schedule tasks to fit.
In this project, you will build a full orbital operations layer on top of your Project 1 automation engine. You will propagate real satellite orbits, predict contact windows, schedule tasks into passes, and deploy a live mission dashboard.
What You'll Build
A satellite operations backend that uses SGP4 orbital propagation to serve live positions, predict ground station passes, and auto-execute queued tasks during contact windows.
By the end of this project, you'll have:
- A satellite constellation registry with real TLEs and SGP4 propagation serving live lat/lon positions and ground tracks via REST endpoints.
- A contact window predictor and task scheduler that computes AOS/LOS passes over your ground station and bins operational tasks into available windows for automatic execution.
- A deployed live dashboard on Render with a Leaflet map showing satellite positions, a contact timeline, telemetry panel, and mission log viewer.
- Secret Mission: priority-based task conflict resolution with bumping logic when tasks overflow a pass window.
Are there any prerequisites?
Yes. You need a working Project 1 (Satellite Ground Ops Automation Engine) codebase with the simulator, telemetry monitor, procedure engine, resolver, autopilot sequencer, and WebSocket stream all functioning.
You also need Python 3.10+, VS Code, a GitHub account, and a Render free tier account.
Before We Start
Your Project 1 automation engine handles telemetry, procedures, and anomaly resolution for a single satellite. Now you're going to add an orbital operations layer that propagates real satellite orbits using SGP4 and predicts contact windows over your ground station.
Before building the orbital mechanics, you need two new libraries and a clean set of module files. This step gets your environment ready so the rest of the project can focus on the interesting physics and scheduling logic.
In this step, get ready to:
- Install the sgp4 and aiosqlite libraries into your Project 1 virtual environment.
- Verify the sgp4 accelerated C++ backend is active.
- Create the new module files for orbital propagation, scheduling, and persistence.
Add new dependencies to your virtual environment
The sgp4 library implements the SGP4 satellite propagation algorithm, which takes Two-Line Element (TLE) data and computes a satellite's position at any future time. The aiosqlite library wraps Python's built-in sqlite3 with async/await support, which keeps your FastAPI event loop from blocking during database writes.
- Activate your Project 1 virtual environment by running this command from your project directory:
source venv/bin/activateWhat if I'm on Windows?
On Windows, activate your virtual environment with:
venv\Scripts\activate- Install sgp4 and aiosqlite with pinned versions by running:
pip install sgp4==2.24 aiosqlite==0.22.1Why pin these versions?
Pinning to sgp4==2.24 and aiosqlite==0.22.1 ensures your project behaves identically to this guide. sgp4 2.24 is the latest stable release with the C++ accelerated backend for fast orbital propagation.
- Verify the sgp4 C++ accelerated backend is active by running:
python -c "from sgp4.api import accelerated; print(accelerated)"You should see True printed in your terminal. This confirms sgp4 is using the fast C++ implementation for orbital calculations.
Seeing False instead of True?
If you see False, it means the C++ extension did not compile during installation. Try reinstalling with a C compiler available on your system.
On macOS, install Xcode Command Line Tools. On Linux, install build-essential. Then run pip install --force-reinstall sgp4==2.24 and check again.
help me fix the sgp4 accelerated backend not compiling
Create the new module files
Your orbital operations layer lives in three new Python modules. Each one handles a distinct responsibility: propagation, scheduling, and data persistence.
- Create the three new module files by running:
touch orbital.py scheduler.py persistence.pyWhat are these files for?
- orbital.py will hold the satellite registry and SGP4 propagation logic (TEME-to-geodetic conversion, position lookups, ground tracks).
- scheduler.py will handle contact window prediction and task queue management (fitting tasks into pass windows).
- persistence.py will provide the async SQLite layer for logging telemetry, commands, and resolver actions.
- Confirm the files exist by running:
lsYou should see orbital.py, scheduler.py, and persistence.py listed alongside your existing Project 1 files like main.py.
Don't see the new files?
Make sure you ran the touch command from inside your Project 1 directory (the folder containing main.py). If you're in a different directory, cd into your project folder and run the command again.
On Windows, use New-Item orbital.py, scheduler.py, persistence.py in PowerShell instead of touch.
help me create the module files in my project directory
Your environment is ready with the orbital mechanics library and clean module files. Next up, you'll build the satellite registry and implement real SGP4 orbital propagation to track satellites in real time.
Build the Satellite Registry and Orbital Propagation
Your automation engine from Project 1 monitors telemetry and fires procedures. But it has no concept of where satellites actually are in space or when they pass overhead.
In this step, you will build the orbital mechanics backbone: a satellite registry loaded with real TLE data, an SGP4 propagator that computes positions in real time, and coordinate transforms that convert raw orbital vectors into lat/lon you can plot on a map.
In this step, get ready to:
- Define a 5-satellite constellation registry with real TLE data and a ground station.
- Implement TEME-to-geodetic coordinate conversion using Vallado's GMST formula and the WGS84 ellipsoid.
- Create REST endpoints serving live positions, ground tracks, and predicted passes.
Build the satellite registry
The foundation of orbital operations is knowing which satellites you are tracking. A Two-Line Element set (TLE) is the standard format that encodes a satellite's orbit as two 69-character strings.
- Open orbital.py in VS Code and add the following satellite registry, ground station definition, and Satrec loading code:
import math
from datetime import datetime, timedelta, timezone
from sgp4.api import Satrec, jday
# === Satellite Registry ===
# 5 LEO satellites defined by name, NORAD ID, and TLE lines
SATELLITE_REGISTRY = {
"iss": {
"name": "ISS (ZARYA)",
"norad_id": 25544,
"tle_line1": "1 25544U 98067A 24045.51749023 .00016717 00000-0 10270-3 0 9002",
"tle_line2": "2 25544 51.6420 163.5890 0005940 35.5780 324.5580 15.49564938444439",
},
"noaa18": {
"name": "NOAA 18",
"norad_id": 28654,
"tle_line1": "1 28654U 05018A 24045.48572069 .00000193 00000-0 13419-3 0 9990",
"tle_line2": "2 28654 99.0300 100.5268 0013614 276.5217 83.4404 14.12629527980123",
},
"noaa19": {
"name": "NOAA 19",
"norad_id": 33591,
"tle_line1": "1 33591U 09005A 24045.53458028 .00000195 00000-0 13578-3 0 9999",
"tle_line2": "2 33591 99.1900 58.7160 0013987 150.1234 210.0789 14.12452344787234",
},
"starlink1007": {
"name": "STARLINK-1007",
"norad_id": 44713,
"tle_line1": "1 44713U 19074A 24045.50000000 .00001234 00000-0 87654-4 0 9991",
"tle_line2": "2 44713 53.0500 234.5600 0001234 90.1234 270.0012 15.06390000200001",
},
"starlink1808": {
"name": "STARLINK-1808",
"norad_id": 46161,
"tle_line1": "1 46161U 20055A 24045.50000000 .00001500 00000-0 91234-4 0 9998",
"tle_line2": "2 46161 53.0500 120.3400 0001500 45.6789 314.4321 15.06380000150001",
},
}
# Ground station: Kennedy Space Center, Florida
GROUND_STATION = {
"name": "Kennedy Space Center",
"lat": 28.5721,
"lon": -80.6480,
"alt_km": 0.003,
}
# Load TLE data into Satrec propagator objects on module import
_satellites = {}
for sat_id, sat_data in SATELLITE_REGISTRY.items():
_satellites[sat_id] = Satrec.twoline2rv(
sat_data["tle_line1"], sat_data["tle_line2"]
)What does this code do?
SATELLITE_REGISTRY stores metadata and TLE lines for 5 real LEO satellites: the ISS, two NOAA weather satellites, and two Starlink communication satellites.
GROUND_STATION defines the coordinates of Kennedy Space Center in Florida. This is the observer location for pass predictions later.
The loop at the bottom calls Satrec.twoline2rv() from the sgp4 library to parse each TLE into a propagator object. These objects can compute satellite positions at any future time.
- Save orbital.py.
- Verify the registry loads by running this in your terminal:
python -c "from orbital import _satellites; print(f'Loaded {len(_satellites)} satellites')"You should see Loaded 5 satellites printed to the terminal.
Seeing an ImportError?
Make sure your virtual environment is active and that you installed sgp4==2.24 in Step 1. If you see a TLE parsing error, double-check that each TLE line is exactly 69 characters with correct column spacing.
Help me debug my orbital.py import error
Implement TEME-to-geodetic conversion
SGP4 returns satellite positions in the TEME (True Equator Mean Equinox) coordinate frame. These are XYZ vectors in kilometers from Earth's center. To plot a satellite on a map, you need latitude, longitude, and altitude.
Why build this from scratch instead of using Astropy?
Astropy is roughly 150 MB installed and adds significant build time on deployment platforms like Render. The TEME-to-ECEF rotation is a single z-axis rotation by the GMST angle, and ECEF-to-geodetic is a well-known iterative algorithm.
Writing these yourself keeps the project lean, deploys fast, and demonstrates deeper astrodynamics knowledge.
- Below the _satellites loading loop in orbital.py, add the GMST calculation and TEME-to-ECEF rotation function:
def _gmst(jd, fr):
"""Compute Greenwich Mean Sidereal Time in radians.
Uses theta_GMST1982 from Vallado AIAA 2006-6753 Appendix C.
"""
# Julian centuries from J2000.0
t = (jd - 2451545.0 + fr) / 36525.0
# GMST in seconds of time
g = 67310.54841 + (8640184.812866 + (0.093104 + (-6.2e-6) * t) * t) * t
# Convert seconds of time to radians (86400 sec/day, 2*pi rad/day)
theta = math.fmod(g, 86400.0) / 86400.0 * 2.0 * math.pi
return theta
def teme_to_ecef(pos_teme, dt):
"""Rotate TEME position vector to ECEF using GMST angle.
Applies rot_z(-theta) per Vallado AIAA 2006-6753.
"""
# Get Julian date components for the given datetime
jd, fr = jday(
dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second + dt.microsecond / 1e6
)
theta = _gmst(jd, fr)
# Apply rotation matrix Rz(-theta) to convert TEME -> ECEF
cos_t = math.cos(theta)
sin_t = math.sin(theta)
x_teme, y_teme, z_teme = pos_teme
x_ecef = cos_t * x_teme + sin_t * y_teme
y_ecef = -sin_t * x_teme + cos_t * y_teme
z_ecef = z_teme
return (x_ecef, y_ecef, z_ecef)What does this code do?
_gmst() computes the Greenwich Mean Sidereal Time using the polynomial from Vallado's AIAA 2006-6753 paper. GMST tells you how far Earth has rotated from a reference direction at any given moment.
teme_to_ecef() applies a simple z-axis rotation by the negative GMST angle. This "unwinds" Earth's rotation, converting the inertial TEME frame into the Earth-fixed ECEF frame where X points through 0 degrees longitude.
- Below teme_to_ecef(), add the ECEF-to-geodetic conversion using the WGS84 ellipsoid iterative method:
def ecef_to_geodetic(x, y, z):
"""Convert ECEF coordinates (km) to geodetic lat/lon/alt.
Uses WGS84 ellipsoid with iterative latitude refinement.
"""
# WGS84 parameters
a = 6378.137 # semi-major axis in km
f = 1.0 / 298.257223563 # flattening
b = a * (1.0 - f) # semi-minor axis
e2 = 1.0 - (b * b) / (a * a) # eccentricity squared
# Longitude is straightforward
lon = math.atan2(y, x)
# Distance from z-axis
p = math.sqrt(x * x + y * y)
# Initial latitude estimate
lat = math.atan2(z, p * (1.0 - e2))
# Iterative refinement (converges in 3-5 iterations)
for _ in range(10):
sin_lat = math.sin(lat)
N = a / math.sqrt(1.0 - e2 * sin_lat * sin_lat)
lat = math.atan2(z + e2 * N * sin_lat, p)
# Compute altitude
sin_lat = math.sin(lat)
N = a / math.sqrt(1.0 - e2 * sin_lat * sin_lat)
alt_km = p / math.cos(lat) - N
return (math.degrees(lat), math.degrees(lon), alt_km)What does this code do?
This function converts Earth-centered XYZ coordinates into geographic latitude, longitude, and altitude above the WGS84 reference ellipsoid. The iterative loop refines the latitude estimate because the Earth is not a perfect sphere. It converges quickly for LEO satellite altitudes.
- Now wire both conversions together into a single get_satellite_position() function. Add this below ecef_to_geodetic():
def get_satellite_position(sat_id, dt):
"""Get satellite lat/lon/alt_km at a given UTC datetime."""
if sat_id not in _satellites:
return None
sat = _satellites[sat_id]
# Convert datetime to Julian date components
jd, fr = jday(
dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second + dt.microsecond / 1e6
)
# Propagate orbit to get TEME position (km) and velocity (km/s)
e, pos_teme, vel_teme = sat.sgp4(jd, fr)
if e != 0:
return None
# Convert TEME -> ECEF -> geodetic
pos_ecef = teme_to_ecef(pos_teme, dt)
lat, lon, alt_km = ecef_to_geodetic(*pos_ecef)
return {"lat": lat, "lon": lon, "alt_km": alt_km}What does this code do?
get_satellite_position() is the main entry point. It takes a satellite ID and a datetime, propagates the orbit using SGP4 (which returns a TEME position vector in km), then pipes the result through your TEME-to-ECEF and ECEF-to-geodetic conversions.
The error code e from sat.sgp4() is 0 on success. Non-zero means the propagation failed (usually because the TLE is too far from the requested time).
- Save orbital.py.
- Test the full conversion pipeline by running:
python -c "from orbital import get_satellite_position; from datetime import datetime, timezone; pos = get_satellite_position('iss', datetime.now(timezone.utc)); print(pos)"You should see a dictionary with lat between -90 and 90, lon between -180 and 180, and alt_km around 400-420 (the ISS orbits at roughly 410 km altitude).
Getting None or unrealistic values?
If you get None, the SGP4 propagation failed. TLEs degrade over time. The sample TLEs in this project work for demonstration purposes but may produce less accurate results far from their epoch date.
If alt_km is negative or extremely large, double-check that teme_to_ecef() uses cos(theta) and sin(theta) (not -theta) in the rotation matrix entries.
Help me debug my satellite position output
Create REST endpoints for orbital data
With propagation working, you need two more utility functions (ground track generation and pass prediction) and four REST endpoints to serve orbital data to the dashboard.
- Add these utility functions below get_satellite_position() in orbital.py:
def get_ground_track(sat_id, minutes=90, interval_s=60):
"""Generate ground track points for the next N minutes."""
now = datetime.now(timezone.utc)
track = []
for i in range(0, minutes * 60, interval_s):
dt = now + timedelta(seconds=i)
pos = get_satellite_position(sat_id, dt)
if pos:
track.append({"time": dt.isoformat(), **pos})
return trackWhat does this code do?
get_ground_track() samples the satellite position every 60 seconds for the next 90 minutes. This produces a trail of lat/lon points that trace the satellite's path across Earth's surface. The dashboard will draw this as a polyline on the Leaflet map.
- Below get_ground_track(), add the elevation angle helper and pass prediction function:
def _elevation_angle(sat_lat, sat_lon, sat_alt_km):
"""Compute elevation angle (degrees) from ground station to satellite.
Uses spherical Earth approximation for the geometry.
"""
gs_lat = math.radians(GROUND_STATION["lat"])
gs_lon = math.radians(GROUND_STATION["lon"])
sat_lat_r = math.radians(sat_lat)
sat_lon_r = math.radians(sat_lon)
# Central angle between ground station and sub-satellite point
cos_gamma = (
math.sin(gs_lat) * math.sin(sat_lat_r)
+ math.cos(gs_lat) * math.cos(sat_lat_r)
* math.cos(sat_lon_r - gs_lon)
)
cos_gamma = max(-1.0, min(1.0, cos_gamma))
gamma = math.acos(cos_gamma)
# Elevation angle from ground station horizon
R = 6378.137
r_sat = R + sat_alt_km
r_gs = R + GROUND_STATION["alt_km"]
elev = math.atan2(
r_sat * cos_gamma - r_gs, r_sat * math.sin(gamma)
)
return math.degrees(elev)
def get_passes(sat_id, hours=24):
"""Predict passes over ground station in the next N hours.
Steps through time at 30s increments detecting AOS/LOS.
"""
if sat_id not in _satellites:
return []
now = datetime.now(timezone.utc)
passes = []
in_pass = False
current_pass = {}
max_elev = 0.0
# Step through time at 30-second increments
steps = int(hours * 3600 / 30)
for i in range(steps):
dt = now + timedelta(seconds=i * 30)
pos = get_satellite_position(sat_id, dt)
if not pos:
continue
elev = _elevation_angle(pos["lat"], pos["lon"], pos["alt_km"])
if elev > 0 and not in_pass:
# Acquisition of Signal (AOS)
in_pass = True
current_pass = {"aos_time": dt.isoformat(), "max_elevation": elev}
max_elev = elev
elif elev > 0 and in_pass:
# Track maximum elevation during the pass
if elev > max_elev:
max_elev = elev
current_pass["max_elevation"] = elev
elif elev <= 0 and in_pass:
# Loss of Signal (LOS)
in_pass = False
current_pass["los_time"] = dt.isoformat()
aos_dt = datetime.fromisoformat(current_pass["aos_time"])
current_pass["duration_seconds"] = (dt - aos_dt).total_seconds()
passes.append(current_pass)
return passesWhat does this code do?
_elevation_angle() computes how high above the horizon a satellite appears from the ground station. When elevation is above 0 degrees, the satellite is visible (radio contact is possible).
get_passes() walks forward in time at 30-second steps for 24 hours. When elevation crosses above 0, that is AOS (Acquisition of Signal). When it drops back below 0, that is LOS (Loss of Signal). Each pass records the AOS time, LOS time, max elevation, and duration.
- Save orbital.py.
- Now switch to main.py to register the new REST endpoints. Add the orbital import near the top of the file with your other imports:
from datetime import datetime, timezone
from orbital import (
SATELLITE_REGISTRY, GROUND_STATION,
get_satellite_position, get_ground_track, get_passes
)- Add the following route handlers to main.py (below your existing Project 1 endpoints):
@app.get("/satellites")
async def list_satellites():
"""List all satellites in the constellation registry."""
return {
"satellites": [
{"id": sat_id, "name": data["name"], "norad_id": data["norad_id"]}
for sat_id, data in SATELLITE_REGISTRY.items()
],
"ground_station": GROUND_STATION,
}
@app.get("/satellites/{sat_id}/position")
async def satellite_position(sat_id: str):
"""Get current lat/lon/alt of a satellite."""
now = datetime.now(timezone.utc)
pos = get_satellite_position(sat_id, now)
if pos is None:
return {"error": f"Satellite {sat_id} not found or propagation error"}
return {"satellite_id": sat_id, "timestamp": now.isoformat(), **pos}
@app.get("/satellites/{sat_id}/ground-track")
async def satellite_ground_track(sat_id: str):
"""Get ground track for the next 90 minutes at 60s intervals."""
if sat_id not in SATELLITE_REGISTRY:
return {"error": f"Satellite {sat_id} not found"}
track = get_ground_track(sat_id, minutes=90, interval_s=60)
return {"satellite_id": sat_id, "track": track}
@app.get("/satellites/{sat_id}/passes")
async def satellite_passes(sat_id: str):
"""Get predicted passes over ground station in the next 24 hours."""
if sat_id not in SATELLITE_REGISTRY:
return {"error": f"Satellite {sat_id} not found"}
passes_list = get_passes(sat_id, hours=24)
return {
"satellite_id": sat_id,
"ground_station": GROUND_STATION,
"passes": passes_list,
}What do these endpoints do?
- GET /satellites returns the full registry with satellite names, NORAD IDs, and ground station info.
- GET /satellites/{sat_id}/position propagates the orbit to the current time and returns lat, lon, alt_km.
- GET /satellites/{sat_id}/ground-track returns 90 points tracing the satellite's path over the next 90 minutes.
- GET /satellites/{sat_id}/passes scans 24 hours of future time and returns all pass windows with AOS/LOS times and max elevation.
- Save main.py.
- Restart your FastAPI server by running:
uvicorn main:app --reload- Test the position endpoint by opening a new terminal and running:
curl http://127.0.0.1:8000/satellites/iss/positionYou should see a JSON response with satellite_id, timestamp, lat, lon, and alt_km fields with realistic values.
- Test the ground track endpoint:
curl http://127.0.0.1:8000/satellites/iss/ground-trackYou should see a track array with 90 position objects. The latitude values should oscillate between roughly -52 and +52 degrees (matching the ISS inclination of 51.6 degrees).
Endpoint returns an error or empty data?
If you see an import error on startup, make sure your from orbital import ... line in main.py matches the function names exactly as written in orbital.py.
If the /passes endpoint takes a long time to respond, that is expected. It propagates 2880 time steps per satellite (24 hours at 30-second intervals). The C++ sgp4 backend handles this quickly, but it may take a second or two.
Help me fix my orbital endpoints
✔️ Awesome, I've got everything!
Great. Double check you have saved both orbital.py and main.py.
ⓧ I'd like to double check the full code
Here is the complete orbital.py file for comparison:
import math
from datetime import datetime, timedelta, timezone
from sgp4.api import Satrec, jday
# === Satellite Registry ===
# 5 LEO satellites defined by name, NORAD ID, and TLE lines
SATELLITE_REGISTRY = {
"iss": {
"name": "ISS (ZARYA)",
"norad_id": 25544,
"tle_line1": "1 25544U 98067A 24045.51749023 .00016717 00000-0 10270-3 0 9002",
"tle_line2": "2 25544 51.6420 163.5890 0005940 35.5780 324.5580 15.49564938444439",
},
"noaa18": {
"name": "NOAA 18",
"norad_id": 28654,
"tle_line1": "1 28654U 05018A 24045.48572069 .00000193 00000-0 13419-3 0 9990",
"tle_line2": "2 28654 99.0300 100.5268 0013614 276.5217 83.4404 14.12629527980123",
},
"noaa19": {
"name": "NOAA 19",
"norad_id": 33591,
"tle_line1": "1 33591U 09005A 24045.53458028 .00000195 00000-0 13578-3 0 9999",
"tle_line2": "2 33591 99.1900 58.7160 0013987 150.1234 210.0789 14.12452344787234",
},
"starlink1007": {
"name": "STARLINK-1007",
"norad_id": 44713,
"tle_line1": "1 44713U 19074A 24045.50000000 .00001234 00000-0 87654-4 0 9991",
"tle_line2": "2 44713 53.0500 234.5600 0001234 90.1234 270.0012 15.06390000200001",
},
"starlink1808": {
"name": "STARLINK-1808",
"norad_id": 46161,
"tle_line1": "1 46161U 20055A 24045.50000000 .00001500 00000-0 91234-4 0 9998",
"tle_line2": "2 46161 53.0500 120.3400 0001500 45.6789 314.4321 15.06380000150001",
},
}
# Ground station: Kennedy Space Center, Florida
GROUND_STATION = {
"name": "Kennedy Space Center",
"lat": 28.5721,
"lon": -80.6480,
"alt_km": 0.003,
}
# Load TLE data into Satrec propagator objects on module import
_satellites = {}
for sat_id, sat_data in SATELLITE_REGISTRY.items():
_satellites[sat_id] = Satrec.twoline2rv(
sat_data["tle_line1"], sat_data["tle_line2"]
)
def _gmst(jd, fr):
"""Compute Greenwich Mean Sidereal Time in radians.
Uses theta_GMST1982 from Vallado AIAA 2006-6753 Appendix C.
"""
# Julian centuries from J2000.0
t = (jd - 2451545.0 + fr) / 36525.0
# GMST in seconds of time
g = 67310.54841 + (8640184.812866 + (0.093104 + (-6.2e-6) * t) * t) * t
# Convert seconds of time to radians (86400 sec/day, 2*pi rad/day)
theta = math.fmod(g, 86400.0) / 86400.0 * 2.0 * math.pi
return theta
def teme_to_ecef(pos_teme, dt):
"""Rotate TEME position vector to ECEF using GMST angle.
Applies rot_z(-theta) per Vallado AIAA 2006-6753.
"""
# Get Julian date components for the given datetime
jd, fr = jday(
dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second + dt.microsecond / 1e6
)
theta = _gmst(jd, fr)
# Apply rotation matrix Rz(-theta) to convert TEME -> ECEF
cos_t = math.cos(theta)
sin_t = math.sin(theta)
x_teme, y_teme, z_teme = pos_teme
x_ecef = cos_t * x_teme + sin_t * y_teme
y_ecef = -sin_t * x_teme + cos_t * y_teme
z_ecef = z_teme
return (x_ecef, y_ecef, z_ecef)
def ecef_to_geodetic(x, y, z):
"""Convert ECEF coordinates (km) to geodetic lat/lon/alt.
Uses WGS84 ellipsoid with iterative latitude refinement.
"""
# WGS84 parameters
a = 6378.137 # semi-major axis in km
f = 1.0 / 298.257223563 # flattening
b = a * (1.0 - f) # semi-minor axis
e2 = 1.0 - (b * b) / (a * a) # eccentricity squared
# Longitude is straightforward
lon = math.atan2(y, x)
# Distance from z-axis
p = math.sqrt(x * x + y * y)
# Initial latitude estimate
lat = math.atan2(z, p * (1.0 - e2))
# Iterative refinement (converges in 3-5 iterations)
for _ in range(10):
sin_lat = math.sin(lat)
N = a / math.sqrt(1.0 - e2 * sin_lat * sin_lat)
lat = math.atan2(z + e2 * N * sin_lat, p)
# Compute altitude
sin_lat = math.sin(lat)
N = a / math.sqrt(1.0 - e2 * sin_lat * sin_lat)
alt_km = p / math.cos(lat) - N
return (math.degrees(lat), math.degrees(lon), alt_km)
def get_satellite_position(sat_id, dt):
"""Get satellite lat/lon/alt_km at a given UTC datetime."""
if sat_id not in _satellites:
return None
sat = _satellites[sat_id]
# Convert datetime to Julian date components
jd, fr = jday(
dt.year, dt.month, dt.day,
dt.hour, dt.minute, dt.second + dt.microsecond / 1e6
)
# Propagate orbit to get TEME position (km) and velocity (km/s)
e, pos_teme, vel_teme = sat.sgp4(jd, fr)
if e != 0:
return None
# Convert TEME -> ECEF -> geodetic
pos_ecef = teme_to_ecef(pos_teme, dt)
lat, lon, alt_km = ecef_to_geodetic(*pos_ecef)
return {"lat": lat, "lon": lon, "alt_km": alt_km}
def get_ground_track(sat_id, minutes=90, interval_s=60):
"""Generate ground track points for the next N minutes."""
now = datetime.now(timezone.utc)
track = []
for i in range(0, minutes * 60, interval_s):
dt = now + timedelta(seconds=i)
pos = get_satellite_position(sat_id, dt)
if pos:
track.append({"time": dt.isoformat(), **pos})
return track
def _elevation_angle(sat_lat, sat_lon, sat_alt_km):
"""Compute elevation angle (degrees) from ground station to satellite.
Uses spherical Earth approximation for the geometry.
"""
gs_lat = math.radians(GROUND_STATION["lat"])
gs_lon = math.radians(GROUND_STATION["lon"])
sat_lat_r = math.radians(sat_lat)
sat_lon_r = math.radians(sat_lon)
# Central angle between ground station and sub-satellite point
cos_gamma = (
math.sin(gs_lat) * math.sin(sat_lat_r)
+ math.cos(gs_lat) * math.cos(sat_lat_r)
* math.cos(sat_lon_r - gs_lon)
)
cos_gamma = max(-1.0, min(1.0, cos_gamma))
gamma = math.acos(cos_gamma)
# Elevation angle from ground station horizon
R = 6378.137
r_sat = R + sat_alt_km
r_gs = R + GROUND_STATION["alt_km"]
elev = math.atan2(
r_sat * cos_gamma - r_gs, r_sat * math.sin(gamma)
)
return math.degrees(elev)
def get_passes(sat_id, hours=24):
"""Predict passes over ground station in the next N hours.
Steps through time at 30s increments detecting AOS/LOS.
"""
if sat_id not in _satellites:
return []
now = datetime.now(timezone.utc)
passes = []
in_pass = False
current_pass = {}
max_elev = 0.0
# Step through time at 30-second increments
steps = int(hours * 3600 / 30)
for i in range(steps):
dt = now + timedelta(seconds=i * 30)
pos = get_satellite_position(sat_id, dt)
if not pos:
continue
elev = _elevation_angle(pos["lat"], pos["lon"], pos["alt_km"])
if elev > 0 and not in_pass:
# Acquisition of Signal (AOS)
in_pass = True
current_pass = {"aos_time": dt.isoformat(), "max_elevation": elev}
max_elev = elev
elif elev > 0 and in_pass:
# Track maximum elevation during the pass
if elev > max_elev:
max_elev = elev
current_pass["max_elevation"] = elev
elif elev <= 0 and in_pass:
# Loss of Signal (LOS)
in_pass = False
current_pass["los_time"] = dt.isoformat()
aos_dt = datetime.fromisoformat(current_pass["aos_time"])
current_pass["duration_seconds"] = (dt - aos_dt).total_seconds()
passes.append(current_pass)
return passesYour orbital propagation layer is live. Next up, you will build a scheduler that queues operational tasks against these predicted pass windows and auto-executes them through the autopilot.
Build the Contact Window Scheduler
Your satellite registry can now serve live positions and ground tracks for any satellite in the constellation.
The next challenge is figuring out when a satellite will be overhead. A ground station can only communicate with a satellite during a contact window (also called a pass). Outside that window, the satellite is below the horizon and unreachable. In this step, you will predict contact windows, queue operational tasks against them, and auto-execute those tasks through the autopilot sequencer from Project 1.
In this step, get ready to:
- Implement pass prediction using elevation angle stepping over 24 hours.
- Build a task queue with greedy first-fit scheduling into pass windows.
- Connect the scheduler to the autopilot sequencer for automatic task execution.
Implement pass prediction
A satellite is "visible" to the ground station when its elevation angle is above 0 degrees. The moment it rises above 0 is called AOS (Acquisition of Signal). When it drops back below 0, that is LOS (Loss of Signal).
You will step forward through 24 hours of time in 30-second increments, compute the elevation angle at each step, and record every AOS/LOS transition as a pass window.
- Open scheduler.py and add the imports, data models, and predict_passes() function by pasting the following code:
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import math
import uuid
import asyncio
from orbital import SATELLITE_REGISTRY, GROUND_STATION, get_satellite_position
@dataclass
class PassWindow:
"""A predicted contact window over the ground station."""
satellite_id: str
aos_time: datetime # Acquisition of Signal
los_time: datetime # Loss of Signal
max_elevation: float # degrees above horizon
duration_seconds: float
# Task durations in seconds
TASK_DURATIONS = {
"payload_downlink": 120,
"housekeeping_dump": 60,
"attitude_upload": 90,
}
@dataclass
class Task:
"""An operational task queued against a pass window."""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
satellite_id: str = ""
task_type: str = ""
pass_id: str = ""
status: str = "PENDING" # PENDING, SCHEDULED, DEFERRED, EXECUTED
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
# In-memory task queue
task_queue: list[Task] = []
def predict_passes(sat_id: str, hours: int = 24) -> list[PassWindow]:
"""Predict passes over the ground station using 30s elevation stepping."""
passes = []
now = datetime.now(timezone.utc)
step_seconds = 30
total_steps = int(hours * 3600 / step_seconds)
# Ground station coordinates in radians
gs_lat = math.radians(GROUND_STATION["lat"])
gs_lon = math.radians(GROUND_STATION["lon"])
gs_alt_km = GROUND_STATION["alt_km"]
in_pass = False
aos_time = None
max_el = 0.0
for i in range(total_steps):
dt = now + timedelta(seconds=i * step_seconds)
pos = get_satellite_position(sat_id, dt)
if pos is None:
continue
# Convert satellite position to radians
sat_lat = math.radians(pos["lat"])
sat_lon = math.radians(pos["lon"])
sat_alt_km = pos["alt_km"]
# Compute elevation angle using slant range geometry
earth_radius = 6371.0
gs_r = earth_radius + gs_alt_km
sat_r = earth_radius + sat_alt_km
# Central angle between ground station and satellite sub-point
cos_gamma = (math.sin(gs_lat) * math.sin(sat_lat) +
math.cos(gs_lat) * math.cos(sat_lat) *
math.cos(sat_lon - gs_lon))
gamma = math.acos(max(-1, min(1, cos_gamma)))
# Elevation angle from ground station to satellite
elevation = math.degrees(
math.atan2(sat_r * math.cos(gamma) - gs_r,
sat_r * math.sin(gamma))
)
if elevation > 0 and not in_pass:
# AOS detected
in_pass = True
aos_time = dt
max_el = elevation
elif elevation > 0 and in_pass:
max_el = max(max_el, elevation)
elif elevation <= 0 and in_pass:
# LOS detected
in_pass = False
los_time = dt
duration = (los_time - aos_time).total_seconds()
passes.append(PassWindow(
satellite_id=sat_id,
aos_time=aos_time,
los_time=los_time,
max_elevation=round(max_el, 2),
duration_seconds=duration,
))
aos_time = None
max_el = 0.0
return passesWhat does this code do?
The PassWindow dataclass represents a single contact opportunity with a satellite. It records when the pass starts (AOS), when it ends (LOS), the peak elevation, and how long the window lasts.
The Task dataclass models an operational job (like a payload downlink) that needs to run during a pass. Each task gets a unique short ID and tracks its lifecycle status.
The predict_passes() function is the core predictor. It steps through time in 30-second increments, calculates the satellite's elevation above the ground station at each step, and records AOS/LOS transitions. The elevation formula uses the central angle between the ground station and the satellite's sub-point to compute the geometric line-of-sight angle.
- Save scheduler.py.
Why 30-second steps?
A typical LEO satellite pass lasts 5 to 15 minutes. Stepping every 30 seconds gives you 10 to 30 data points per pass, which is enough to catch AOS and LOS accurately without wasting compute. For 5 satellites over 24 hours, that is only about 14,400 propagation calls total.
Build the task queue API
With pass prediction working, you can now schedule operational tasks into those windows. The scheduler uses a greedy first-fit algorithm: it iterates through queued tasks and assigns each one to the earliest pass that has enough remaining time.
- In scheduler.py, add the queue_task() and _schedule_tasks() functions below the predict_passes() function:
def queue_task(satellite_id: str, task_type: str, pass_id: str = "") -> Task:
"""Queue a task and run greedy first-fit scheduling."""
task = Task(
satellite_id=satellite_id,
task_type=task_type,
pass_id=pass_id,
status="PENDING",
)
task_queue.append(task)
# Run greedy first-fit scheduling for this satellite
passes = predict_passes(satellite_id)
_schedule_tasks(passes, satellite_id)
return task
def _schedule_tasks(passes: list[PassWindow], satellite_id: str):
"""Greedy first-fit: assign tasks to the earliest pass with enough time."""
# Calculate remaining time in each pass window
pass_remaining = {}
for i, pw in enumerate(passes):
pass_id = f"{satellite_id}-pass-{i}"
already_scheduled = sum(
TASK_DURATIONS[t.task_type]
for t in task_queue
if t.status == "SCHEDULED" and t.pass_id == pass_id
)
pass_remaining[pass_id] = pw.duration_seconds - already_scheduled
# Try to fit each PENDING task into the earliest available pass
for task in task_queue:
if task.satellite_id != satellite_id or task.status != "PENDING":
continue
duration_needed = TASK_DURATIONS.get(task.task_type, 0)
scheduled = False
for i, pw in enumerate(passes):
pass_id = f"{satellite_id}-pass-{i}"
if pass_remaining.get(pass_id, 0) >= duration_needed:
task.status = "SCHEDULED"
task.pass_id = pass_id
pass_remaining[pass_id] -= duration_needed
scheduled = True
break
if not scheduled:
task.status = "DEFERRED"How does greedy first-fit scheduling work?
The algorithm calculates how much time remains in each upcoming pass (total duration minus time already claimed by SCHEDULED tasks). It then iterates through PENDING tasks and assigns each to the first pass with enough remaining capacity.
If no pass can fit the task, its status becomes DEFERRED. This is intentionally simple. The Secret Mission upgrades this to priority-aware scheduling with task displacement.
- Save scheduler.py.
Tasks always getting DEFERRED?
If every task is marked DEFERRED, the predicted passes may have very short durations (under 60 seconds). Check your ground station coordinates in orbital.py to make sure they match a real location with good visibility to your satellite orbits.
Also verify that predict_passes() returns at least one pass. If the list is empty, the satellite's TLE may have decayed or the ground station latitude may be too extreme for the orbit inclination.
Help me debug why my tasks are all DEFERRED
Connect to the autopilot sequencer
The final piece connects scheduling to execution. When a pass window opens (the current time reaches AOS), the scheduler triggers the autopilot to run all SCHEDULED tasks for that pass in sequence.
- In scheduler.py, add the autopilot integration and background monitor below the _schedule_tasks() function:
async def trigger_pass_execution(pass_window: PassWindow):
"""Execute all SCHEDULED tasks for a pass through the autopilot sequencer."""
# Find the pass_id for this window
pass_id = None
passes = predict_passes(pass_window.satellite_id)
for i, pw in enumerate(passes):
if pw.aos_time == pass_window.aos_time:
pass_id = f"{pass_window.satellite_id}-pass-{i}"
break
if pass_id is None:
return
# Map task types to Project 1 procedure names
procedure_map = {
"payload_downlink": "safe_mode_recovery",
"housekeeping_dump": "telemetry_dump",
"attitude_upload": "attitude_correction",
}
scheduled_tasks = [
t for t in task_queue
if t.pass_id == pass_id and t.status == "SCHEDULED"
]
for task in scheduled_tasks:
procedure_name = procedure_map.get(task.task_type, "telemetry_dump")
print(f"[SCHEDULER] Executing {task.task_type} via procedure "
f"'{procedure_name}' for satellite {task.satellite_id}")
# In production, this calls the autopilot sequencer from Project 1
# autopilot.execute_procedure(procedure_name)
task.status = "EXECUTED"
return scheduled_tasks
async def pass_monitor():
"""Background task that checks for AOS times and triggers execution."""
while True:
now = datetime.now(timezone.utc)
for sat_id in SATELLITE_REGISTRY:
passes = predict_passes(sat_id, hours=1)
for pw in passes:
# Check if we are within the pass window
if pw.aos_time <= now <= pw.los_time:
await trigger_pass_execution(pw)
await asyncio.sleep(30)What does the pass monitor do?
The pass_monitor() coroutine runs in the background every 30 seconds. It checks whether the current UTC time falls within any predicted pass window for any satellite. If it does, it fires trigger_pass_execution() which runs all SCHEDULED tasks for that pass through the procedure engine.
The procedure_map dictionary maps each task type to a procedure name from Project 1's engine. In a production system, this would call the autopilot sequencer directly. For this demo, it logs execution and marks tasks as EXECUTED.
- Save scheduler.py.
Now register the scheduler endpoints and background task in your main application.
- Open main.py and add the scheduler imports alongside your existing orbital imports:
from scheduler import (
predict_passes, queue_task, task_queue,
TASK_DURATIONS, pass_monitor
)- Inside your FastAPI lifespan function, start the pass monitor as a background task:
# Add inside your lifespan function, after existing startup tasks
asyncio.create_task(pass_monitor())- Add the three scheduler REST endpoints to main.py:
@app.post("/scheduler/queue-task")
async def api_queue_task(satellite_id: str, task_type: str, pass_id: str = ""):
"""Queue an operational task against a satellite pass."""
if task_type not in TASK_DURATIONS:
return {"error": f"Invalid task type. Choose from: {list(TASK_DURATIONS.keys())}"}
task = queue_task(satellite_id, task_type, pass_id)
return {"task_id": task.id, "status": task.status, "pass_id": task.pass_id}
@app.get("/scheduler/passes")
async def api_scheduler_passes(satellite_id: str = ""):
"""Get upcoming passes with their queued tasks."""
from orbital import SATELLITE_REGISTRY
results = []
sat_ids = [satellite_id] if satellite_id else list(SATELLITE_REGISTRY.keys())
for sid in sat_ids:
passes = predict_passes(sid)
for i, pw in enumerate(passes):
pass_id = f"{sid}-pass-{i}"
tasks = [t for t in task_queue if t.pass_id == pass_id]
results.append({
"pass_id": pass_id,
"satellite_id": sid,
"aos_time": pw.aos_time.isoformat(),
"los_time": pw.los_time.isoformat(),
"max_elevation": pw.max_elevation,
"duration_seconds": pw.duration_seconds,
"tasks": [{"id": t.id, "type": t.task_type, "status": t.status} for t in tasks],
})
return results
@app.get("/scheduler/tasks")
async def api_scheduler_tasks():
"""Get all tasks with their current status."""
return [
{
"id": t.id,
"satellite_id": t.satellite_id,
"task_type": t.task_type,
"pass_id": t.pass_id,
"status": t.status,
"created_at": t.created_at.isoformat(),
}
for t in task_queue
]What do these endpoints do?
- POST /scheduler/queue-task accepts a satellite ID and task type, queues the task, runs the scheduler, and returns the assigned status.
- GET /scheduler/passes returns all upcoming pass windows for a satellite (or all satellites) with their associated tasks.
- GET /scheduler/tasks returns every task in the queue with its current lifecycle status.
- Save main.py.
- Start (or restart) the server by running:
uvicorn main:app --reload- Queue 3 tasks against the first satellite by running these commands in a separate terminal:
curl -X POST "http://localhost:8000/scheduler/queue-task?satellite_id=ISS&task_type=payload_downlink"
curl -X POST "http://localhost:8000/scheduler/queue-task?satellite_id=ISS&task_type=housekeeping_dump"
curl -X POST "http://localhost:8000/scheduler/queue-task?satellite_id=ISS&task_type=attitude_upload"- Verify the scheduler assigned statuses by running:
curl http://localhost:8000/scheduler/tasksYou should see 3 tasks in the response. Tasks that fit within the first pass are marked SCHEDULED with a pass_id assigned. Any task that overflows is marked DEFERRED.
Getting an import error on startup?
Make sure the import statement in main.py exactly matches the function names exported from scheduler.py: predict_passes, queue_task, task_queue, TASK_DURATIONS, and pass_monitor.
Also check that scheduler.py imports from orbital correctly. If you see a circular import error, ensure orbital.py does not import from scheduler.py.
Help me fix this scheduler import error
✔️ Awesome, I've got everything!
Great. Double check you have saved both scheduler.py and main.py.
ⓧ I'd like to double check the full code
Here is the complete scheduler.py for this step:
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import math
import uuid
import asyncio
from orbital import SATELLITE_REGISTRY, GROUND_STATION, get_satellite_position
@dataclass
class PassWindow:
"""A predicted contact window over the ground station."""
satellite_id: str
aos_time: datetime # Acquisition of Signal
los_time: datetime # Loss of Signal
max_elevation: float # degrees above horizon
duration_seconds: float
# Task durations in seconds
TASK_DURATIONS = {
"payload_downlink": 120,
"housekeeping_dump": 60,
"attitude_upload": 90,
}
@dataclass
class Task:
"""An operational task queued against a pass window."""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
satellite_id: str = ""
task_type: str = ""
pass_id: str = ""
status: str = "PENDING" # PENDING, SCHEDULED, DEFERRED, EXECUTED
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
# In-memory task queue
task_queue: list[Task] = []
def predict_passes(sat_id: str, hours: int = 24) -> list[PassWindow]:
"""Predict passes over the ground station using 30s elevation stepping."""
passes = []
now = datetime.now(timezone.utc)
step_seconds = 30
total_steps = int(hours * 3600 / step_seconds)
# Ground station coordinates in radians
gs_lat = math.radians(GROUND_STATION["lat"])
gs_lon = math.radians(GROUND_STATION["lon"])
gs_alt_km = GROUND_STATION["alt_km"]
in_pass = False
aos_time = None
max_el = 0.0
for i in range(total_steps):
dt = now + timedelta(seconds=i * step_seconds)
pos = get_satellite_position(sat_id, dt)
if pos is None:
continue
# Convert satellite position to radians
sat_lat = math.radians(pos["lat"])
sat_lon = math.radians(pos["lon"])
sat_alt_km = pos["alt_km"]
# Compute elevation angle using slant range geometry
earth_radius = 6371.0
gs_r = earth_radius + gs_alt_km
sat_r = earth_radius + sat_alt_km
# Central angle between ground station and satellite sub-point
cos_gamma = (math.sin(gs_lat) * math.sin(sat_lat) +
math.cos(gs_lat) * math.cos(sat_lat) *
math.cos(sat_lon - gs_lon))
gamma = math.acos(max(-1, min(1, cos_gamma)))
# Elevation angle from ground station to satellite
elevation = math.degrees(
math.atan2(sat_r * math.cos(gamma) - gs_r,
sat_r * math.sin(gamma))
)
if elevation > 0 and not in_pass:
# AOS detected
in_pass = True
aos_time = dt
max_el = elevation
elif elevation > 0 and in_pass:
max_el = max(max_el, elevation)
elif elevation <= 0 and in_pass:
# LOS detected
in_pass = False
los_time = dt
duration = (los_time - aos_time).total_seconds()
passes.append(PassWindow(
satellite_id=sat_id,
aos_time=aos_time,
los_time=los_time,
max_elevation=round(max_el, 2),
duration_seconds=duration,
))
aos_time = None
max_el = 0.0
return passes
def queue_task(satellite_id: str, task_type: str, pass_id: str = "") -> Task:
"""Queue a task and run greedy first-fit scheduling."""
task = Task(
satellite_id=satellite_id,
task_type=task_type,
pass_id=pass_id,
status="PENDING",
)
task_queue.append(task)
# Run greedy first-fit scheduling for this satellite
passes = predict_passes(satellite_id)
_schedule_tasks(passes, satellite_id)
return task
def _schedule_tasks(passes: list[PassWindow], satellite_id: str):
"""Greedy first-fit: assign tasks to the earliest pass with enough time."""
# Calculate remaining time in each pass window
pass_remaining = {}
for i, pw in enumerate(passes):
pass_id = f"{satellite_id}-pass-{i}"
already_scheduled = sum(
TASK_DURATIONS[t.task_type]
for t in task_queue
if t.status == "SCHEDULED" and t.pass_id == pass_id
)
pass_remaining[pass_id] = pw.duration_seconds - already_scheduled
# Try to fit each PENDING task into the earliest available pass
for task in task_queue:
if task.satellite_id != satellite_id or task.status != "PENDING":
continue
duration_needed = TASK_DURATIONS.get(task.task_type, 0)
scheduled = False
for i, pw in enumerate(passes):
pass_id = f"{satellite_id}-pass-{i}"
if pass_remaining.get(pass_id, 0) >= duration_needed:
task.status = "SCHEDULED"
task.pass_id = pass_id
pass_remaining[pass_id] -= duration_needed
scheduled = True
break
if not scheduled:
task.status = "DEFERRED"
async def trigger_pass_execution(pass_window: PassWindow):
"""Execute all SCHEDULED tasks for a pass through the autopilot sequencer."""
# Find the pass_id for this window
pass_id = None
passes = predict_passes(pass_window.satellite_id)
for i, pw in enumerate(passes):
if pw.aos_time == pass_window.aos_time:
pass_id = f"{pass_window.satellite_id}-pass-{i}"
break
if pass_id is None:
return
# Map task types to Project 1 procedure names
procedure_map = {
"payload_downlink": "safe_mode_recovery",
"housekeeping_dump": "telemetry_dump",
"attitude_upload": "attitude_correction",
}
scheduled_tasks = [
t for t in task_queue
if t.pass_id == pass_id and t.status == "SCHEDULED"
]
for task in scheduled_tasks:
procedure_name = procedure_map.get(task.task_type, "telemetry_dump")
print(f"[SCHEDULER] Executing {task.task_type} via procedure "
f"'{procedure_name}' for satellite {task.satellite_id}")
# In production, this calls the autopilot sequencer from Project 1
# autopilot.execute_procedure(procedure_name)
task.status = "EXECUTED"
return scheduled_tasks
async def pass_monitor():
"""Background task that checks for AOS times and triggers execution."""
while True:
now = datetime.now(timezone.utc)
for sat_id in SATELLITE_REGISTRY:
passes = predict_passes(sat_id, hours=1)
for pw in passes:
# Check if we are within the pass window
if pw.aos_time <= now <= pw.los_time:
await trigger_pass_execution(pw)
await asyncio.sleep(30)Your scheduler can now predict contact windows, intelligently assign tasks to passes, and auto-execute them when the pass opens. Next up, you will persist all telemetry and command data to SQLite and build a live dashboard to visualize everything.
Build the Mission Log and Dashboard
Your scheduler is queueing tasks and firing them through the autopilot during pass windows. The missing piece is visibility. Without persistent logs and a visual interface, operators have no way to review what happened after a pass ends.
In this step, you will build a SQLite-backed mission log that records every telemetry reading, command execution, and resolver action. Then you will create a live HTML dashboard with a Leaflet map showing satellite positions, a contact timeline, real-time telemetry, and a scrollable audit log.
In this step, get ready to:
- Implement the SQLite persistence layer with write and read functions.
- Build a dashboard HTML page with a Leaflet map, ground tracks, and contact timeline.
- Add a live telemetry panel and mission log table to the dashboard.
Implement the SQLite persistence layer
Every telemetry reading, command execution, and resolver decision needs to be stored for post-pass review. The persistence module wraps aiosqlite so the async event loop never blocks on database writes.
- Open persistence.py and add the database initialization function by pasting this code:
import aiosqlite
from datetime import datetime
# Path to the SQLite database file
DB_PATH = "mission_log.sqlite"
async def init_db():
"""Create mission log tables if they don't exist."""
async with aiosqlite.connect(DB_PATH) as db:
# Telemetry readings from all satellites
await db.execute("""
CREATE TABLE IF NOT EXISTS telemetry_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
satellite_id TEXT NOT NULL,
channel TEXT NOT NULL,
value REAL NOT NULL,
alarm_state TEXT NOT NULL
)
""")
# Command executions during pass windows
await db.execute("""
CREATE TABLE IF NOT EXISTS command_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
satellite_id TEXT NOT NULL,
task_type TEXT NOT NULL,
procedure_name TEXT NOT NULL,
outcome TEXT NOT NULL
)
""")
# Resolver autonomous actions
await db.execute("""
CREATE TABLE IF NOT EXISTS resolver_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
alert_condition TEXT NOT NULL,
procedure_fired TEXT NOT NULL,
result TEXT NOT NULL
)
""")
await db.commit()What does this code do?
DB_PATH sets the SQLite file location. On Render's free tier this file is ephemeral (resets on redeploy), which is fine for a demo.
init_db() creates three tables if they don't already exist. Each table captures a different event type from the automation engine: raw telemetry, executed commands, and resolver-triggered procedures.
- Save persistence.py.
- You won't see a change yet. The next chunk adds the write functions that populate these tables.
- Below init_db() in persistence.py, add the three write functions:
async def log_telemetry(satellite_id, channel, value, alarm_state):
"""Write a telemetry reading to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO telemetry_log (timestamp, satellite_id, channel, value, alarm_state) VALUES (?, ?, ?, ?, ?)",
(datetime.utcnow().isoformat(), satellite_id, channel, value, alarm_state)
)
await db.commit()
async def log_command(satellite_id, task_type, procedure_name, outcome):
"""Write a command execution record to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO command_log (timestamp, satellite_id, task_type, procedure_name, outcome) VALUES (?, ?, ?, ?, ?)",
(datetime.utcnow().isoformat(), satellite_id, task_type, procedure_name, outcome)
)
await db.commit()
async def log_resolver(alert_condition, procedure_fired, result):
"""Write a resolver action record to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO resolver_log (timestamp, alert_condition, procedure_fired, result) VALUES (?, ?, ?, ?)",
(datetime.utcnow().isoformat(), alert_condition, procedure_fired, result)
)
await db.commit()What does this code do?
Each write function opens a short-lived connection, inserts one row with the current UTC timestamp, and commits immediately. This pattern keeps connections brief and avoids holding locks during the telemetry simulation loop.
log_telemetry() captures every sensor reading. log_command() records each task the autopilot fires during a pass. log_resolver() records autonomous corrective actions from Project 1's resolver.
- Save persistence.py.
- You won't see a change yet. The next chunk adds read functions that query these tables.
- Below the write functions in persistence.py, add the three read functions with time-range filtering:
async def get_telemetry_logs(start_time=None, end_time=None):
"""Retrieve telemetry log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM telemetry_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_command_logs(start_time=None, end_time=None):
"""Retrieve command log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM command_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_resolver_logs(start_time=None, end_time=None):
"""Retrieve resolver log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM resolver_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]What does this code do?
Each read function builds a SELECT query dynamically. If start_time or end_time are provided, it appends WHERE clauses to filter by timestamp range.
Setting db.row_factory = aiosqlite.Row enables dict-like access on each row. The dict(row) conversion produces JSON-serializable dictionaries that FastAPI can return directly.
- Save persistence.py.
Seeing an import error for aiosqlite?
Make sure you installed aiosqlite==0.22.1 in Step 1. Run pip show aiosqlite to confirm it's in your environment.
help me debug an aiosqlite import error in my persistence module
- Now wire the persistence layer into main.py. Add the following imports, lifespan hook, logging hooks, log endpoints, and dashboard route:
import persistence
from fastapi.responses import FileResponse
from typing import Optional
# --- Inside your lifespan function, add at the top of the startup block: ---
await persistence.init_db()
# --- Hook logging into existing event handlers ---
# In your telemetry monitor callback, add:
await persistence.log_telemetry(satellite_id, channel, value, alarm_state)
# In your procedure engine execution callback, add:
await persistence.log_command(satellite_id, task_type, procedure_name, outcome)
# In your resolver action callback, add:
await persistence.log_resolver(alert_condition, procedure_fired, result)
# --- Log query endpoints ---
@app.get("/logs/telemetry")
async def get_telemetry_logs(start_time: Optional[str] = None, end_time: Optional[str] = None):
"""Retrieve telemetry log with optional time-range filter."""
return await persistence.get_telemetry_logs(start_time, end_time)
@app.get("/logs/commands")
async def get_command_logs(start_time: Optional[str] = None, end_time: Optional[str] = None):
"""Retrieve command log with optional time-range filter."""
return await persistence.get_command_logs(start_time, end_time)
@app.get("/logs/resolver")
async def get_resolver_logs(start_time: Optional[str] = None, end_time: Optional[str] = None):
"""Retrieve resolver log with optional time-range filter."""
return await persistence.get_resolver_logs(start_time, end_time)
# --- Serve the dashboard at root ---
@app.get("/", response_class=FileResponse)
async def serve_dashboard():
"""Serve the mission operations dashboard."""
return "dashboard.html"What does this code do?
await persistence.init_db() creates the SQLite tables on app startup. The three logging hooks call the write functions you just built whenever the telemetry monitor, procedure engine, or resolver fires an event.
The three GET /logs/* endpoints expose the read functions as REST queries. Optional start_time and end_time query params allow filtering by ISO timestamp range.
The GET / route uses FastAPI's FileResponse to serve the dashboard HTML file directly from disk.
- Save main.py.
- Start the server and visit http://localhost:8000/logs/telemetry in your browser. You should see an empty JSON array [] (no telemetry recorded yet).
Getting a 500 error on the logs endpoint?
Check that await persistence.init_db() runs before any request hits the log endpoints. It must be inside your lifespan startup block, not at module level.
help me debug a 500 error on my /logs/telemetry endpoint
Build the dashboard HTML page
The dashboard is a single HTML file that uses Leaflet 1.9.4 to render an interactive map. It fetches satellite positions from your API every 5 seconds and draws ground track polylines.
- Create a new file called dashboard.html in your project root and paste the following:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>GroundOps Console - Mission Dashboard</title>
<!-- Leaflet CSS -->
<link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css"
integrity="sha256-p4NxAoJBhIIN+hmNHrzRCf9tD/miZyoHS5obTRR9BMY="
crossorigin="" />
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: monospace; background: #1a1a2e; color: #eee; }
h1 { padding: 10px 20px; background: #16213e; font-size: 1.2em; }
#map { width: 100%; height: 400px; }
.panel { padding: 15px 20px; }
.panel h2 { font-size: 1em; margin-bottom: 10px; color: #0f9d58; }
/* Timeline bars */
.timeline-bar { display: flex; align-items: center; margin: 4px 0; }
.timeline-bar .label { width: 120px; font-size: 0.8em; }
.timeline-bar .bar { height: 20px; background: #4285f4; border-radius: 3px;
display: flex; align-items: center; padding: 0 6px;
font-size: 0.7em; color: #fff; min-width: 40px; }
/* Telemetry table */
table { width: 100%; border-collapse: collapse; font-size: 0.8em; }
th, td { padding: 4px 8px; border: 1px solid #333; text-align: left; }
th { background: #16213e; }
.WARNING { background: #f9a825; color: #000; }
.CRITICAL { background: #d32f2f; color: #fff; }
.NOMINAL { background: transparent; }
</style>
</head>
<body>
<h1>GroundOps Console - Mission Dashboard</h1>
<!-- Leaflet Map -->
<div id="map"></div>
<!-- Contact Timeline -->
<div class="panel">
<h2>Contact Timeline (Next 24h)</h2>
<div id="timeline"></div>
</div>
<!-- Telemetry Panel -->
<div class="panel">
<h2>Live Telemetry</h2>
<table id="telemetry-table">
<thead><tr><th>Satellite</th><th>Channel</th><th>Value</th><th>State</th></tr></thead>
<tbody></tbody>
</table>
</div>
<!-- Log Table -->
<div class="panel">
<h2>Mission Log (Last 50 Actions)</h2>
<table id="log-table">
<thead><tr><th>Time</th><th>Type</th><th>Details</th><th>Outcome</th></tr></thead>
<tbody></tbody>
</table>
</div>
<!-- Leaflet JS -->
<script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"
integrity="sha256-20nQCchB9co0qIjJZRGuk2/Z9VM+kNiyxNV1lvTlZBo="
crossorigin=""></script>
<script>
// --- Configuration ---
const GROUND_STATION = { lat: 42.36, lon: -71.09, name: "Ground Station" };
const REFRESH_INTERVAL = 5000; // 5 seconds
// --- Initialize Leaflet map ---
var map = L.map('map').setView([GROUND_STATION.lat, GROUND_STATION.lon], 2);
L.tileLayer('https://tile.openstreetmap.org/{z}/{x}/{y}.png', {
attribution: '© OpenStreetMap contributors',
maxZoom: 18
}).addTo(map);
// Ground station marker
L.marker([GROUND_STATION.lat, GROUND_STATION.lon], {
icon: L.divIcon({ className: 'gs-icon', html: '📡', iconSize: [24, 24] })
}).addTo(map).bindPopup(GROUND_STATION.name);
// Track satellite markers and polylines
const satMarkers = {};
const satTracks = {};
// --- Fetch satellite list and initialize markers ---
async function initSatellites() {
const res = await fetch('/satellites');
const satellites = await res.json();
for (const sat of satellites) {
// Create marker for each satellite
satMarkers[sat.id] = L.marker([0, 0]).addTo(map).bindPopup(sat.name);
// Fetch and draw ground track
const trackRes = await fetch(`/satellites/${sat.id}/ground-track`);
const trackData = await trackRes.json();
const latlngs = trackData.map(p => [p.lat, p.lon]);
satTracks[sat.id] = L.polyline(latlngs, { color: '#4fc3f7', weight: 1.5, opacity: 0.7 }).addTo(map);
}
}
// --- Update satellite positions every 5s ---
async function updatePositions() {
for (const satId of Object.keys(satMarkers)) {
try {
const res = await fetch(`/satellites/${satId}/position`);
const pos = await res.json();
satMarkers[satId].setLatLng([pos.lat, pos.lon]);
} catch (e) { console.error('Position update failed:', satId, e); }
}
}
// --- Load contact timeline ---
async function loadTimeline() {
const container = document.getElementById('timeline');
const res = await fetch('/scheduler/passes');
const passes = await res.json();
container.innerHTML = '';
for (const pass of passes.slice(0, 10)) {
const duration = pass.duration_seconds || 600;
const barWidth = Math.max(40, (duration / 600) * 200);
const tasks = (pass.tasks || []).map(t => t.task_type).join(', ');
container.innerHTML += `
<div class="timeline-bar">
<span class="label">${pass.satellite_id}</span>
<div class="bar" style="width:${barWidth}px">
${tasks || 'No tasks'}
</div>
</div>`;
}
}
// --- Load telemetry panel ---
async function loadTelemetry() {
const tbody = document.querySelector('#telemetry-table tbody');
const res = await fetch('/logs/telemetry?start_time=' + new Date(Date.now() - 30000).toISOString());
const logs = await res.json();
tbody.innerHTML = '';
for (const row of logs.slice(0, 20)) {
tbody.innerHTML += `
<tr class="${row.alarm_state}">
<td>${row.satellite_id}</td>
<td>${row.channel}</td>
<td>${Number(row.value).toFixed(2)}</td>
<td>${row.alarm_state}</td>
</tr>`;
}
}
// --- Load mission log ---
async function loadMissionLog() {
const tbody = document.querySelector('#log-table tbody');
// Fetch both command and resolver logs
const [cmdRes, resRes] = await Promise.all([
fetch('/logs/commands'),
fetch('/logs/resolver')
]);
const commands = (await cmdRes.json()).map(r => ({...r, type: 'COMMAND'}));
const resolverActions = (await resRes.json()).map(r => ({...r, type: 'RESOLVER'}));
// Merge and sort by timestamp descending, take last 50
const all = [...commands, ...resolverActions]
.sort((a, b) => b.timestamp.localeCompare(a.timestamp))
.slice(0, 50);
tbody.innerHTML = '';
for (const row of all) {
const details = row.type === 'COMMAND'
? `${row.satellite_id} - ${row.task_type} (${row.procedure_name})`
: `${row.alert_condition} -> ${row.procedure_fired}`;
const outcome = row.outcome || row.result || '';
tbody.innerHTML += `
<tr>
<td>${row.timestamp}</td>
<td>${row.type}</td>
<td>${details}</td>
<td>${outcome}</td>
</tr>`;
}
}
// --- Initialize and start refresh loops ---
initSatellites().then(() => {
updatePositions();
setInterval(updatePositions, REFRESH_INTERVAL);
});
loadTimeline();
setInterval(loadTimeline, 30000);
loadTelemetry();
setInterval(loadTelemetry, REFRESH_INTERVAL);
loadMissionLog();
setInterval(loadMissionLog, 10000);
</script>
</body>
</html>What does this dashboard do?
The map section uses Leaflet 1.9.4 loaded from unpkg CDN. On page load, initSatellites() fetches the satellite list from /satellites, creates a marker for each one, and draws ground track polylines from /satellites/{id}/ground-track.
Every 5 seconds, updatePositions() fetches each satellite's current lat/lon from /satellites/{id}/position and moves the marker on the map.
The contact timeline fetches upcoming passes from /scheduler/passes and renders them as horizontal bars with task labels. The telemetry panel polls recent readings and colors cells yellow for WARNING and red for CRITICAL. The log table merges command and resolver logs, sorted by timestamp, showing the last 50 entries.
- Save dashboard.html.
- Confirm the file exists by running:
ls dashboard.html- You should see dashboard.html listed in the output.
Dashboard not loading at localhost:8000?
Make sure the GET / route in main.py returns "dashboard.html" and uses response_class=FileResponse. Also verify that dashboard.html is in the same directory where you run uvicorn.
help me debug why my FastAPI FileResponse isn't serving dashboard.html
Add the telemetry panel and log table
The telemetry panel and log table are already included in the dashboard HTML above. They update on separate intervals: telemetry every 5 seconds, logs every 10 seconds. The color-coded anomaly highlighting uses CSS classes matching the alarm state strings from your telemetry monitor.
- Restart your server by running:
uvicorn main:app --reload- Open http://localhost:8000 in your browser.
- You should see the full dashboard: a Leaflet map with satellite markers moving every 5 seconds, ground track lines drawn across the map, the ground station marker, a contact timeline showing upcoming passes with task labels, a telemetry table with color-coded anomaly cells, and a log table showing recent command and resolver actions.
What should I see?
The satellite markers should move along their orbital ground tracks. If your telemetry simulation is running, you will see readings appear in the telemetry panel with yellow (WARNING) and red (CRITICAL) highlighting. The mission log table populates as the autopilot executes tasks during passes and the resolver fires corrective procedures.
If the map shows but satellites aren't moving, wait a few seconds for the first position refresh cycle to complete.
Satellites not moving on the map?
Open your browser's developer console (F12) and check for fetch errors. The most common issues are: the /satellites endpoint returning an unexpected format, or CORS blocking local requests. Since the HTML is served from the same origin as the API, CORS should not be an issue. Check that your orbital.py endpoints are still registered correctly in main.py.
help me debug why satellite markers aren't updating on my Leaflet map
✔️ Awesome, I've got everything!
Great. Double check you've saved all three files: persistence.py, main.py, and dashboard.html.
ⓧ I'd like to double check the full code
Here is the complete persistence.py:
import aiosqlite
from datetime import datetime
# Path to the SQLite database file
DB_PATH = "mission_log.sqlite"
async def init_db():
"""Create mission log tables if they don't exist."""
async with aiosqlite.connect(DB_PATH) as db:
# Telemetry readings from all satellites
await db.execute("""
CREATE TABLE IF NOT EXISTS telemetry_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
satellite_id TEXT NOT NULL,
channel TEXT NOT NULL,
value REAL NOT NULL,
alarm_state TEXT NOT NULL
)
""")
# Command executions during pass windows
await db.execute("""
CREATE TABLE IF NOT EXISTS command_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
satellite_id TEXT NOT NULL,
task_type TEXT NOT NULL,
procedure_name TEXT NOT NULL,
outcome TEXT NOT NULL
)
""")
# Resolver autonomous actions
await db.execute("""
CREATE TABLE IF NOT EXISTS resolver_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
alert_condition TEXT NOT NULL,
procedure_fired TEXT NOT NULL,
result TEXT NOT NULL
)
""")
await db.commit()
async def log_telemetry(satellite_id, channel, value, alarm_state):
"""Write a telemetry reading to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO telemetry_log (timestamp, satellite_id, channel, value, alarm_state) VALUES (?, ?, ?, ?, ?)",
(datetime.utcnow().isoformat(), satellite_id, channel, value, alarm_state)
)
await db.commit()
async def log_command(satellite_id, task_type, procedure_name, outcome):
"""Write a command execution record to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO command_log (timestamp, satellite_id, task_type, procedure_name, outcome) VALUES (?, ?, ?, ?, ?)",
(datetime.utcnow().isoformat(), satellite_id, task_type, procedure_name, outcome)
)
await db.commit()
async def log_resolver(alert_condition, procedure_fired, result):
"""Write a resolver action record to the mission log."""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"INSERT INTO resolver_log (timestamp, alert_condition, procedure_fired, result) VALUES (?, ?, ?, ?)",
(datetime.utcnow().isoformat(), alert_condition, procedure_fired, result)
)
await db.commit()
async def get_telemetry_logs(start_time=None, end_time=None):
"""Retrieve telemetry log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM telemetry_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_command_logs(start_time=None, end_time=None):
"""Retrieve command log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM command_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_resolver_logs(start_time=None, end_time=None):
"""Retrieve resolver log entries with optional time-range filter."""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
query = "SELECT * FROM resolver_log"
params = []
filters = []
if start_time:
filters.append("timestamp >= ?")
params.append(start_time)
if end_time:
filters.append("timestamp <= ?")
params.append(end_time)
if filters:
query += " WHERE " + " AND ".join(filters)
query += " ORDER BY timestamp DESC"
async with db.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [dict(row) for row in rows]Your mission operations dashboard is live with real-time satellite tracking, contact scheduling visibility, and a full audit trail. Next up, you will deploy everything to Render so the world can see your ground operations console in action.
Deploy to Render
Your mission operations system runs locally with full orbital propagation, scheduling, and a live dashboard. Now it needs to be reachable from anywhere.
In this step, you will prepare your project for cloud deployment, push it to GitHub, and deploy a live instance on Render so anyone can access your satellite operations dashboard from a shareable URL.
In this step, get ready to:
- Create deployment configuration files for Render.
- Push your project to GitHub with a professional README.
- Deploy and verify your live dashboard on Render.
Prepare for deployment
Render needs two things to deploy your app: a list of dependencies to install, and instructions for how to start the server.
- Create a requirements.txt file in your project root by running:
touch requirements.txt- Open requirements.txt and add all your pinned dependencies:
# Core web framework and ASGI server
fastapi[standard]
uvicorn[standard]
# Orbital propagation
sgp4==2.24
# Async SQLite for mission logging
aiosqlite==0.22.1What does this file do?
Render reads requirements.txt during the build step to install every Python package your app needs.
Pinning sgp4==2.24 and aiosqlite==0.22.1 ensures the deployed environment matches your local setup exactly.
Include Project 1 dependencies too
If your Project 1 codebase uses additional packages (e.g., websockets, pydantic), add them to requirements.txt as well. Run pip freeze locally to see everything installed in your virtual environment.
- Create a render.yaml file in your project root. This is a Render Blueprint that tells Render how to build and run your service:
# Render Blueprint for GroundOps Console
services:
- type: web
name: groundops-console
runtime: python
buildCommand: pip install -r requirements.txt
startCommand: uvicorn main:app --host 0.0.0.0 --port $PORTWhat does each field mean?
- type: web tells Render this is a publicly accessible web service.
- runtime: python selects the Python runtime environment with pip available.
- buildCommand runs during deployment to install your packages.
- startCommand is what Render executes to boot your app. The $PORT variable is provided automatically by Render.
Render provides the PORT environment variable automatically. Your app needs to read it.
- In main.py, find the section where the app starts up (the lifespan event or bottom of the file).
- Add an import for os at the top of the file if it is not already present, and ensure the app reads the port from the environment:
import os
# Read PORT from environment (Render provides this automatically)
port = int(os.environ.get("PORT", 8000))Why read PORT from the environment?
Render assigns a dynamic port to each web service. Locally you fall back to 8000, but in production the PORT variable ensures Uvicorn binds to the correct port that Render routes traffic to.
Your background telemetry simulation loop also needs to run in production. FastAPI's lifespan event handles this cleanly.
- Confirm your lifespan function in main.py uses asyncio.create_task to start the background telemetry simulation loop and pass monitor on startup:
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app):
# Start background tasks on startup
asyncio.create_task(telemetry_simulation_loop())
asyncio.create_task(pass_monitor_loop())
yield
# Cleanup on shutdown (if needed)Why asyncio.create_task inside lifespan?
The lifespan context manager runs once when FastAPI starts. By creating tasks here, the telemetry simulator and pass monitor run alongside the web server in the same process.
On Render's free tier, you get a single web process. This pattern keeps everything running without needing a separate worker service.
- Save requirements.txt, render.yaml, and main.py.
- Verify your files exist by running:
ls requirements.txt render.yamlYou should see both files listed in the output.
Files not showing up?
Make sure you created the files in the project root directory (the same folder that contains main.py). Render looks for render.yaml at the repository root by default.
Help me find the right location for my config files.
Push to GitHub
Render deploys directly from a GitHub repository. You will push your code and include a README that documents your system architecture for anyone who visits the repo.
- Create a README.md file in your project root with the following content:
# GroundOps Console - Orbital Operations Layer
A satellite operations backend built on top of the Project 1 automation engine.
Propagates real satellite orbits using SGP4, predicts contact windows over a
ground station, queues operational tasks against upcoming passes, and
auto-executes them during the pass window.
## Architecture
```
TLE Data
│
▼
orbital.py (SGP4 propagation, TEME-to-geodetic, registry)
│
├──▶ REST: /satellites, /satellites/{id}/position, /ground-track, /passes
│
▼
scheduler.py (pass prediction, task queue, greedy first-fit scheduling)
│
├──▶ REST: /scheduler/queue-task, /scheduler/passes, /scheduler/tasks
│
▼
autopilot sequencer (executes tasks during pass window)
│
▼
persistence.py (SQLite: telemetry_log, command_log, resolver_log)
│
├──▶ REST: /logs/telemetry, /logs/commands, /logs/resolver
│
▼
dashboard (Leaflet map, contact timeline, telemetry panel, log viewer)
└──▶ GET /
```
## Data Flow
1. **SGP4 Propagation** - orbital.py loads TLEs into Satrec objects and
propagates positions in the TEME frame, then converts to lat/lon/alt.
2. **Contact Windows** - scheduler.py steps forward in 30s increments over 24h,
detecting AOS/LOS crossings when elevation > 0 degrees.
3. **Task Scheduling** - Greedy first-fit algorithm assigns queued tasks to
available pass windows based on duration.
4. **Auto-Execution** - When a pass window opens, the autopilot sequencer
fires all scheduled tasks in sequence via the procedure engine.
5. **Persistence** - Every telemetry reading, command execution, and resolver
action is logged to SQLite for post-pass audit.
## Local Setup
```bash
python -m venv venv
source venv/bin/activate # or venv\Scripts\activate on Windows
pip install -r requirements.txt
uvicorn main:app --reload
```
Visit http://localhost:8000 for the live dashboard.
## API Endpoints
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/` | GET | Live dashboard (HTML) |
| `/satellites` | GET | List all tracked satellites |
| `/satellites/{id}/position` | GET | Current lat/lon/alt |
| `/satellites/{id}/ground-track` | GET | Next 90 min positions |
| `/satellites/{id}/passes` | GET | Next 24h passes over ground station |
| `/scheduler/queue-task` | POST | Queue a task for a satellite |
| `/scheduler/passes` | GET | Upcoming passes with queued tasks |
| `/scheduler/tasks` | GET | All tasks with status |
| `/logs/telemetry` | GET | Telemetry log (filterable) |
| `/logs/commands` | GET | Command execution log |
| `/logs/resolver` | GET | Resolver action log |
## Deployment
Deployed on Render (free tier). The render.yaml Blueprint defines the service.
SQLite database is ephemeral on Render free tier and resets on redeploy.
## Tech Stack
- **Python 3.10+** with FastAPI and Uvicorn
- **sgp4** for orbital propagation (C++ accelerated backend)
- **aiosqlite** for async database operations
- **Leaflet.js** for the interactive map
- **Render** for cloud hostingWhat makes a good project README?
The architecture diagram shows how data flows between modules. The API endpoint table gives visitors a quick reference without reading code.
Together these demonstrate system-level thinking, which is exactly what hiring managers look for in mission operations software portfolios.
- Initialize a git repository (if not already done) and commit all your files by running:
git init
git add .
git commit -m "GroundOps Console: orbital ops layer with dashboard"- Create a new public repository on GitHub. You can do this at github.com/new.
- Name it something descriptive like groundops-console.
- Push your code to GitHub by running:
git remote add origin [[REPO_URL="https://github.com/YOUR_USERNAME/groundops-console.git"]]
git branch -M main
git push -u origin mainYou should see your files appear in the GitHub repository page after refreshing.
Push rejected or authentication failed?
GitHub requires a personal access token or SSH key for authentication. If you see a credentials error, check that your GitHub authentication is configured for the command line.
Help me troubleshoot my git push.
Deploy and verify on Render
With your code on GitHub and the render.yaml Blueprint in place, you can connect the repo to Render and deploy.
- Open the Render Dashboard in your browser.
- Click New in the top navigation, then select Web Service.
- Select Build and deploy from a Git repository, then click Next.
- Connect your GitHub account if you have not already, then select your groundops-console repository.
- Render auto-detects your render.yaml settings. Verify the fields match:
- Language: Python 3.
- Build Command: pip install -r requirements.txt.
- Start Command: uvicorn main:app --host 0.0.0.0 --port $PORT.
- Select the Free instance type.
- Click Create Web Service.
Render kicks off the build. You can watch the deploy progress on the Events page in the dashboard.
What happens during deployment?
Render clones your repo, runs pip install -r requirements.txt to install all dependencies, then starts Uvicorn with the port Render assigns.
The free tier provides 512 MB RAM and 0.1 CPU. Your service will spin down after 15 minutes of inactivity, then wake on the next request.
- Wait for the deploy to show a green Live status. This typically takes 2-3 minutes.
- Click the .onrender.com URL at the top of your service page.
Your live dashboard should load with satellites moving on the Leaflet map, the contact timeline showing upcoming passes, telemetry updating with color-coded anomalies, and the log table showing recent actions.
- Verify the scheduler endpoints respond by visiting your live URL with /scheduler/passes appended.
- Verify satellite positions by visiting /satellites on your live URL.
- Copy the full .onrender.com URL. This is your shareable mission operations dashboard.
Deploy failed or dashboard not loading?
- Check the Logs tab on your Render service page for import errors or missing dependencies.
- If you see a ModuleNotFoundError, add the missing package to requirements.txt, commit, and push. Render auto-redeploys on every push.
- If the dashboard loads but satellites are not moving, give it 10-15 seconds. The first request after a cold start needs time for the background tasks to initialize.
Help me debug my Render deployment.
✔️ Awesome, I've got everything!
Great. Double check you have saved all your files and that the Render deploy shows a green Live status.
ⓧ I'd like to double check the full code
Here is the full requirements.txt:
# Core web framework and ASGI server
fastapi[standard]
uvicorn[standard]
# Orbital propagation
sgp4==2.24
# Async SQLite for mission logging
aiosqlite==0.22.1Here is the full render.yaml:
# Render Blueprint for GroundOps Console
services:
- type: web
name: groundops-console
runtime: python
buildCommand: pip install -r requirements.txt
startCommand: uvicorn main:app --host 0.0.0.0 --port $PORTThe key changes in main.py:
import os
import asyncio
from contextlib import asynccontextmanager
# Read PORT from environment (Render provides this automatically)
port = int(os.environ.get("PORT", 8000))
@asynccontextmanager
async def lifespan(app):
# Start background tasks on startup
asyncio.create_task(telemetry_simulation_loop())
asyncio.create_task(pass_monitor_loop())
yield
# Cleanup on shutdown (if needed)Your satellite operations system is live. Anyone with the link can watch your constellation orbit the Earth in real-time, see scheduled tasks queue against upcoming passes, and review the mission log. Next, take on the Secret Mission to add priority-based task conflict resolution.
Secret mission
Priority-Based Task Conflict Resolution
Your greedy scheduler works, but real mission operations have competing priorities. What happens when a critical payload downlink needs to happen during a pass that's already packed with routine housekeeping? In this extension, you'll add priority-based scheduling with bumping logic that displaces lower-priority tasks when a high-priority task demands the slot.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs on Render's free tier, so there are no ongoing costs while the service is idle.
Resources you used:
- Render web service (free tier, spins down after 15 minutes of inactivity).
- GitHub repository (free, persists indefinitely).
- Local project files on your Desktop.
Keep everything running
No action needed. Choose this if you're still actively building or want to keep the live dashboard accessible.
- Your Render free-tier service automatically sleeps after 15 minutes of inactivity and wakes on the next request. There is no cost while it sleeps.
- Your GitHub repository remains accessible at no cost.
- Your local project files stay on your Desktop.
Pause - I'll come back to this later
The Render free tier auto-sleeps after 15 minutes of inactivity. There is nothing to pause manually.
- Your service will spin down on its own and resume when someone visits the URL.
- Your local files and GitHub repo remain intact for when you return.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each step below.
- Delete the Render web service: open the Render Dashboard, click your web service to open it, click Settings in the service's left panel, scroll to the bottom of the page, click Delete Web Service, type the service name to confirm, then click Delete.
- Delete the GitHub repository: open your repository on GitHub, click Settings, scroll down to the Danger Zone section, click Delete this repository, type the repository name to confirm, then click Delete this repository.
- Delete your local project files by running this command:
cd ~/Desktop
rm -rf groundops-consoleWhat about the SQLite database?
The SQLite database file on Render is ephemeral. It resets on every redeploy and is automatically removed when you delete the web service. No separate cleanup is needed.
That's a Wrap!
That's a Wrap!
Nice work! You've just built a full orbital operations layer that propagates real satellite orbits, predicts contact windows, schedules tasks against passes, and auto-executes them through your autopilot sequencer.
You've learned how to:
- Build a satellite constellation registry with real SGP4 orbital propagation, custom TEME-to-geodetic coordinate conversion using Vallado's GMST formula and WGS84, and a contact window predictor computing AOS/LOS/max-elevation passes over a ground station.
- Build a task scheduler with greedy bin-packing that queues operational tasks into pass windows, integrates with the autopilot sequencer to auto-execute during contact, and persists all telemetry, commands, and resolver actions to SQLite for post-pass audit.
- Deploy a live mission operations dashboard with a Leaflet map showing satellite positions, a contact timeline, real-time telemetry panel, and log viewer to Render as a shareable URL.
- Secret Mission: Added priority-based task conflict resolution with bumping logic that displaces lower-priority tasks when a high-priority task overflows a pass window.
Ready to quiz yourself?