Mission Operations Scheduler with Yamcs

Build a satellite mission scheduler with resolvers and cockpit configs in Python.

Introduction

30 Second Summary

Satellites orbiting Earth only have brief windows to communicate with ground stations. During these passes, mission controllers need to send the right commands in the right order, adapting in real time to conditions like battery voltage and thermal states.

In this project, you will build a Python-based Mission Operations Scheduler that sequences commands across simulated orbital passes, evaluates live spacecraft telemetry to decide which procedures to run, and manages cockpit configurations for different mission phases.

What You'll Build

A mission operations system that integrates with Yamcs to schedule satellite command activities on a timeline, resolve which procedures to execute based on spacecraft state, and display operational status through a deployed Flask dashboard.

By the end of this project, you'll have:

  • A pass scheduler that creates chained command sequences on the Yamcs Timeline, simulating orbital contact windows.
  • A resolver engine that reads live telemetry and conditionally selects commands based on spacecraft health thresholds.
  • A cockpit configuration system with switchable JSON profiles for commissioning, nominal, and safe-mode mission phases, plus an auto-pilot controller that ties everything together.
  • A Flask dashboard deployed on Render displaying timeline state, resolver decisions, and active configuration, alongside a recorded demo video showing the full system running against live Yamcs.
  • Secret Mission: Add a "Test as We Fly" validation mode that dry-runs the auto-pilot against simulated fault injection scenarios and generates a pass/fail verification report.

Are there any prerequisites?

This project builds on your completed Project 1 setup, where you configured Yamcs in Docker and built a telemetry monitoring system. You should already have Docker Desktop, Python 3.10, Cursor, Git, and a GitHub account ready to go.

Before We Start

Before you can schedule satellite commands and build resolver logic, you need a working development environment connected to a live Yamcs instance.

In this step, you will start the Yamcs simulator from Project 1, scaffold a new Python project with all required dependencies, and verify that your code can talk to the spacecraft simulator.

In this step, get ready to:
  • Start Yamcs and confirm the web UI is accessible.
  • Create a project directory with a virtual environment and install dependencies.
  • Verify connectivity to Yamcs by listing available commands.
Start the Yamcs Docker container

Your Yamcs simulator runs inside a Docker container you set up in Project 1. You need it running for the scheduler and resolver modules to have a live mission control system to talk to.

  • Open Docker Desktop by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar, then typing Docker Desktop and pressing Enter.
  • Wait for Docker Desktop to fully start. You should see a green indicator showing the engine is running.
  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Terminal (macOS) or PowerShell (Windows) and pressing Enter.
  • Start the Yamcs container from Project 1 by running this command:
docker start yamcs

What does this command do?

This restarts the stopped yamcs container you created in Project 1. The container runs the Yamcs mission control server with a built-in spacecraft simulator on port 8090.

  • Verify that Yamcs is running by opening your browser and navigating to:
http://localhost:8090
  • You should see the Yamcs web interface with the simulator instance listed.

Don't see the Yamcs web UI?

If docker start yamcs returned an error saying no such container, the container was removed. Recreate it by running:

docker run -d --name yamcs -p 8090:8090 yamcs/example-simulation:5.13.0

Wait about 10 seconds after the container starts, then refresh your browser at localhost:8090.

Help me troubleshoot this issue.

Create the project directory and virtual environment

Each project gets its own folder and isolated Python environment. This keeps dependencies separate from your system Python and other projects.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir mission-ops
cd mission-ops
  • Create a Python virtual environment by running this command:
python3 -m venv venv
  • Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate
  • You should see (venv) appear at the start of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

On Windows, if you see an error about execution policies, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned first, then try the activate command again.

On macOS, make sure you typed source before the path. Without it, the shell does not execute the activation script.

Help me activate my virtual environment.

Install dependencies and create the project structure

Your project uses four Python packages. The yamcs-client library connects to the mission control server. Flask serves the web dashboard. The WSGI servers (waitress for local Windows development, gunicorn for deployment) run the Flask app in production mode.

  • Create the requirements.txt file inside your mission-ops folder by running this command:

macOS

cat > requirements.txt << 'EOF'
yamcs-client==2.0.0
Flask==3.1.3
waitress==3.0.2
gunicorn==26.0.0
EOF

Windows

@"
yamcs-client==2.0.0
Flask==3.1.3
waitress==3.0.2
gunicorn==26.0.0
"@ | Out-File -Encoding utf8 requirements.txt
  • Install the project dependencies by running:
pip install -r requirements.txt

What are these packages?

The yamcs-client package (2.0.0) is the official Python client for Yamcs mission control. It provides Timeline, MDB, and Processor APIs you will use to schedule activities and read telemetry.

Flask (3.1.3) is a lightweight web framework for building the cockpit dashboard. Waitress (3.0.2) is a Windows-compatible WSGI server for local development. Gunicorn (26.0.0) is the UNIX WSGI server used on Render for deployment.

  • You should see Successfully installed followed by a list of packages in your terminal output.

Seeing installation errors?

If pip is not found, make sure your virtual environment is active. You should see (venv) in your prompt.

If a package fails to download, check your internet connection and try running the install command again.

Help me fix this installation error.

  • Create the project file structure by running these commands:

macOS

mkdir configs logs
touch scheduler.py resolvers.py autopilot.py app.py

Windows

mkdir configs, logs
New-Item scheduler.py, resolvers.py, autopilot.py, app.py
  • Confirm the files and folders exist by running:
ls
  • You should see app.py, autopilot.py, configs, logs, requirements.txt, resolvers.py, scheduler.py, and venv in the output.

What is this file structure?

Each Python file maps to a major system component. scheduler.py handles timeline pass scheduling. resolvers.py evaluates telemetry to decide which commands to run. autopilot.py combines them into an autonomous loop.

The app.py file is the Flask dashboard. The configs/ directory will hold mission phase profiles, and logs/ stores autopilot audit trails.

✔️ Awesome, I've got everything!

Great. Double check you have saved requirements.txt and that all files appear in your ls output.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt for comparison:

yamcs-client==2.0.0
Flask==3.1.3
waitress==3.0.2
gunicorn==26.0.0
Verify connectivity to Yamcs

Before building the scheduler and resolver modules, you need to confirm that your Python environment can successfully connect to the Yamcs simulator and query its Mission Database.

  • Open Cursor by pressing Cmd+Space (macOS) or the Windows key (Windows), then typing Cursor and pressing Enter.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the mission-ops folder on your Desktop, then click Open.
  • Create a new file called test_connection.py in the mission-ops folder by right-clicking in the Explorer sidebar and selecting New File.
  • Paste the following test script into test_connection.py:
from yamcs.client import YamcsClient

# Connect to the Yamcs server running in Docker
client = YamcsClient("localhost:8090")

# Access the Mission Database for the simulator instance
mdb = client.get_mdb(instance="simulator")

# List all available commands in the simulator
print("Available commands:")
print("=" * 40)
for command in mdb.list_commands():
    print(f"  {command.qualified_name}")

print("\nConnection successful!")
print(f"Found commands in the simulator MDB.")

What does this script do?

The script creates a YamcsClient connection to localhost:8090 (where your Docker container is serving Yamcs). It then accesses the Mission Database (MDB) for the simulator instance.

The MDB contains all command and telemetry definitions for the simulated spacecraft. Listing commands confirms that your Python client can reach Yamcs and read its configuration. You will see commands like /YSS/SIMULATOR/SWITCH_VOLTAGE_ON that you will schedule in later steps.

  • Save test_connection.py.
  • Run the test script in your terminal (make sure your virtual environment is still active) by running:
python test_connection.py
  • You should see a list of simulator commands printed in your terminal, followed by Connection successful! at the end.

Seeing a connection error?

If you see ConnectionRefusedError, your Yamcs Docker container may not be running. Run docker ps to check if the yamcs container is listed. If not, start it with docker start yamcs.

If you see ModuleNotFoundError: No module named 'yamcs', your virtual environment is not active. Activate it again, then rerun the script.

Help me connect to Yamcs.

Your development environment is ready and verified. Next up, you will use the yamcs-client Timeline API to build a pass scheduler that chains command activities across simulated orbital passes.

Build the Pass Scheduler

In the previous step, you set up your project environment and confirmed that your Python code can reach the Yamcs simulator. Now it's time to build the core scheduling engine that sequences satellite command activities across orbital passes.

Satellites communicate with ground stations during brief contact windows called passes. During each pass, the ground system needs to execute a precise sequence of commands in order. If one command fails, subsequent commands should not fire. The Yamcs Timeline API handles this through activity chaining, where each activity only starts after its predecessor succeeds.

In this step, get ready to:
  • Build a pass scheduler module that chains command activities using the Yamcs Timeline API.
  • Define a nominal mission configuration with a pass template specifying which commands to execute.
  • Load pass windows from a JSON file and schedule multiple passes on the Yamcs timeline.
Create the pass scheduler core module

The scheduler module connects to Yamcs and creates TimelineActivity objects that represent satellite commands. Each activity is chained to the previous one using OnSuccess triggers, so the next command only fires after the preceding one completes successfully.

  • Open scheduler.py in Cursor and replace any placeholder content with the following imports and connection setup:
import json
from datetime import datetime, timedelta, timezone

from yamcs.client import (
    CommandActivity,
    OnSuccess,
    TimelineActivity,
    YamcsClient,
)

# Yamcs connection settings (must match your running Docker container)
YAMCS_URL = "localhost:8090"
YAMCS_INSTANCE = "simulator"


def get_timeline_client():
    # Connect to Yamcs and return a timeline client for scheduling activities
    client = YamcsClient(YAMCS_URL)
    return client.get_timeline_client(instance=YAMCS_INSTANCE)

What does this code do?

The imports bring in the Yamcs client classes you need for timeline scheduling. CommandActivity wraps a spacecraft command. OnSuccess is the trigger that chains activities together. TimelineActivity represents a scheduled action on the Yamcs timeline.

The get_timeline_client() function creates a reusable connection to the Yamcs timeline service on your local Docker instance.

Why timeline-based scheduling instead of direct commanding?

The Yamcs Timeline API provides native activity chaining with triggers like OnSuccess and OnFailure. This maps directly to how real mission operations work: planned activities on a timeline where each step depends on the outcome of the previous one.

Direct commanding would require you to manually poll for completion and handle sequencing yourself. The Timeline API handles all of that automatically.

Now you need the core function that builds a sequence of chained activities from a list of command definitions.

  • Add the following function below get_timeline_client() in scheduler.py:
def create_pass_schedule(pass_name, start_time, activities_def):
    timeline = get_timeline_client()
    saved_items = []
    previous_item = None

    for i, act in enumerate(activities_def):
        # First activity starts at the given time, rest chain via OnSuccess
        if previous_item is None:
            start = start_time
        else:
            start = OnSuccess(previous_item)

        # Build the timeline activity with a command payload
        item = TimelineActivity(
            name=f"{pass_name}/{act['name']}",
            start=start,
            duration=timedelta(seconds=act.get("duration", 30)),
            activity=CommandActivity(
                command=act["command"],
                args=act.get("args", {}),
            ),
            tags=["pass-schedule", pass_name],
        )
        timeline.save_item(item)
        saved_items.append(item)
        previous_item = item

    return saved_items

What does this code do?

This function takes a pass name, a start time, and a list of activity definitions. It loops through each activity and creates a TimelineActivity on the Yamcs timeline.

  • The first activity in the sequence starts at the provided start_time (a fixed datetime).
  • Every subsequent activity uses OnSuccess(previous_item) as its start trigger, creating a chain where each command only fires if the prior one succeeded.
  • Each activity is tagged with pass-schedule and the pass name so you can filter them in the Yamcs UI.
  • Save scheduler.py.

Seeing import errors?

Make sure your virtual environment is activated. If you see ModuleNotFoundError: No module named 'yamcs', the yamcs-client package is not installed in the active environment.

Re-activate your virtual environment and confirm with pip list | grep yamcs. Help me fix my yamcs-client import error in scheduler.py

Define the nominal mission configuration

The pass scheduler needs to know which commands to execute during a pass. You define this as a configuration profile that specifies the command sequence, thresholds, and autopilot settings for a given mission phase.

  • Create a new file called nominal.json inside the configs/ directory with the following content:
{
  "phase": "nominal",
  "description": "Standard mission operations with full automation",
  "monitoring": {
    "parameters": [
      "/YSS/SIMULATOR/BatteryVoltage1",
      "/YSS/SIMULATOR/BatteryVoltage2",
      "/YSS/SIMULATOR/BatteryVoltage3"
    ],
    "polling_interval_seconds": 10
  },
  "thresholds": {
    "battery_voltage_critical": 45.0,
    "battery_voltage_low": 50.0
  },
  "resolvers_enabled": ["power", "thermal"],
  "autopilot": {
    "enabled": true,
    "mode": "autonomous",
    "max_commands_per_pass": 10
  },
  "pass_template": [
    {"name": "power-on-instruments", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 1}, "duration": 10},
    {"name": "configure-heater", "command": "/YSS/SIMULATOR/SET_HEATER", "args": {"heater_id": 1, "temperature": 25.0, "duration": 120}, "duration": 5},
    {"name": "power-on-payload", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 2}, "duration": 10}
  ]
}

What does this configuration define?

This is the "nominal" mission phase profile. It defines everything the scheduler and resolvers need to operate during standard operations.

  • The pass_template array is the key piece for this step. It defines three chained commands: power on instruments, configure the heater, then power on the payload.
  • The thresholds and resolvers_enabled fields will be used in the next step when you build the resolver engine.
  • The autopilot section controls whether the system runs autonomously. You will use this in a later step.

Now add the function that reads the pass template from a config and passes it to the scheduler.

  • Switch back to scheduler.py and add this function below create_pass_schedule():
def schedule_from_config(config, pass_name, start_time):
    # Extract the pass template from config and schedule it
    return create_pass_schedule(pass_name, start_time, config["pass_template"])
  • Save configs/nominal.json and scheduler.py.
  • You won't see a change yet. The next chunk adds the pass loader and runs the full scheduler.
Load pass windows and run the scheduler

Real satellite operations schedule commands across multiple ground station contact windows throughout the day. You will define these windows in a JSON file, then build a loader function that schedules activities for each pass.

  • Create a new file called passes.json in the project root (next to scheduler.py) with the following content:
[
  {"name": "Morning-Pass-1", "offset_minutes": 5},
  {"name": "Afternoon-Pass-2", "offset_minutes": 95},
  {"name": "Evening-Pass-3", "offset_minutes": 185}
]

What do these pass windows represent?

Each entry simulates a ground station contact window. The offset_minutes field is the number of minutes from "now" when the pass begins. In a real system, these would be computed from orbital mechanics.

Now add the function that reads these pass windows and schedules activities for each one.

  • In scheduler.py, add the following function below schedule_from_config():
def schedule_passes_from_file(pass_file, config):
    # Load pass windows from JSON and schedule activities for each one
    with open(pass_file, "r") as f:
        windows = json.load(f)

    now = datetime.now(tz=timezone.utc)
    results = []

    for window in windows:
        offset = window.get("offset_minutes", 0)
        start = now + timedelta(minutes=offset)
        name = window["name"]
        items = schedule_from_config(config, name, start)
        results.append({
            "pass_name": name,
            "start": start.isoformat(),
            "activities_scheduled": len(items),
        })
        print(f"Scheduled {name} at {start.isoformat()} ({len(items)} activities)")

    return results

What does this code do?

This function reads pass windows from a JSON file and iterates through each one. For each pass, it calculates the start time by adding the offset to the current UTC time, then calls schedule_from_config() to create the chained activities on the Yamcs timeline.

The function returns a summary of what it scheduled, which is useful for logging and for the Flask dashboard later.

Finally, add the main block that ties everything together so you can run the scheduler directly.

  • Add the following at the bottom of scheduler.py:
if __name__ == "__main__":
    import os
    config_path = os.path.join("configs", "nominal.json")
    with open(config_path, "r") as f:
        config = json.load(f)

    print("Mission Operations Pass Scheduler")
    print("=" * 40)
    results = schedule_passes_from_file("passes.json", config)
    print(f"\nScheduled {len(results)} passes successfully.")
  • Save scheduler.py and passes.json.

Now run the scheduler to verify everything works against your live Yamcs instance.

  • Run the scheduler by executing this command in your terminal:
python scheduler.py

You should see output similar to this:

Mission Operations Pass Scheduler
========================================
Scheduled Morning-Pass-1 at 2026-06-17T10:05:00+00:00 (3 activities)
Scheduled Afternoon-Pass-2 at 2026-06-17T11:35:00+00:00 (3 activities)
Scheduled Evening-Pass-3 at 2026-06-17T13:05:00+00:00 (3 activities)

Scheduled 3 passes successfully.
  • Open the Yamcs web UI at http://localhost:8090 in your browser.
  • Navigate to **Timeline** in the left sidebar.
  • You should see your scheduled activities listed with names like Morning-Pass-1/power-on-instruments.

Seeing a ConnectionFailure error?

This means the Yamcs Docker container is not running or not reachable. Verify your container is up by running docker ps and checking for the yamcs container.

If the container is not listed, start it with docker start yamcs and wait about 10 seconds for the server to initialize before running the scheduler again. Help me fix my Yamcs connection error when running scheduler.py

✔️ Awesome, I've got everything!

Great. Make sure you have saved all three files: scheduler.py, passes.json, and configs/nominal.json.

ⓧ I'd like to double check the full code

Here is the complete scheduler.py file:

import json
from datetime import datetime, timedelta, timezone

from yamcs.client import (
    CommandActivity,
    OnSuccess,
    TimelineActivity,
    YamcsClient,
)

YAMCS_URL = "localhost:8090"
YAMCS_INSTANCE = "simulator"


def get_timeline_client():
    client = YamcsClient(YAMCS_URL)
    return client.get_timeline_client(instance=YAMCS_INSTANCE)


def create_pass_schedule(pass_name, start_time, activities_def):
    timeline = get_timeline_client()
    saved_items = []
    previous_item = None

    for i, act in enumerate(activities_def):
        if previous_item is None:
            start = start_time
        else:
            start = OnSuccess(previous_item)

        item = TimelineActivity(
            name=f"{pass_name}/{act['name']}",
            start=start,
            duration=timedelta(seconds=act.get("duration", 30)),
            activity=CommandActivity(
                command=act["command"],
                args=act.get("args", {}),
            ),
            tags=["pass-schedule", pass_name],
        )
        timeline.save_item(item)
        saved_items.append(item)
        previous_item = item

    return saved_items


def schedule_from_config(config, pass_name, start_time):
    return create_pass_schedule(pass_name, start_time, config["pass_template"])


def schedule_passes_from_file(pass_file, config):
    with open(pass_file, "r") as f:
        windows = json.load(f)

    now = datetime.now(tz=timezone.utc)
    results = []

    for window in windows:
        offset = window.get("offset_minutes", 0)
        start = now + timedelta(minutes=offset)
        name = window["name"]
        items = schedule_from_config(config, name, start)
        results.append({
            "pass_name": name,
            "start": start.isoformat(),
            "activities_scheduled": len(items),
        })
        print(f"Scheduled {name} at {start.isoformat()} ({len(items)} activities)")

    return results


if __name__ == "__main__":
    import os
    config_path = os.path.join("configs", "nominal.json")
    with open(config_path, "r") as f:
        config = json.load(f)

    print("Mission Operations Pass Scheduler")
    print("=" * 40)
    results = schedule_passes_from_file("passes.json", config)
    print(f"\nScheduled {len(results)} passes successfully.")

Here is the complete passes.json file:

[
  {"name": "Morning-Pass-1", "offset_minutes": 5},
  {"name": "Afternoon-Pass-2", "offset_minutes": 95},
  {"name": "Evening-Pass-3", "offset_minutes": 185}
]

Here is the complete configs/nominal.json file:

{
  "phase": "nominal",
  "description": "Standard mission operations with full automation",
  "monitoring": {
    "parameters": [
      "/YSS/SIMULATOR/BatteryVoltage1",
      "/YSS/SIMULATOR/BatteryVoltage2",
      "/YSS/SIMULATOR/BatteryVoltage3"
    ],
    "polling_interval_seconds": 10
  },
  "thresholds": {
    "battery_voltage_critical": 45.0,
    "battery_voltage_low": 50.0
  },
  "resolvers_enabled": ["power", "thermal"],
  "autopilot": {
    "enabled": true,
    "mode": "autonomous",
    "max_commands_per_pass": 10
  },
  "pass_template": [
    {"name": "power-on-instruments", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 1}, "duration": 10},
    {"name": "configure-heater", "command": "/YSS/SIMULATOR/SET_HEATER", "args": {"heater_id": 1, "temperature": 25.0, "duration": 120}, "duration": 5},
    {"name": "power-on-payload", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 2}, "duration": 10}
  ]
}

Your pass scheduler is working. You now have a module that can sequence any list of commands across orbital passes on the Yamcs timeline. Next up, you will build the resolver engine that reads live telemetry and decides which commands are safe to execute before each pass.

Implement Resolvers

Your pass scheduler can now sequence command activities on the Yamcs timeline. But a real mission operations system does not blindly execute every pass plan.

Before issuing commands, a flight operations team checks current spacecraft state to decide what is safe to execute. In this step, you will build a resolver engine that reads live telemetry from the simulator and conditionally selects which command procedures to run based on battery voltage thresholds.

In this step, get ready to:
  • Create a resolver module that subscribes to live telemetry parameters and evaluates threshold rules.
  • Implement decision logic that returns different command sequences based on spacecraft health state.
  • Connect the resolver to the cockpit configuration so it evaluates current state and logs decisions with timestamps.
Create the resolver module with telemetry subscription

The scheduler schedules commands, but resolvers decide whether those commands are safe to execute. A resolver reads live telemetry values from the spacecraft and compares them against threshold rules defined in the cockpit configuration.

  • Create a new file called resolvers.py in your project folder (the same folder as scheduler.py).
  • Add the imports, connection constants, and telemetry reading functions by pasting this code:
from datetime import datetime, timezone

from yamcs.client import YamcsClient

YAMCS_URL = "localhost:8090"
YAMCS_INSTANCE = "simulator"


def get_processor():
    client = YamcsClient(YAMCS_URL)
    return client.get_processor(instance=YAMCS_INSTANCE, processor="realtime")


def get_spacecraft_state():
    processor = get_processor()
    params = [
        "/YSS/SIMULATOR/BatteryVoltage1",
        "/YSS/SIMULATOR/BatteryVoltage2",
        "/YSS/SIMULATOR/BatteryVoltage3",
    ]
    state = {}
    for p in params:
        try:
            val = processor.get_parameter_value(p)
            state[p] = val.eng_value if val else None
        except Exception:
            state[p] = None
    return state

What does this code do?

get_processor() connects to the Yamcs realtime processor. This is the live data-processing pipeline that receives telemetry from the simulated spacecraft.

get_spacecraft_state() reads the current engineering values (calibrated voltage readings) from three battery parameters. It wraps each read in a try/except because telemetry might be temporarily unavailable. The result is a dictionary mapping parameter paths to their current voltage values.

  • Save resolvers.py.

You will not see output yet. The next chunk adds the decision logic that uses this telemetry data.

Implement the power resolver decision logic

The power resolver is the core safety gate. It computes the average battery voltage and returns a different action depending on whether the spacecraft is in a critical, low, or nominal state.

  • In resolvers.py, add the resolve_power() function below the get_spacecraft_state() function:
def resolve_power(state, thresholds):
    voltages = [v for v in state.values() if v is not None]
    if not voltages:
        return {
            "action": "HOLD",
            "command": None,
            "args": {},
            "reason": "No telemetry available - holding current state",
        }

    avg = sum(voltages) / len(voltages)

    if avg < thresholds["battery_voltage_critical"]:
        return {
            "action": "SAFE_MODE",
            "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_OFF",
            "args": {"voltage_num": 2},
            "reason": f"Critical voltage ({avg:.1f}V) - shedding payload",
        }
    elif avg < thresholds["battery_voltage_low"]:
        return {
            "action": "REDUCE_POWER",
            "command": "/YSS/SIMULATOR/SET_HEATER",
            "args": {"heater_id": 1, "temperature": 15.0, "duration": 60},
            "reason": f"Low voltage ({avg:.1f}V) - reducing heater load",
        }
    else:
        return {
            "action": "NOMINAL",
            "command": None,
            "args": {},
            "reason": f"Nominal voltage ({avg:.1f}V) - no action needed",
        }

What does this code do?

resolve_power() takes the current spacecraft state and the threshold values from the cockpit config. It averages all available battery voltages and makes a tiered decision.

  • If no telemetry is available, it returns HOLD (do nothing until data returns).
  • If average voltage drops below the critical threshold (45.0V in nominal config), it returns SAFE_MODE and sheds the payload by powering it off.
  • If voltage is below the low threshold (50.0V) but above critical, it returns REDUCE_POWER and reduces heater load.
  • Otherwise, voltage is nominal. No corrective action needed.
  • Save resolvers.py.

You will not see a change yet. The next chunk adds the thermal resolver.

  • Below resolve_power(), add the resolve_thermal() function:
def resolve_thermal(state, thresholds):
    voltages = [v for v in state.values() if v is not None]
    if not voltages:
        return {
            "action": "HOLD",
            "command": None,
            "args": {},
            "reason": "No telemetry for thermal assessment",
        }

    max_v = max(voltages)
    if max_v > 58.0:
        return {
            "action": "COOL_DOWN",
            "command": "/YSS/SIMULATOR/SET_HEATER",
            "args": {"heater_id": 2, "temperature": 10.0, "duration": 180},
            "reason": f"Elevated readings ({max_v:.1f}V) - reducing thermal load",
        }
    return {
        "action": "NOMINAL",
        "command": None,
        "args": {},
        "reason": f"Thermal nominal ({max_v:.1f}V)",
    }

What does this code do?

resolve_thermal() checks whether any single battery reading is elevated above 58.0V. If so, it returns a COOL_DOWN action to reduce thermal load on heater 2. Otherwise it reports nominal.

This is a simpler resolver than the power one. In a real spacecraft system, thermal resolvers would check dedicated temperature sensors. Here you are using voltage readings as a proxy since the simulator exposes battery parameters.

  • Save resolvers.py.

You will not see output yet. The next chunk wires both resolvers into the cockpit config.

Seeing a syntax error?

Make sure there is a blank line between the end of resolve_power() and the def resolve_thermal line. Python requires blank lines between top-level function definitions.

help me fix a syntax error in my resolvers.py file

Connect the resolver to the cockpit configuration

Individual resolvers do not know which config is active. The run_resolvers() function bridges this gap by reading which resolvers are enabled in the current cockpit configuration, running each one, and packaging the decisions with a timestamp.

  • Below resolve_thermal(), add the run_resolvers() function:
def run_resolvers(config):
    state = get_spacecraft_state()
    thresholds = config["thresholds"]
    enabled = config.get("resolvers_enabled", [])

    decisions = {"timestamp": datetime.now(tz=timezone.utc).isoformat(), "state": state, "resolutions": {}}

    if "power" in enabled:
        decisions["resolutions"]["power"] = resolve_power(state, thresholds)
    if "thermal" in enabled:
        decisions["resolutions"]["thermal"] = resolve_thermal(state, thresholds)

    return decisions

What does this code do?

run_resolvers() is the integration point. It takes the active cockpit configuration, fetches live telemetry via get_spacecraft_state(), then conditionally runs only the resolvers listed in resolvers_enabled.

The nominal config enables both power and thermal. A safe-mode config might enable only power. This is how different mission phases get different resolution behaviour without changing any code.

  • Save resolvers.py.

You will not see output yet. The next chunk adds a test harness so you can run the module directly.

  • At the bottom of resolvers.py, add the __main__ block to run the resolvers against live Yamcs telemetry:
if __name__ == "__main__":
    import json, os
    with open(os.path.join("configs", "nominal.json"), "r") as f:
        config = json.load(f)

    print("Mission Resolvers")
    print("=" * 40)
    decisions = run_resolvers(config)
    print(f"Timestamp: {decisions['timestamp']}")
    print(f"\nState: {decisions['state']}")
    for name, res in decisions["resolutions"].items():
        print(f"\n[{name.upper()}] {res['action']}: {res['reason']}")

What does this code do?

The __main__ block loads the nominal cockpit configuration and calls run_resolvers(). It prints the raw spacecraft state (voltage readings from the simulator) and each resolver's decision with its action and reasoning.

  • Save resolvers.py.
  • Make sure your Yamcs Docker container is still running.
  • Run the resolver module by running this command:
python resolvers.py

You should see output showing the current battery voltage readings and the resolver decisions. It will look something like this:

Mission Resolvers
========================================
Timestamp: 2026-06-17T10:23:45.123456+00:00

State: {'/YSS/SIMULATOR/BatteryVoltage1': 53.2, ...}

[POWER] NOMINAL: Nominal voltage (53.4V) - no action needed

[THERMAL] NOMINAL: Thermal nominal (54.1V)

What should I see?

The exact voltage values will differ because the simulator generates slightly varying readings each time. The important thing is that you see a timestamp, state values for all three battery parameters, and resolver decisions for both POWER and THERMAL.

Under normal simulator conditions, both resolvers return NOMINAL because the simulated voltages hover around 50-55V, which is above both thresholds.

Connection refused or timeout?

This means the Yamcs Docker container is not running. Start it with docker start yamcs and wait about 10 seconds for the simulator to initialize before retrying.

help me debug a connection error when running resolvers.py against my Yamcs Docker container

✔️ Awesome, I've got everything!

Great. Double check you have saved resolvers.py and that it ran successfully against live Yamcs.

ⓧ I'd like to double check the full code

Here is the complete resolvers.py file:

from datetime import datetime, timezone

from yamcs.client import YamcsClient

YAMCS_URL = "localhost:8090"
YAMCS_INSTANCE = "simulator"


def get_processor():
    client = YamcsClient(YAMCS_URL)
    return client.get_processor(instance=YAMCS_INSTANCE, processor="realtime")


def get_spacecraft_state():
    processor = get_processor()
    params = [
        "/YSS/SIMULATOR/BatteryVoltage1",
        "/YSS/SIMULATOR/BatteryVoltage2",
        "/YSS/SIMULATOR/BatteryVoltage3",
    ]
    state = {}
    for p in params:
        try:
            val = processor.get_parameter_value(p)
            state[p] = val.eng_value if val else None
        except Exception:
            state[p] = None
    return state


def resolve_power(state, thresholds):
    voltages = [v for v in state.values() if v is not None]
    if not voltages:
        return {
            "action": "HOLD",
            "command": None,
            "args": {},
            "reason": "No telemetry available - holding current state",
        }

    avg = sum(voltages) / len(voltages)

    if avg < thresholds["battery_voltage_critical"]:
        return {
            "action": "SAFE_MODE",
            "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_OFF",
            "args": {"voltage_num": 2},
            "reason": f"Critical voltage ({avg:.1f}V) - shedding payload",
        }
    elif avg < thresholds["battery_voltage_low"]:
        return {
            "action": "REDUCE_POWER",
            "command": "/YSS/SIMULATOR/SET_HEATER",
            "args": {"heater_id": 1, "temperature": 15.0, "duration": 60},
            "reason": f"Low voltage ({avg:.1f}V) - reducing heater load",
        }
    else:
        return {
            "action": "NOMINAL",
            "command": None,
            "args": {},
            "reason": f"Nominal voltage ({avg:.1f}V) - no action needed",
        }


def resolve_thermal(state, thresholds):
    voltages = [v for v in state.values() if v is not None]
    if not voltages:
        return {
            "action": "HOLD",
            "command": None,
            "args": {},
            "reason": "No telemetry for thermal assessment",
        }

    max_v = max(voltages)
    if max_v > 58.0:
        return {
            "action": "COOL_DOWN",
            "command": "/YSS/SIMULATOR/SET_HEATER",
            "args": {"heater_id": 2, "temperature": 10.0, "duration": 180},
            "reason": f"Elevated readings ({max_v:.1f}V) - reducing thermal load",
        }
    return {
        "action": "NOMINAL",
        "command": None,
        "args": {},
        "reason": f"Thermal nominal ({max_v:.1f}V)",
    }


def run_resolvers(config):
    state = get_spacecraft_state()
    thresholds = config["thresholds"]
    enabled = config.get("resolvers_enabled", [])

    decisions = {"timestamp": datetime.now(tz=timezone.utc).isoformat(), "state": state, "resolutions": {}}

    if "power" in enabled:
        decisions["resolutions"]["power"] = resolve_power(state, thresholds)
    if "thermal" in enabled:
        decisions["resolutions"]["thermal"] = resolve_thermal(state, thresholds)

    return decisions


if __name__ == "__main__":
    import json, os
    with open(os.path.join("configs", "nominal.json"), "r") as f:
        config = json.load(f)

    print("Mission Resolvers")
    print("=" * 40)
    decisions = run_resolvers(config)
    print(f"Timestamp: {decisions['timestamp']}")
    print(f"\nState: {decisions['state']}")
    for name, res in decisions["resolutions"].items():
        print(f"\n[{name.upper()}] {res['action']}: {res['reason']}")

Your resolvers are live and reading telemetry from the simulated spacecraft. Next up, you will build cockpit configuration profiles that let you switch between mission phases and control which resolvers and pass plans are active.

Script Cockpit Configurations

Your scheduler knows how to plan passes and your resolvers know how to evaluate spacecraft health. The missing piece is telling them how to behave during different mission phases.

Real satellite missions cycle through distinct operational phases: commissioning (initial checkout), nominal (full operations), and safe mode (emergency power conservation). Each phase changes which resolvers run, how aggressively the autopilot acts, and which commands are allowed. In this step, you will define those phase profiles as JSON configs and build a Flask dashboard to view them at runtime.

In this step, get ready to:
  • Define cockpit configuration profiles for commissioning, nominal, and safe-mode mission phases.
  • Build a configuration manager that loads and serves profiles through Flask API routes.
  • Build a web dashboard that displays configurations, autopilot logs, and system status.
Define cockpit configuration profiles

Each configuration profile is a JSON file that specifies which telemetry parameters to monitor, what voltage thresholds trigger resolver actions, which resolvers are enabled, and what commands the autopilot is allowed to run.

You already have configs/nominal.json from the previous step. Now you will create two more profiles for the other mission phases.

  • Create a new file called commissioning.json inside your configs folder, then paste the following:
{
  "phase": "commissioning",
  "description": "Initial spacecraft checkout and subsystem verification",
  "monitoring": {
    "parameters": [
      "/YSS/SIMULATOR/BatteryVoltage1",
      "/YSS/SIMULATOR/BatteryVoltage2",
      "/YSS/SIMULATOR/BatteryVoltage3"
    ],
    "polling_interval_seconds": 5
  },
  "thresholds": {
    "battery_voltage_critical": 45.0,
    "battery_voltage_low": 50.0
  },
  "resolvers_enabled": ["power"],
  "autopilot": {
    "enabled": false,
    "mode": "manual",
    "max_commands_per_pass": 2
  },
  "pass_template": [
    {"name": "power-on-bus", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 1}, "duration": 10},
    {"name": "checkout-payload", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_ON", "args": {"voltage_num": 2}, "duration": 10}
  ]
}

What makes commissioning different?

During commissioning, the spacecraft is being checked out for the first time. Autopilot is disabled ("enabled": false) so operators maintain manual control.

Only the power resolver is active, and the pass template is limited to two basic power-on commands. The polling interval is faster (5 seconds) because operators need real-time feedback during initial checkout.

  • Save configs/commissioning.json.
  • Create another new file called safe_mode.json inside the same configs folder, then paste the following:
{
  "phase": "safe_mode",
  "description": "Minimum power state - essential systems only",
  "monitoring": {
    "parameters": [
      "/YSS/SIMULATOR/BatteryVoltage1"
    ],
    "polling_interval_seconds": 2
  },
  "thresholds": {
    "battery_voltage_critical": 42.0,
    "battery_voltage_low": 48.0
  },
  "resolvers_enabled": ["power"],
  "autopilot": {
    "enabled": true,
    "mode": "safe",
    "max_commands_per_pass": 3
  },
  "pass_template": [
    {"name": "shed-payload", "command": "/YSS/SIMULATOR/SWITCH_VOLTAGE_OFF", "args": {"voltage_num": 2}, "duration": 10}
  ]
}

What makes safe mode different?

Safe mode is the spacecraft's emergency state. It monitors only BatteryVoltage1 (the primary bus) and polls every 2 seconds for rapid response.

The thresholds are lowered (critical at 42V instead of 45V) because the spacecraft is already in a degraded state. The only command in the pass template is shed-payload which powers off non-essential systems to conserve battery.

  • Save configs/safe_mode.json.
  • Confirm all three config files exist by running:
ls configs/

You should see commissioning.json, nominal.json, and safe_mode.json listed.

Missing a config file?

Make sure you created the files inside the configs/ folder, not in the project root. The filenames must be exactly commissioning.json and safe_mode.json (with an underscore, not a hyphen).

help me find where my config files ended up

Build the configuration manager with Flask routes

The configuration manager is a Flask application that reads your JSON profiles from disk and serves them through API endpoints. This gives the autopilot (and eventually your dashboard) a runtime interface to load and inspect configurations.

  • Create a new file called app.py in your project root folder, then paste the following code:
import json
import os

from flask import Flask, jsonify, render_template_string, request

app = Flask(__name__)

CONFIGS_DIR = "configs"
LOGS_DIR = "logs"


@app.route("/api/configs")
def api_configs():
    configs = []
    if os.path.isdir(CONFIGS_DIR):
        for f in sorted(os.listdir(CONFIGS_DIR)):
            if f.endswith(".json"):
                with open(os.path.join(CONFIGS_DIR, f), "r") as fh:
                    data = json.load(fh)
                configs.append({
                    "name": f.replace(".json", ""),
                    "phase": data.get("phase", "unknown"),
                    "autopilot_enabled": data.get("autopilot", {}).get("enabled", False),
                })
    return jsonify(configs)


@app.route("/api/configs/<name>")
def api_get_config(name):
    path = os.path.join(CONFIGS_DIR, f"{name}.json")
    if not os.path.exists(path):
        return jsonify({"error": f"Config '{name}' not found"}), 404
    with open(path, "r") as f:
        return jsonify(json.load(f))

What does this code do?

The /api/configs route scans the configs/ directory, reads each JSON file, and returns a summary list with the phase name and autopilot status. This is how the dashboard discovers which configurations exist.

The /api/configs/<name> route returns the full JSON content of a specific config file. If the config does not exist, it returns a 404 error.

  • Save app.py. You will not be able to run it yet because the dashboard and main block are not in place.
Build the Flask web dashboard

The dashboard gives operators a visual interface to inspect which configurations are loaded and review autopilot activity logs. It is a single HTML page served by Flask with JavaScript that calls the API routes you just created.

  • In app.py, find the line that says LOGS_DIR = "logs". Copy the following HTML template constant below it, and above the @app.route("/api/configs") line:
DASHBOARD_HTML = """<!DOCTYPE html>
<html>
<head>
    <title>Mission Operations Cockpit</title>
    <style>
        body { font-family: monospace; background: #1a1a2e; color: #e0e0e0; margin: 0; padding: 20px; }
        h1 { color: #00d4aa; border-bottom: 2px solid #00d4aa; padding-bottom: 10px; }
        h2 { color: #4ecdc4; }
        .card { background: #16213e; border: 1px solid #0f3460; border-radius: 8px; padding: 15px; margin: 10px 0; }
        .config-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 15px; }
        pre { background: #0f3460; padding: 10px; border-radius: 4px; overflow-x: auto; font-size: 12px; }
        .badge { display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 11px; }
        .badge-nominal { background: #00ff88; color: #000; }
        .badge-safe { background: #ff4444; color: #fff; }
        .badge-commission { background: #ffaa00; color: #000; }
    </style>
</head>
<body>
    <h1>Mission Operations Cockpit</h1>
    <p>Satellite Operations Scheduler and Configuration System</p>
    <h2>Cockpit Configurations</h2>
    <div class="config-grid" id="configs"><p>Loading...</p></div>
    <h2>Auto-Pilot Log (Recent)</h2>
    <div class="card" id="logs"><p>Loading...</p></div>
    <h2>API Endpoints</h2>
    <div class="card">
        <pre>GET  /api/configs          - List all configurations
GET  /api/configs/&lt;name&gt;  - Get specific configuration
GET  /api/logs             - Get autopilot execution logs
GET  /api/status           - System status summary</pre>
    </div>
    <script>
        fetch('/api/configs').then(r=>r.json()).then(data=>{
            document.getElementById('configs').innerHTML = data.map(c=>
                '<div class="card"><h3>'+c.name+'</h3><p>Phase: '+c.phase+'</p><p>Autopilot: '+(c.autopilot_enabled?'Enabled':'Disabled')+'</p></div>'
            ).join('');
        });
        fetch('/api/logs').then(r=>r.json()).then(data=>{
            const el = document.getElementById('logs');
            if(!data.length){el.innerHTML='<p>No logs yet. Run autopilot.py to generate.</p>';return;}
            el.innerHTML='<pre>'+JSON.stringify(data.slice(-5),null,2)+'</pre>';
        });
    </script>
</body>
</html>"""

What does this HTML dashboard do?

The dashboard uses a dark monospace theme styled for mission operations. When the page loads, JavaScript fetches /api/configs to display configuration cards and /api/logs to show recent autopilot decisions.

The API endpoints reference section gives operators a quick reminder of available routes for programmatic access.

  • Save app.py. Now add the remaining routes and the main block. Find the closing line of the api_get_config function and paste the following code below it:
@app.route("/")
def dashboard():
    return render_template_string(DASHBOARD_HTML)


@app.route("/api/logs")
def api_logs():
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    if not os.path.exists(log_file):
        return jsonify([])
    with open(log_file, "r") as f:
        try:
            return jsonify(json.load(f))
        except json.JSONDecodeError:
            return jsonify([])


@app.route("/api/status")
def api_status():
    configs = []
    if os.path.isdir(CONFIGS_DIR):
        configs = [f.replace(".json", "") for f in os.listdir(CONFIGS_DIR) if f.endswith(".json")]
    log_count = 0
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    if os.path.exists(log_file):
        with open(log_file, "r") as f:
            try:
                log_count = len(json.load(f))
            except json.JSONDecodeError:
                pass
    return jsonify({
        "system": "Mission Operations Cockpit",
        "configs_available": configs,
        "log_entries": log_count,
    })


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    if os.name == "nt":
        from waitress import serve
        print(f"Starting on http://localhost:{port} (waitress)")
        serve(app, host="0.0.0.0", port=port)
    else:
        app.run(host="0.0.0.0", port=port)

What do these routes and the main block do?

The / route serves the HTML dashboard template. The /api/logs route reads the autopilot's JSON audit trail from the logs/ directory (which will be populated in the next step).

The /api/status route returns a summary of available configs and log entry count. The main block uses waitress on Windows (since gunicorn is Unix-only) and Flask's built-in server on other platforms.

  • Save app.py.
  • Create the logs directory that the app expects by running:
mkdir logs
  • Start the Flask dashboard by running:
python app.py
  • Open your browser and navigate to http://localhost:5000.

You should see the Mission Operations Cockpit dashboard with a dark background, three configuration cards (commissioning, nominal, safe_mode), and a log section showing "No logs yet."

Dashboard not loading?

If you see a ModuleNotFoundError for Flask or waitress, make sure your virtual environment is activated and that you installed the dependencies from requirements.txt in Step 1.

If the browser shows a connection error, confirm the terminal displays a message with the port number (5000). Make sure no other process is using that port.

help me debug why my Flask dashboard is not starting

  • Press Ctrl+C in your terminal to stop the Flask server after confirming the dashboard works.

✔️ Awesome, I've got everything!

Great. Double check you have saved app.py and that the dashboard displayed all three configuration cards.

ⓧ I'd like to double check the full code

Here is the complete app.py file for comparison:

import json
import os

from flask import Flask, jsonify, render_template_string, request

app = Flask(__name__)

CONFIGS_DIR = "configs"
LOGS_DIR = "logs"

DASHBOARD_HTML = """<!DOCTYPE html>
<html>
<head>
    <title>Mission Operations Cockpit</title>
    <style>
        body { font-family: monospace; background: #1a1a2e; color: #e0e0e0; margin: 0; padding: 20px; }
        h1 { color: #00d4aa; border-bottom: 2px solid #00d4aa; padding-bottom: 10px; }
        h2 { color: #4ecdc4; }
        .card { background: #16213e; border: 1px solid #0f3460; border-radius: 8px; padding: 15px; margin: 10px 0; }
        .config-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 15px; }
        pre { background: #0f3460; padding: 10px; border-radius: 4px; overflow-x: auto; font-size: 12px; }
        .badge { display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 11px; }
        .badge-nominal { background: #00ff88; color: #000; }
        .badge-safe { background: #ff4444; color: #fff; }
        .badge-commission { background: #ffaa00; color: #000; }
    </style>
</head>
<body>
    <h1>Mission Operations Cockpit</h1>
    <p>Satellite Operations Scheduler and Configuration System</p>
    <h2>Cockpit Configurations</h2>
    <div class="config-grid" id="configs"><p>Loading...</p></div>
    <h2>Auto-Pilot Log (Recent)</h2>
    <div class="card" id="logs"><p>Loading...</p></div>
    <h2>API Endpoints</h2>
    <div class="card">
        <pre>GET  /api/configs          - List all configurations
GET  /api/configs/&lt;name&gt;  - Get specific configuration
GET  /api/logs             - Get autopilot execution logs
GET  /api/status           - System status summary</pre>
    </div>
    <script>
        fetch('/api/configs').then(r=>r.json()).then(data=>{
            document.getElementById('configs').innerHTML = data.map(c=>
                '<div class="card"><h3>'+c.name+'</h3><p>Phase: '+c.phase+'</p><p>Autopilot: '+(c.autopilot_enabled?'Enabled':'Disabled')+'</p></div>'
            ).join('');
        });
        fetch('/api/logs').then(r=>r.json()).then(data=>{
            const el = document.getElementById('logs');
            if(!data.length){el.innerHTML='<p>No logs yet. Run autopilot.py to generate.</p>';return;}
            el.innerHTML='<pre>'+JSON.stringify(data.slice(-5),null,2)+'</pre>';
        });
    </script>
</body>
</html>"""


@app.route("/")
def dashboard():
    return render_template_string(DASHBOARD_HTML)


@app.route("/api/configs")
def api_configs():
    configs = []
    if os.path.isdir(CONFIGS_DIR):
        for f in sorted(os.listdir(CONFIGS_DIR)):
            if f.endswith(".json"):
                with open(os.path.join(CONFIGS_DIR, f), "r") as fh:
                    data = json.load(fh)
                configs.append({
                    "name": f.replace(".json", ""),
                    "phase": data.get("phase", "unknown"),
                    "autopilot_enabled": data.get("autopilot", {}).get("enabled", False),
                })
    return jsonify(configs)


@app.route("/api/configs/<name>")
def api_get_config(name):
    path = os.path.join(CONFIGS_DIR, f"{name}.json")
    if not os.path.exists(path):
        return jsonify({"error": f"Config '{name}' not found"}), 404
    with open(path, "r") as f:
        return jsonify(json.load(f))


@app.route("/api/logs")
def api_logs():
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    if not os.path.exists(log_file):
        return jsonify([])
    with open(log_file, "r") as f:
        try:
            return jsonify(json.load(f))
        except json.JSONDecodeError:
            return jsonify([])


@app.route("/api/status")
def api_status():
    configs = []
    if os.path.isdir(CONFIGS_DIR):
        configs = [f.replace(".json", "") for f in os.listdir(CONFIGS_DIR) if f.endswith(".json")]
    log_count = 0
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    if os.path.exists(log_file):
        with open(log_file, "r") as f:
            try:
                log_count = len(json.load(f))
            except json.JSONDecodeError:
                pass
    return jsonify({
        "system": "Mission Operations Cockpit",
        "configs_available": configs,
        "log_entries": log_count,
    })


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    if os.name == "nt":
        from waitress import serve
        print(f"Starting on http://localhost:{port} (waitress)")
        serve(app, host="0.0.0.0", port=port)
    else:
        app.run(host="0.0.0.0", port=port)

Your cockpit configuration system is now live. You have three mission phase profiles that define how the spacecraft should be operated, plus a Flask dashboard that serves them to any operator or automated system. Next up, you will wire the autopilot to consume these configs and make autonomous decisions.

Build the Auto-Pilot and Deploy

You have a scheduler that plans passes, resolvers that evaluate spacecraft state, configs that define mission phases, and a Flask dashboard that displays them all.

The missing piece is the auto-pilot controller that ties everything together into a continuous operations loop. It loads a cockpit config, queries live telemetry through resolvers, schedules corrective commands on the timeline, and writes a JSON audit trail of every decision. Once it works locally, you will deploy the dashboard to Render and record a demo video proving the full system end-to-end.

In this step, get ready to:
  • Build the auto-pilot controller that runs scheduling loops against live Yamcs.
  • Deploy the Flask dashboard to Render on the free tier.
  • Record a 2-3 minute demo video showing the system running end-to-end.
Create the auto-pilot controller

The auto-pilot is the orchestration layer that satellite operators rely on during autonomous passes. It loads the active cockpit config, calls resolvers against live telemetry, and schedules corrective commands on the Yamcs timeline without human intervention.

  • Create a new file called autopilot.py in your project root.
  • Add the imports, constants, and utility functions by pasting the following code:
import json
import os
import time
from datetime import datetime, timedelta, timezone

from yamcs.client import CommandActivity, TimelineActivity

from scheduler import get_timeline_client
from resolvers import run_resolvers

LOGS_DIR = "logs"


def ensure_dirs():
    os.makedirs(LOGS_DIR, exist_ok=True)


def load_config(name):
    with open(os.path.join("configs", f"{name}.json"), "r") as f:
        return json.load(f)

What does this code do?

The imports pull in your existing get_timeline_client from scheduler.py and run_resolvers from resolvers.py. The auto-pilot reuses these rather than duplicating logic.

The ensure_dirs() helper creates the logs/ directory if it does not exist. The load_config() helper reads a named JSON config from the configs/ folder.

  • Below the utility functions, add the audit logging function:
def log_decision(entry):
    ensure_dirs()
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    logs = []
    if os.path.exists(log_file):
        with open(log_file, "r") as f:
            try:
                logs = json.load(f)
            except json.JSONDecodeError:
                logs = []
    logs.append(entry)
    logs = logs[-100:]
    with open(log_file, "w") as f:
        json.dump(logs, f, indent=2)

What does this code do?

Every auto-pilot decision gets appended to logs/autopilot_log.json. The list is capped at 100 entries so the file never grows unbounded. This is the same file your Flask dashboard reads from the /api/logs endpoint.

  • Add the core decision engine that runs one cycle of the auto-pilot:
def execute_cycle(config_name="nominal", dry_run=False):
    config = load_config(config_name)

    if not config["autopilot"]["enabled"]:
        entry = {
            "timestamp": datetime.now(tz=timezone.utc).isoformat(),
            "config": config_name,
            "action": "SKIPPED",
            "reason": "Autopilot disabled in configuration",
            "dry_run": dry_run,
        }
        log_decision(entry)
        return entry

    decisions = run_resolvers(config)
    actions = []

    for resolver_name, resolution in decisions["resolutions"].items():
        if resolution["action"] == "NOMINAL" or resolution["command"] is None:
            continue

        if dry_run:
            actions.append({
                "resolver": resolver_name,
                "action": resolution["action"],
                "command": resolution["command"],
                "reason": resolution["reason"],
                "scheduled": False,
                "dry_run": True,
            })
        else:
            try:
                timeline = get_timeline_client()
                now = datetime.now(tz=timezone.utc)
                item = TimelineActivity(
                    name=f"autopilot/{resolver_name}/{resolution['action']}",
                    start=now + timedelta(seconds=5),
                    duration=timedelta(seconds=30),
                    activity=CommandActivity(
                        command=resolution["command"],
                        args=resolution["args"],
                    ),
                    tags=["autopilot", config_name],
                )
                timeline.save_item(item)
                actions.append({
                    "resolver": resolver_name,
                    "action": resolution["action"],
                    "command": resolution["command"],
                    "reason": resolution["reason"],
                    "scheduled": True,
                })
            except Exception as e:
                actions.append({
                    "resolver": resolver_name,
                    "action": resolution["action"],
                    "error": str(e),
                    "scheduled": False,
                })

    entry = {
        "timestamp": datetime.now(tz=timezone.utc).isoformat(),
        "config": config_name,
        "state": decisions["state"],
        "actions": actions,
        "action": "EXECUTED" if actions else "NO_ACTION",
        "dry_run": dry_run,
    }
    log_decision(entry)
    return entry

What does this code do?

The execute_cycle() function is the heart of the auto-pilot. It loads the named cockpit config, checks whether autopilot is enabled, then calls run_resolvers() to evaluate live telemetry.

For each resolver that returns a non-nominal action, it either records a dry-run entry or schedules a real TimelineActivity on the Yamcs timeline 5 seconds in the future. Every decision is logged to the JSON audit trail.

  • Add the loop controller and main entry point at the bottom of the file:
def run_autopilot(config_name="nominal", cycles=5, interval=10, dry_run=False):
    print(f"Auto-Pilot: config={config_name}, cycles={cycles}, dry_run={dry_run}")
    print("=" * 40)

    for i in range(cycles):
        print(f"\n[Cycle {i+1}/{cycles}]")
        result = execute_cycle(config_name, dry_run=dry_run)
        print(f"  Action: {result['action']}")
        for a in result.get("actions", []):
            status = "DRY-RUN" if a.get("dry_run") else ("OK" if a.get("scheduled") else "FAIL")
            print(f"  [{status}] {a.get('resolver')}: {a.get('reason', a.get('error', ''))}")
        if i < cycles - 1:
            time.sleep(interval)

    print(f"\nCompleted {cycles} cycles.")


if __name__ == "__main__":
    run_autopilot(config_name="nominal", cycles=5, interval=10)

What does this code do?

The run_autopilot() function loops for a configurable number of cycles with a sleep interval between each. It prints a status line per cycle so you can watch the auto-pilot make decisions in real time.

When you run autopilot.py directly, it defaults to the nominal config with 5 cycles spaced 10 seconds apart.

  • Save autopilot.py.
  • Make sure your Yamcs Docker container is running, then test the auto-pilot by running:
python autopilot.py
  • You should see 5 cycles print to the terminal with actions like NO_ACTION or EXECUTED depending on your simulator's current battery voltage.

Seeing a connection error?

Make sure your Yamcs Docker container is running. Check with docker ps and start it with docker start yamcs if needed.

If you see an import error for scheduler or resolvers, make sure you are running the command from your project root directory where all .py files live.

help me debug my auto-pilot connection issue

✔️ Awesome, I've got everything!

Great. Double check you have saved autopilot.py and that logs/autopilot_log.json was created in your project folder.

ⓧ I'd like to double check the full code

import json
import os
import time
from datetime import datetime, timedelta, timezone

from yamcs.client import CommandActivity, TimelineActivity

from scheduler import get_timeline_client
from resolvers import run_resolvers

LOGS_DIR = "logs"


def ensure_dirs():
    os.makedirs(LOGS_DIR, exist_ok=True)


def load_config(name):
    with open(os.path.join("configs", f"{name}.json"), "r") as f:
        return json.load(f)


def log_decision(entry):
    ensure_dirs()
    log_file = os.path.join(LOGS_DIR, "autopilot_log.json")
    logs = []
    if os.path.exists(log_file):
        with open(log_file, "r") as f:
            try:
                logs = json.load(f)
            except json.JSONDecodeError:
                logs = []
    logs.append(entry)
    logs = logs[-100:]
    with open(log_file, "w") as f:
        json.dump(logs, f, indent=2)


def execute_cycle(config_name="nominal", dry_run=False):
    config = load_config(config_name)

    if not config["autopilot"]["enabled"]:
        entry = {
            "timestamp": datetime.now(tz=timezone.utc).isoformat(),
            "config": config_name,
            "action": "SKIPPED",
            "reason": "Autopilot disabled in configuration",
            "dry_run": dry_run,
        }
        log_decision(entry)
        return entry

    decisions = run_resolvers(config)
    actions = []

    for resolver_name, resolution in decisions["resolutions"].items():
        if resolution["action"] == "NOMINAL" or resolution["command"] is None:
            continue

        if dry_run:
            actions.append({
                "resolver": resolver_name,
                "action": resolution["action"],
                "command": resolution["command"],
                "reason": resolution["reason"],
                "scheduled": False,
                "dry_run": True,
            })
        else:
            try:
                timeline = get_timeline_client()
                now = datetime.now(tz=timezone.utc)
                item = TimelineActivity(
                    name=f"autopilot/{resolver_name}/{resolution['action']}",
                    start=now + timedelta(seconds=5),
                    duration=timedelta(seconds=30),
                    activity=CommandActivity(
                        command=resolution["command"],
                        args=resolution["args"],
                    ),
                    tags=["autopilot", config_name],
                )
                timeline.save_item(item)
                actions.append({
                    "resolver": resolver_name,
                    "action": resolution["action"],
                    "command": resolution["command"],
                    "reason": resolution["reason"],
                    "scheduled": True,
                })
            except Exception as e:
                actions.append({
                    "resolver": resolver_name,
                    "action": resolution["action"],
                    "error": str(e),
                    "scheduled": False,
                })

    entry = {
        "timestamp": datetime.now(tz=timezone.utc).isoformat(),
        "config": config_name,
        "state": decisions["state"],
        "actions": actions,
        "action": "EXECUTED" if actions else "NO_ACTION",
        "dry_run": dry_run,
    }
    log_decision(entry)
    return entry


def run_autopilot(config_name="nominal", cycles=5, interval=10, dry_run=False):
    print(f"Auto-Pilot: config={config_name}, cycles={cycles}, dry_run={dry_run}")
    print("=" * 40)

    for i in range(cycles):
        print(f"\n[Cycle {i+1}/{cycles}]")
        result = execute_cycle(config_name, dry_run=dry_run)
        print(f"  Action: {result['action']}")
        for a in result.get("actions", []):
            status = "DRY-RUN" if a.get("dry_run") else ("OK" if a.get("scheduled") else "FAIL")
            print(f"  [{status}] {a.get('resolver')}: {a.get('reason', a.get('error', ''))}")
        if i < cycles - 1:
            time.sleep(interval)

    print(f"\nCompleted {cycles} cycles.")


if __name__ == "__main__":
    run_autopilot(config_name="nominal", cycles=5, interval=10)
Deploy to Render

Your dashboard runs locally, but Yamcs lives inside Docker on your machine. Render cannot reach your local Yamcs instance, so the deployed dashboard will serve the cached config files and log data without a live connection. The demo video you record next proves the full local integration.

  • Create a file called render.yaml in your project root. This tells Render how to build and run your service:
services:
  - type: web
    name: mission-ops-cockpit
    runtime: python
    buildCommand: pip install -r requirements.txt
    startCommand: gunicorn app:app --bind 0.0.0.0:$PORT
    plan: free
    envVars:
      - key: DEMO_MODE
        value: "true"

What does this file do?

The render.yaml is a Render Blueprint that pre-configures your service. The buildCommand installs your Python dependencies. The startCommand uses gunicorn (a production WSGI server for Linux) to serve your Flask app.

The DEMO_MODE environment variable signals to the app that there is no live Yamcs connection. Your app already reads from local JSON files, so it works without Yamcs being reachable.

  • Save render.yaml.
  • Initialize a Git repository and push to GitHub by running these commands:
git init
git add .
git commit -m "Mission operations scheduler"
  • Go to github.com and create a new repository. Name it something like mission-ops-cockpit.
  • Do not initialize it with a README or .gitignore (your local repo already has commits).
  • Push your local repository to GitHub by running:
git remote add origin [[YOUR_REPO_URL="https://github.com/YOUR_USERNAME/mission-ops-cockpit.git"]]
git branch -M main
git push -u origin main
  • Go to render.com and sign up for a free account if you do not have one.
  • In the Render Dashboard, click + New in the top-right corner.
  • Select Web Service.
  • Connect your GitHub account if prompted, then select your mission-ops-cockpit repository and click Connect.
  • In the Language field, select Python 3.
  • Set the Build Command to:
pip install -r requirements.txt
  • Set the Start Command to:
gunicorn app:app --bind 0.0.0.0:$PORT
  • For the instance type, choose Free.
  • Expand the Advanced section and add an environment variable with key DEMO_MODE and value true.
  • Click Deploy.

Why gunicorn instead of waitress?

Render runs Linux servers, and gunicorn is the standard production WSGI server for UNIX systems. Your app.py uses waitress locally on Windows because gunicorn does not run on Windows. On Render, the Linux-native gunicorn handles production traffic.

  • Wait for the build to complete (1-2 minutes). Render shows deploy progress on the service's Events page.
  • Once deployed, Render gives your service a unique .onrender.com URL. Open it in your browser.
  • You should see the Mission Operations Cockpit dashboard with your three cockpit configurations listed and the API endpoints section.

Dashboard not loading on Render?

Check the deploy logs in the Render Dashboard for error messages. A common issue is a missing dependency in requirements.txt. Make sure gunicorn is listed there.

If you see a timeout, your app might not be binding to the correct port. Confirm your Start Command includes --bind 0.0.0.0:$PORT.

help me debug my Render deployment

Record a demo video

The demo video is proof that your system works end-to-end against live Yamcs. This is the deliverable you would share with a hiring manager to demonstrate your satellite operations engineering skills.

  • Make sure your Yamcs Docker container is running and the web UI is accessible at http://localhost:8090.
  • Open a screen recording tool. On Windows, press Win+G to open the Game Bar recorder. On macOS, press Cmd+Shift+5 to open Screen Recording. Alternatively, download OBS for either platform.
  • Start your screen recording and run the auto-pilot with 3 cycles by running:
python autopilot.py

Use 3 cycles for a shorter demo

The default is 5 cycles. You can edit the if __name__ block temporarily to cycles=3 for a tighter video, or just stop the recording after 3 cycles complete.

  • While the auto-pilot runs, switch to your browser and open the Yamcs Timeline UI at http://localhost:8090. Navigate to the Timeline view to show activities appearing.
  • Switch back to the terminal to show resolver decisions printing.
  • Run the auto-pilot again with the safe_mode config to demonstrate a config switch:
python -c "from autopilot import run_autopilot; run_autopilot(config_name='safe_mode', cycles=3, interval=10)"
  • Stop the screen recording. Your video should be 2-3 minutes.
  • Upload the video to YouTube as an unlisted video, or attach it to a GitHub Release on your repository.
  • Add the video link and your live Render dashboard URL to your repository's README.md.

What should the README include?

Add a short project description, your live Render dashboard URL (the .onrender.com link), and the demo video link. This makes your repository a complete portfolio piece that demonstrates timeline scheduling, resolver logic, and cockpit configuration management.

Secret mission

Test as We Fly Validation Mode

Real satellite programs validate every mission-critical system before launch using the "Test as We Fly" methodology. Can you build a fault injection harness that dry-runs your auto-pilot against simulated failures and generates a structured pass/fail verification report?

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 uses Docker locally (no ongoing cost when stopped) and a Render free-tier service that auto-sleeps after 15 minutes of inactivity.

Resources you used:

  • Docker container running Yamcs (no ongoing cost, can be stopped).
  • Render free-tier web service (spins down after 15 min inactivity, 750 hours/month free).
  • GitHub repository (free).
  • Local project folder with Python files, configs, and logs.

Keep everything running

No action needed. Choose this if you are still actively building or want to revisit the project soon.

  • Your Docker container stays running and Yamcs remains accessible at localhost:8090.
  • Your Render service stays deployed on the free tier. It auto-sleeps after 15 minutes of inactivity and wakes up when someone visits the URL.
  • Your GitHub repository remains available at no cost.
  • Your local project files remain on your Desktop.

Pause - I'll come back to this later

Shut down running processes to free up system resources while keeping all files intact for later.

  • Stop the Yamcs Docker container by running:
docker stop yamcs
  • Your Render service automatically sleeps after 15 minutes of inactivity. No action needed.
  • Your local project files and GitHub repository remain untouched.
  • To resume later, start the container again with docker start yamcs.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop and remove the Yamcs Docker container by running:
docker stop yamcs
docker rm yamcs
  • Delete the Render web service:
  • Open the Render Dashboard and click on your mission-ops-cockpit service.
  • Click Settings in the left pane.
  • Scroll to the bottom of the Settings page and click Delete Web Service.
  • Confirm the deletion when prompted.
  • Delete the GitHub repository:
  • Open your repository on GitHub.
  • Click Settings (the tab at the far right of the repository navigation).
  • Scroll to the bottom and click Delete this repository in the Danger Zone section.
  • Follow the confirmation prompts to complete deletion.
  • Delete your local project folder by running:

macOS

cd ~/Desktop
rm -rf mission-ops

Windows

cd ~/Desktop
rmdir /s /q mission-ops

Nice Work!

Nice Work!

Nice work! You just built a full mission operations scheduler and cockpit configuration system that integrates with Yamcs to sequence satellite commands, resolve telemetry conditions, and manage operational phases autonomously.

You've learned how to:

  • Built a timeline-based pass scheduler using the Yamcs Python client Timeline API with chained OnSuccess activities simulating orbital pass subschedules.
  • Implemented a resolver engine that evaluates real-time telemetry to conditionally select which command procedures to execute based on spacecraft state thresholds.
  • Created a cockpit configuration system with switchable JSON profiles for commissioning, nominal, and safe-mode mission phases, and deployed a Flask dashboard on Render with a recorded demo video.
  • Secret Mission: Added Test as We Fly validation with fault injection scenarios and automated pass/fail reporting.

Ready to quiz yourself?