Satellite Ground Ops Automation Engine

Introduction

30 Second Summary

Every satellite in orbit generates thousands of telemetry readings per second. Ground operators cannot watch every parameter around the clock, so modern missions rely on automation engines to detect anomalies and fire corrective procedures without human intervention.

In this project, you will build a GroundOps Console automation engine: the backend brain of a satellite mission operations system that monitors telemetry, detects anomalies, and automatically executes corrective procedures.

What You'll Build

A FastAPI application with an async satellite simulator, JSON-configurable alarm monitor, declarative procedure engine, automated resolver system, and an unattended autopilot sequencer, all viewable through a live WebSocket telemetry dashboard.

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

• A satellite simulator producing realistic battery, thermal, and comms telemetry with random-walk drift.
• A JSON-configurable telemetry monitor that evaluates alarm limits and manages an alert lifecycle.
• A declarative procedure engine and automated resolver system that maps anomalies to corrective actions and executes them unattended via an autopilot sequencer.
• A test-as-you-fly pytest suite that proves the full automation chain works end-to-end with no mocks of core logic.
• Secret Mission: Add fault-injection testing to prove the autopilot handles procedure timeouts gracefully.

Are there any prerequisites?

This project assumes working knowledge of Python, FastAPI, and async/await. You should be comfortable writing classes, using decorators, and reading JSON configuration files.

Set Up the Project

Every module in this system (simulator, monitor, resolver, autopilot) shares a common set of data structures. Defining these models first gives you a typed contract that every component must respect.

In this step, you'll scaffold the project directory, install dependencies, and create the Pydantic data models that flow through the entire automation chain.

In this step, get ready to:

• Create the project directory and virtual environment.
• Install dependencies and create configuration files.
• Define the data models for telemetry, alerts, and procedures.

Create the project directory and virtual environment

You need an isolated Python environment to keep this project's dependencies separate from your system packages.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir groundops-console
cd groundops-console

• Create a Python virtual environment by running:

python -m venv venv

• Activate the virtual environment using the command for your operating system:

macOS

source venv/bin/activate

This command also works on Linux.

Windows

venv\Scripts\Activate.ps1

You should see (venv) appear at the start of your terminal prompt.

Don't see (venv) in your prompt?

On Windows PowerShell, you may need to allow script execution first. Run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser and then retry the activation command.

On macOS/Linux, make sure you typed source before the path. Without it, the script runs in a subshell and the activation doesn't stick.

Help me activate my Python virtual environment.

• Open VS Code in your project folder by running:

code .

Command not found?

If code is not recognized, open VS Code manually. Press Cmd+Space (macOS) or the Windows key (Windows), search for Visual Studio Code, and open it.

In VS Code, click File in the top menu bar. Click Open Folder and select the groundops-console folder on your Desktop.

Help me open my project in VS Code.

• In VS Code, open the integrated terminal by pressing Ctrl+` (backtick key, top-left of keyboard).
• Make sure (venv) still appears in the prompt. If not, re-run the activation command from above.

Now create the requirements.txt file that pins your project dependencies.

• In the VS Code Explorer sidebar, click the new file icon at the top of the file tree.
• Name the file requirements.txt and paste the following content:

fastapi==0.138.0
uvicorn[standard]==0.49.0
pydantic==2.13.4
pytest==9.1.1
httpx==0.28.1

What are these dependencies?

• fastapi is the async web framework that handles REST endpoints and WebSocket connections.
• uvicorn is the ASGI server that runs the FastAPI app. The [standard] extra adds WebSocket support and hot-reload.
• pydantic provides typed data models with validation.
• pytest is the test runner for the test-as-you-fly suite.
• httpx is an async HTTP client used for testing FastAPI endpoints.

• Save requirements.txt.
• Install all dependencies by running this command in the integrated terminal:

pip install -r requirements.txt

You should see pip downloading and installing each package. The final line shows a success message followed by the package names.

Install failing or hanging?

Make sure (venv) is visible in your prompt. Installing without the virtual environment active may require elevated permissions and pollutes your system Python.

If you see a network timeout, check your internet connection and try again.

Help me fix my pip install.

• Create the project's directory structure by running:

mkdir config static tests

• Confirm the directories exist by running:

ls

You should see config, static, tests, venv, and requirements.txt in the output.

Finally, create the pytest configuration file so tests can import your project modules.

• In the VS Code Explorer sidebar, click the new file icon.
• Name the file pyproject.toml and paste the following content:

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]

What does this configuration do?

pythonpath = ["."] tells pytest to add the project root to Python's import path. Without it, test files inside tests/ cannot import modules like models.py from the parent directory.

testpaths = ["tests"] tells pytest where to look for test files when you run it from the project root.

• Save pyproject.toml.

Create the data models module

The automation engine passes structured data between components: telemetry frames from the simulator, alerts from the monitor, procedure outcomes from the engine.

Defining all models in one module gives every component a shared vocabulary.

• In the VS Code Explorer sidebar, click the new file icon.
• Name the file models.py.
• Add the imports and telemetry models by pasting the following code:

import datetime
from enum import Enum
from typing import Optional

from pydantic import BaseModel


class ADCSMode(str, Enum):
    NOMINAL = "NOMINAL"
    SAFE = "SAFE"
    DETUMBLE = "DETUMBLE"


class TelemetryFrame(BaseModel):
    timestamp: datetime.datetime
    satellite_id: str
    battery_voltage: float
    obc_temperature: float
    adcs_mode: ADCSMode
    comms_signal_strength: float


class AlertSeverity(str, Enum):
    NOMINAL = "NOMINAL"
    WARNING = "WARNING"
    CRITICAL = "CRITICAL"


class Alert(BaseModel):
    timestamp: datetime.datetime
    satellite_id: str
    parameter: str
    value: float
    severity: AlertSeverity
    message: str
    resolved: bool = False

What does this code do?

• ADCSMode is a string enum representing the satellite's attitude determination and control system state (nominal, safe mode, or detumble).
• TelemetryFrame represents one snapshot of satellite housekeeping data. Every second, the simulator produces one of these.
• AlertSeverity and Alert model the alarm system. When a telemetry parameter drifts out of limits, the monitor creates an Alert with the relevant severity.

• Add the procedure and automation models below the Alert class. Paste this code at the bottom of models.py:

class ProcedureStepResult(str, Enum):
    SUCCESS = "SUCCESS"
    FAILURE = "FAILURE"
    TIMEOUT = "TIMEOUT"


class ProcedureStep(BaseModel):
    name: str
    action: str
    precondition: Optional[str] = None
    timeout_seconds: float = 30.0


class Procedure(BaseModel):
    name: str
    description: str
    steps: list[ProcedureStep]


class StepOutcome(BaseModel):
    step_name: str
    result: ProcedureStepResult
    message: str
    timestamp: datetime.datetime


class ProcedureOutcome(BaseModel):
    procedure_name: str
    started_at: datetime.datetime
    completed_at: datetime.datetime
    step_outcomes: list[StepOutcome]
    success: bool


class ResolverRule(BaseModel):
    name: str
    parameter: str
    severity: AlertSeverity
    procedure_name: str


class AutopilotResult(BaseModel):
    sequence_name: str
    started_at: datetime.datetime
    completed_at: datetime.datetime
    procedure_outcomes: list[ProcedureOutcome]
    all_succeeded: bool

What does this code do?

• ProcedureStepResult tracks whether each step in a procedure succeeded, failed, or timed out. Timeout is a distinct state because satellite commands can hang without explicitly failing.
• ProcedureStep and Procedure define the shape of a corrective procedure loaded from JSON. Each procedure has named steps with actions and timeouts.
• StepOutcome and ProcedureOutcome record what actually happened when a procedure ran. This creates an audit trail.
• ResolverRule maps an alert condition (parameter + severity) to a corrective procedure name. The resolver uses these rules to decide which procedure to fire.
• AutopilotResult captures the outcome of running a full sequence of procedures unattended.

• Save models.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py and move on to the verification step below.

ⓧ I'd like to double check the full code

Here is the complete models.py file for comparison:

import datetime
from enum import Enum
from typing import Optional

from pydantic import BaseModel


class ADCSMode(str, Enum):
    NOMINAL = "NOMINAL"
    SAFE = "SAFE"
    DETUMBLE = "DETUMBLE"


class TelemetryFrame(BaseModel):
    timestamp: datetime.datetime
    satellite_id: str
    battery_voltage: float
    obc_temperature: float
    adcs_mode: ADCSMode
    comms_signal_strength: float


class AlertSeverity(str, Enum):
    NOMINAL = "NOMINAL"
    WARNING = "WARNING"
    CRITICAL = "CRITICAL"


class Alert(BaseModel):
    timestamp: datetime.datetime
    satellite_id: str
    parameter: str
    value: float
    severity: AlertSeverity
    message: str
    resolved: bool = False


class ProcedureStepResult(str, Enum):
    SUCCESS = "SUCCESS"
    FAILURE = "FAILURE"
    TIMEOUT = "TIMEOUT"


class ProcedureStep(BaseModel):
    name: str
    action: str
    precondition: Optional[str] = None
    timeout_seconds: float = 30.0


class Procedure(BaseModel):
    name: str
    description: str
    steps: list[ProcedureStep]


class StepOutcome(BaseModel):
    step_name: str
    result: ProcedureStepResult
    message: str
    timestamp: datetime.datetime


class ProcedureOutcome(BaseModel):
    procedure_name: str
    started_at: datetime.datetime
    completed_at: datetime.datetime
    step_outcomes: list[StepOutcome]
    success: bool


class ResolverRule(BaseModel):
    name: str
    parameter: str
    severity: AlertSeverity
    procedure_name: str


class AutopilotResult(BaseModel):
    sequence_name: str
    started_at: datetime.datetime
    completed_at: datetime.datetime
    procedure_outcomes: list[ProcedureOutcome]
    all_succeeded: bool

Verify the setup

• Confirm that Python can import your models by running this command in the integrated terminal:

python -c "from models import TelemetryFrame; print('Models OK')"

You should see Models OK printed with no errors.

Seeing a ModuleNotFoundError?

Make sure your terminal's working directory is the groundops-console folder (where models.py lives). Run ls and confirm you see models.py in the listing.

If you see an import error for pydantic, your virtual environment may not be active. Check for (venv) in your prompt and re-activate if needed.

Help me fix my models import error.

• Also verify your installed packages by running:

pip list

You should see fastapi, uvicorn, pydantic, pytest, and httpx at the pinned versions in the output.

Your project skeleton is ready. Next up, you'll build the satellite simulator and wire it to a live WebSocket stream.

Build the Satellite Simulator and WebSocket Stream

In the previous step you set up the project skeleton and defined all the data models your system will use. Now it's time to bring the satellite to life.

A real ground operations console receives telemetry from a spacecraft every few seconds. You'll build a simulator that produces that telemetry stream, a FastAPI backend that broadcasts it over a WebSocket, and a minimal browser dashboard that renders it in real time.

In this step, get ready to:

• Build a satellite simulator that produces drifting telemetry and accepts commands.
• Wire up FastAPI with a WebSocket endpoint that streams telemetry every second.
• Create a dark-themed HTML telemetry viewer with color-coded anomaly highlighting.

Build the satellite simulator

The simulator is the heart of your test environment. It produces telemetry frames with realistic random-walk drift on key parameters (battery voltage, temperature, comms signal strength) and exposes a command interface that resets parameters to nominal.

• Create a new file called simulator.py in the root of your groundops-console folder (alongside models.py).
• Add the full satellite simulator class by pasting this code:

import datetime
import random

from models import ADCSMode, TelemetryFrame


class SatelliteSimulator:
    def __init__(self, satellite_id: str = "SAT-01"):
        self.satellite_id = satellite_id
        self.battery_voltage = 28.0
        self.obc_temperature = 25.0
        self.adcs_mode = ADCSMode.NOMINAL
        self.comms_signal_strength = -80.0
        self._running = False

    def generate_frame(self) -> TelemetryFrame:
        # Apply random-walk drift to each parameter
        self.battery_voltage += random.uniform(-0.3, 0.2)
        self.battery_voltage = max(20.0, min(32.0, self.battery_voltage))

        self.obc_temperature += random.uniform(-0.5, 0.6)
        self.obc_temperature = max(-10.0, min(60.0, self.obc_temperature))

        self.comms_signal_strength += random.uniform(-2.0, 2.0)
        self.comms_signal_strength = max(-120.0, min(-50.0, self.comms_signal_strength))

        return TelemetryFrame(
            timestamp=datetime.datetime.now(datetime.timezone.utc),
            satellite_id=self.satellite_id,
            battery_voltage=round(self.battery_voltage, 2),
            obc_temperature=round(self.obc_temperature, 2),
            adcs_mode=self.adcs_mode,
            comms_signal_strength=round(self.comms_signal_strength, 2),
        )

    def stop(self):
        self._running = False

    def command(self, action: str) -> bool:
        # Execute a command and reset the relevant parameter to nominal
        if action == "reset_battery_controller":
            self.battery_voltage = 28.0
            return True
        elif action == "activate_heater":
            self.obc_temperature = 25.0
            return True
        elif action == "switch_adcs_nominal":
            self.adcs_mode = ADCSMode.NOMINAL
            return True
        elif action == "reset_comms":
            self.comms_signal_strength = -80.0
            return True
        elif action == "power_cycle_obc":
            self.obc_temperature = 22.0
            self.battery_voltage = 27.5
            return True
        return False

What does this code do?

The SatelliteSimulator class models a spacecraft with four housekeeping parameters.

• generate_frame() applies a small random-walk drift to each parameter on every call, clamped within realistic bounds. This simulates gradual degradation the way a real satellite drifts over time.
• command() simulates sending a command to the spacecraft. Each recognized action resets the relevant parameter to its nominal value, mimicking how ground operators recover from anomalies.
• The drift is intentionally asymmetric (slightly biased downward for battery, upward for temperature) so anomalies emerge naturally over time.

• Save simulator.py.
• Verify the simulator imports cleanly by running this command in your terminal:

python -c "from simulator import SatelliteSimulator; s = SatelliteSimulator(); print(s.generate_frame())"

You should see a TelemetryFrame printed with a UTC timestamp, satellite ID SAT-01, and values close to the nominals (battery ~28V, temperature ~25C, comms ~-80dBm).

Seeing a ModuleNotFoundError?

• Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt.
• Confirm models.py is in the same directory as simulator.py. Both files must be at the root of groundops-console/.

Help me debug a ModuleNotFoundError when importing SatelliteSimulator

Wire up FastAPI with WebSocket telemetry streaming

The FastAPI application is the server that ties everything together. It runs a background loop that generates telemetry every second and pushes it to any connected WebSocket clients.

• Create a new file called main.py in the root of your groundops-console folder.
• Add the imports, module instances, lifespan, and telemetry loop by pasting this code:

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from autopilot import AutopilotSequencer
from monitor import TelemetryMonitor
from procedures import ProcedureEngine
from resolvers import ResolverSystem
from simulator import SatelliteSimulator

# Module-level instances shared across the application
simulator = SatelliteSimulator()
monitor = TelemetryMonitor()
engine = ProcedureEngine()
resolvers = ResolverSystem()
autopilot = AutopilotSequencer(engine)
connected_clients: list[WebSocket] = []


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start background telemetry loop on startup
    task = asyncio.create_task(telemetry_loop())
    yield
    # Clean up on shutdown
    simulator.stop()
    task.cancel()


app = FastAPI(title="GroundOps Console", lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")


async def telemetry_loop():
    while True:
        frame = simulator.generate_frame()
        alerts = monitor.evaluate(frame)

        # Auto-resolve any alerts by firing the matched procedure
        for alert in alerts:
            procedure_name = resolvers.match(alert)
            if procedure_name:
                outcome = await engine.execute(procedure_name, simulator.command)
                resolvers.log_action(alert, outcome)

        # Broadcast telemetry to all connected WebSocket clients
        frame_data = frame.model_dump_json()
        for client in connected_clients.copy():
            try:
                await client.send_text(frame_data)
            except Exception:
                connected_clients.remove(client)

        await asyncio.sleep(1.0)

What does this code do?

• The lifespan context manager starts the background telemetry loop when the server boots and cancels it on shutdown. This is FastAPI's recommended pattern for managing long-running background tasks.
• telemetry_loop() generates a telemetry frame every second, evaluates it for anomalies, auto-fires corrective procedures when needed, and broadcasts the raw frame to every connected WebSocket client.
• The connected_clients.copy() pattern avoids mutating the list while iterating. If a send fails (client disconnected), the client is silently removed.

• Now add all the route handlers below the telemetry_loop function in the same main.py file:

@app.get("/")
async def root():
    return FileResponse("static/index.html")


@app.websocket("/ws/telemetry")
async def telemetry_ws(websocket: WebSocket):
    await websocket.accept()
    connected_clients.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        connected_clients.remove(websocket)


@app.get("/api/alerts")
async def get_alerts():
    return {"alerts": [a.model_dump() for a in monitor.alerts[-50:]]}


@app.get("/api/procedures")
async def get_procedures():
    return {"procedures": list(engine.procedures.keys())}


@app.post("/api/procedures/{name}/execute")
async def execute_procedure(name: str):
    if name not in engine.procedures:
        return {"error": f"Procedure '{name}' not found"}
    outcome = await engine.execute(name, simulator.command)
    return outcome.model_dump()


class AutopilotRequest(BaseModel):
    sequence_name: str
    procedure_names: list[str]


@app.post("/api/autopilot/execute")
async def execute_autopilot(request: AutopilotRequest):
    result = await autopilot.execute_sequence(
        request.sequence_name, request.procedure_names, simulator.command
    )
    return result.model_dump()


@app.get("/api/autopilot/history")
async def get_autopilot_history():
    return {"history": [r.model_dump() for r in autopilot.history]}


@app.get("/api/resolver/history")
async def get_resolver_history():
    return {"history": resolvers.history}

What do these endpoints do?

• GET / serves the HTML dashboard from static/index.html.
• /ws/telemetry is the WebSocket endpoint. Clients connect here to receive live telemetry frames as JSON.
• The REST endpoints (/api/alerts, /api/procedures, /api/autopilot) provide query and control access for the modules you'll build in the next steps.

• Save main.py.

Why does main.py import modules that don't exist yet?

The imports for monitor, procedures, resolvers, and autopilot reference modules you'll create in the next steps. The server won't start until all these modules exist.

You're writing the full application scaffold now so that each subsequent step slots in cleanly without restructuring main.py. You'll start the server at the end of the next step once the missing modules are in place.

Syntax errors in main.py?

• Check that the @asynccontextmanager decorator is directly above the async def lifespan line with no blank lines between them.
• Make sure the f-string in execute_procedure uses matching quotes. The outer string uses double quotes and the inner uses single quotes.

Help me fix a syntax error in my FastAPI main.py

Create the minimal HTML telemetry viewer

The dashboard connects to your WebSocket endpoint and renders each telemetry frame with color-coding. Green means nominal, yellow means warning, red means critical.

• Create a new file called index.html inside your static/ directory.
• Paste the full ops console HTML into the file:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>GroundOps Console</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: 'Courier New', monospace; background: #0a0e14; color: #c5c8c6; padding: 20px; }
        h1 { color: #66d9ef; margin-bottom: 20px; font-size: 1.4rem; }
        .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 16px; margin-bottom: 20px; }
        .card { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 16px; }
        .card h2 { color: #a6e22e; font-size: 0.9rem; margin-bottom: 12px; text-transform: uppercase; }
        .metric { display: flex; justify-content: space-between; padding: 6px 0; border-bottom: 1px solid #2a3040; }
        .metric:last-child { border-bottom: none; }
        .label { color: #888; }
        .value { font-weight: bold; }
        .nominal { color: #a6e22e; }
        .warning { color: #e6db74; }
        .critical { color: #f92672; }
        .status-bar { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 12px 16px; margin-bottom: 20px; display: flex; align-items: center; gap: 12px; }
        .status-dot { width: 10px; height: 10px; border-radius: 50%; background: #888; }
        .status-dot.connected { background: #a6e22e; }
        .log { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 16px; max-height: 300px; overflow-y: auto; }
        .log h2 { color: #a6e22e; font-size: 0.9rem; margin-bottom: 12px; text-transform: uppercase; }
        .log-entry { font-size: 0.8rem; padding: 4px 0; border-bottom: 1px solid #1a1f29; }
    </style>
</head>
<body>
    <h1>GroundOps Console - Telemetry Monitor</h1>
    <div class="status-bar">
        <div class="status-dot" id="statusDot"></div>
        <span id="statusText">Connecting...</span>
    </div>
    <div class="grid">
        <div class="card">
            <h2>Battery</h2>
            <div class="metric">
                <span class="label">Voltage</span>
                <span class="value" id="batteryVoltage">--</span>
            </div>
        </div>
        <div class="card">
            <h2>Thermal</h2>
            <div class="metric">
                <span class="label">OBC Temp</span>
                <span class="value" id="obcTemperature">--</span>
            </div>
        </div>
        <div class="card">
            <h2>ADCS</h2>
            <div class="metric">
                <span class="label">Mode</span>
                <span class="value" id="adcsMode">--</span>
            </div>
        </div>
        <div class="card">
            <h2>Communications</h2>
            <div class="metric">
                <span class="label">Signal</span>
                <span class="value" id="commsSignal">--</span>
            </div>
        </div>
    </div>
    <div class="log">
        <h2>Event Log</h2>
        <div id="logEntries"></div>
    </div>
    <script>
        const ws = new WebSocket(`ws://${window.location.host}/ws/telemetry`);
        const statusDot = document.getElementById('statusDot');
        const statusText = document.getElementById('statusText');
        const logEntries = document.getElementById('logEntries');

        const limits = {
            battery_voltage: { warning_low: 24.0, warning_high: 30.0, critical_low: 22.0, critical_high: 31.0 },
            obc_temperature: { warning_low: 5.0, warning_high: 45.0, critical_low: 0.0, critical_high: 50.0 },
            comms_signal_strength: { warning_low: -100.0, critical_low: -110.0 }
        };

        function getSeverityClass(value, paramLimits) {
            if (paramLimits.critical_low !== undefined && value <= paramLimits.critical_low) return 'critical';
            if (paramLimits.critical_high !== undefined && value >= paramLimits.critical_high) return 'critical';
            if (paramLimits.warning_low !== undefined && value <= paramLimits.warning_low) return 'warning';
            if (paramLimits.warning_high !== undefined && value >= paramLimits.warning_high) return 'warning';
            return 'nominal';
        }

        ws.onopen = () => {
            statusDot.classList.add('connected');
            statusText.textContent = 'Connected - Receiving telemetry';
            addLog('WebSocket connected');
        };

        ws.onclose = () => {
            statusDot.classList.remove('connected');
            statusText.textContent = 'Disconnected';
            addLog('WebSocket disconnected');
        };

        ws.onmessage = (event) => {
            const frame = JSON.parse(event.data);
            updateTelemetry(frame);
        };

        function updateTelemetry(frame) {
            const battEl = document.getElementById('batteryVoltage');
            const tempEl = document.getElementById('obcTemperature');
            const adcsEl = document.getElementById('adcsMode');
            const commsEl = document.getElementById('commsSignal');

            battEl.textContent = frame.battery_voltage.toFixed(2) + ' V';
            battEl.className = 'value ' + getSeverityClass(frame.battery_voltage, limits.battery_voltage);

            tempEl.textContent = frame.obc_temperature.toFixed(2) + ' C';
            tempEl.className = 'value ' + getSeverityClass(frame.obc_temperature, limits.obc_temperature);

            adcsEl.textContent = frame.adcs_mode;
            adcsEl.className = 'value nominal';

            commsEl.textContent = frame.comms_signal_strength.toFixed(2) + ' dBm';
            commsEl.className = 'value ' + getSeverityClass(frame.comms_signal_strength, limits.comms_signal_strength);
        }

        function addLog(message) {
            const entry = document.createElement('div');
            entry.className = 'log-entry';
            entry.textContent = `[${new Date().toISOString().substr(11, 8)}] ${message}`;
            logEntries.prepend(entry);
            if (logEntries.children.length > 50) {
                logEntries.removeChild(logEntries.lastChild);
            }
        }
    </script>
</body>
</html>

What does the viewer do?

• The page opens a WebSocket to /ws/telemetry on the same host. Every incoming JSON frame updates the four metric cards.
• getSeverityClass() compares each value against hardcoded alarm limits and returns a CSS class: nominal (green), warning (yellow), or critical (red).
• A status dot in the top bar turns green when the WebSocket is connected, giving the learner immediate visual feedback.

• Save static/index.html.
• Confirm all three new files exist by running:

ls simulator.py main.py static/index.html

You should see all three file paths listed in the output.

File not found?

• Make sure index.html is inside the static/ directory, not at the project root.
• Verify your terminal is in the groundops-console/ directory. Run pwd to check.

Help me find where my files ended up

✔️ Awesome, I've got everything!

Great. Double check you've saved all three files before moving on.

ⓧ I'd like to double check the full code

Here's the complete code for each file at this point.

import datetime
import random

from models import ADCSMode, TelemetryFrame


class SatelliteSimulator:
    def __init__(self, satellite_id: str = "SAT-01"):
        self.satellite_id = satellite_id
        self.battery_voltage = 28.0
        self.obc_temperature = 25.0
        self.adcs_mode = ADCSMode.NOMINAL
        self.comms_signal_strength = -80.0
        self._running = False

    def generate_frame(self) -> TelemetryFrame:
        self.battery_voltage += random.uniform(-0.3, 0.2)
        self.battery_voltage = max(20.0, min(32.0, self.battery_voltage))

        self.obc_temperature += random.uniform(-0.5, 0.6)
        self.obc_temperature = max(-10.0, min(60.0, self.obc_temperature))

        self.comms_signal_strength += random.uniform(-2.0, 2.0)
        self.comms_signal_strength = max(-120.0, min(-50.0, self.comms_signal_strength))

        return TelemetryFrame(
            timestamp=datetime.datetime.now(datetime.timezone.utc),
            satellite_id=self.satellite_id,
            battery_voltage=round(self.battery_voltage, 2),
            obc_temperature=round(self.obc_temperature, 2),
            adcs_mode=self.adcs_mode,
            comms_signal_strength=round(self.comms_signal_strength, 2),
        )

    def stop(self):
        self._running = False

    def command(self, action: str) -> bool:
        if action == "reset_battery_controller":
            self.battery_voltage = 28.0
            return True
        elif action == "activate_heater":
            self.obc_temperature = 25.0
            return True
        elif action == "switch_adcs_nominal":
            self.adcs_mode = ADCSMode.NOMINAL
            return True
        elif action == "reset_comms":
            self.comms_signal_strength = -80.0
            return True
        elif action == "power_cycle_obc":
            self.obc_temperature = 22.0
            self.battery_voltage = 27.5
            return True
        return False

import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from autopilot import AutopilotSequencer
from monitor import TelemetryMonitor
from procedures import ProcedureEngine
from resolvers import ResolverSystem
from simulator import SatelliteSimulator

simulator = SatelliteSimulator()
monitor = TelemetryMonitor()
engine = ProcedureEngine()
resolvers = ResolverSystem()
autopilot = AutopilotSequencer(engine)
connected_clients: list[WebSocket] = []


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(telemetry_loop())
    yield
    simulator.stop()
    task.cancel()


app = FastAPI(title="GroundOps Console", lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")


async def telemetry_loop():
    while True:
        frame = simulator.generate_frame()
        alerts = monitor.evaluate(frame)

        for alert in alerts:
            procedure_name = resolvers.match(alert)
            if procedure_name:
                outcome = await engine.execute(procedure_name, simulator.command)
                resolvers.log_action(alert, outcome)

        frame_data = frame.model_dump_json()
        for client in connected_clients.copy():
            try:
                await client.send_text(frame_data)
            except Exception:
                connected_clients.remove(client)

        await asyncio.sleep(1.0)


@app.get("/")
async def root():
    return FileResponse("static/index.html")


@app.websocket("/ws/telemetry")
async def telemetry_ws(websocket: WebSocket):
    await websocket.accept()
    connected_clients.append(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        connected_clients.remove(websocket)


@app.get("/api/alerts")
async def get_alerts():
    return {"alerts": [a.model_dump() for a in monitor.alerts[-50:]]}


@app.get("/api/procedures")
async def get_procedures():
    return {"procedures": list(engine.procedures.keys())}


@app.post("/api/procedures/{name}/execute")
async def execute_procedure(name: str):
    if name not in engine.procedures:
        return {"error": f"Procedure '{name}' not found"}
    outcome = await engine.execute(name, simulator.command)
    return outcome.model_dump()


class AutopilotRequest(BaseModel):
    sequence_name: str
    procedure_names: list[str]


@app.post("/api/autopilot/execute")
async def execute_autopilot(request: AutopilotRequest):
    result = await autopilot.execute_sequence(
        request.sequence_name, request.procedure_names, simulator.command
    )
    return result.model_dump()


@app.get("/api/autopilot/history")
async def get_autopilot_history():
    return {"history": [r.model_dump() for r in autopilot.history]}


@app.get("/api/resolver/history")
async def get_resolver_history():
    return {"history": resolvers.history}

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>GroundOps Console</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: 'Courier New', monospace; background: #0a0e14; color: #c5c8c6; padding: 20px; }
        h1 { color: #66d9ef; margin-bottom: 20px; font-size: 1.4rem; }
        .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 16px; margin-bottom: 20px; }
        .card { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 16px; }
        .card h2 { color: #a6e22e; font-size: 0.9rem; margin-bottom: 12px; text-transform: uppercase; }
        .metric { display: flex; justify-content: space-between; padding: 6px 0; border-bottom: 1px solid #2a3040; }
        .metric:last-child { border-bottom: none; }
        .label { color: #888; }
        .value { font-weight: bold; }
        .nominal { color: #a6e22e; }
        .warning { color: #e6db74; }
        .critical { color: #f92672; }
        .status-bar { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 12px 16px; margin-bottom: 20px; display: flex; align-items: center; gap: 12px; }
        .status-dot { width: 10px; height: 10px; border-radius: 50%; background: #888; }
        .status-dot.connected { background: #a6e22e; }
        .log { background: #1a1f29; border: 1px solid #2a3040; border-radius: 8px; padding: 16px; max-height: 300px; overflow-y: auto; }
        .log h2 { color: #a6e22e; font-size: 0.9rem; margin-bottom: 12px; text-transform: uppercase; }
        .log-entry { font-size: 0.8rem; padding: 4px 0; border-bottom: 1px solid #1a1f29; }
    </style>
</head>
<body>
    <h1>GroundOps Console - Telemetry Monitor</h1>
    <div class="status-bar">
        <div class="status-dot" id="statusDot"></div>
        <span id="statusText">Connecting...</span>
    </div>
    <div class="grid">
        <div class="card">
            <h2>Battery</h2>
            <div class="metric">
                <span class="label">Voltage</span>
                <span class="value" id="batteryVoltage">--</span>
            </div>
        </div>
        <div class="card">
            <h2>Thermal</h2>
            <div class="metric">
                <span class="label">OBC Temp</span>
                <span class="value" id="obcTemperature">--</span>
            </div>
        </div>
        <div class="card">
            <h2>ADCS</h2>
            <div class="metric">
                <span class="label">Mode</span>
                <span class="value" id="adcsMode">--</span>
            </div>
        </div>
        <div class="card">
            <h2>Communications</h2>
            <div class="metric">
                <span class="label">Signal</span>
                <span class="value" id="commsSignal">--</span>
            </div>
        </div>
    </div>
    <div class="log">
        <h2>Event Log</h2>
        <div id="logEntries"></div>
    </div>
    <script>
        const ws = new WebSocket(`ws://${window.location.host}/ws/telemetry`);
        const statusDot = document.getElementById('statusDot');
        const statusText = document.getElementById('statusText');
        const logEntries = document.getElementById('logEntries');

        const limits = {
            battery_voltage: { warning_low: 24.0, warning_high: 30.0, critical_low: 22.0, critical_high: 31.0 },
            obc_temperature: { warning_low: 5.0, warning_high: 45.0, critical_low: 0.0, critical_high: 50.0 },
            comms_signal_strength: { warning_low: -100.0, critical_low: -110.0 }
        };

        function getSeverityClass(value, paramLimits) {
            if (paramLimits.critical_low !== undefined && value <= paramLimits.critical_low) return 'critical';
            if (paramLimits.critical_high !== undefined && value >= paramLimits.critical_high) return 'critical';
            if (paramLimits.warning_low !== undefined && value <= paramLimits.warning_low) return 'warning';
            if (paramLimits.warning_high !== undefined && value >= paramLimits.warning_high) return 'warning';
            return 'nominal';
        }

        ws.onopen = () => {
            statusDot.classList.add('connected');
            statusText.textContent = 'Connected - Receiving telemetry';
            addLog('WebSocket connected');
        };

        ws.onclose = () => {
            statusDot.classList.remove('connected');
            statusText.textContent = 'Disconnected';
            addLog('WebSocket disconnected');
        };

        ws.onmessage = (event) => {
            const frame = JSON.parse(event.data);
            updateTelemetry(frame);
        };

        function updateTelemetry(frame) {
            const battEl = document.getElementById('batteryVoltage');
            const tempEl = document.getElementById('obcTemperature');
            const adcsEl = document.getElementById('adcsMode');
            const commsEl = document.getElementById('commsSignal');

            battEl.textContent = frame.battery_voltage.toFixed(2) + ' V';
            battEl.className = 'value ' + getSeverityClass(frame.battery_voltage, limits.battery_voltage);

            tempEl.textContent = frame.obc_temperature.toFixed(2) + ' C';
            tempEl.className = 'value ' + getSeverityClass(frame.obc_temperature, limits.obc_temperature);

            adcsEl.textContent = frame.adcs_mode;
            adcsEl.className = 'value nominal';

            commsEl.textContent = frame.comms_signal_strength.toFixed(2) + ' dBm';
            commsEl.className = 'value ' + getSeverityClass(frame.comms_signal_strength, limits.comms_signal_strength);
        }

        function addLog(message) {
            const entry = document.createElement('div');
            entry.className = 'log-entry';
            entry.textContent = `[${new Date().toISOString().substr(11, 8)}] ${message}`;
            logEntries.prepend(entry);
            if (logEntries.children.length > 50) {
                logEntries.removeChild(logEntries.lastChild);
            }
        }
    </script>
</body>
</html>

Your simulator is ready to produce telemetry, your FastAPI app is wired up, and your viewer is waiting to render the stream. In the next step you'll build the procedure engine so the server can execute corrective sequences against the simulator.

Build the Procedure Engine

Your simulator is streaming live telemetry, but right now there's no way to respond when something goes wrong. In real satellite operations, corrective actions follow strict step-by-step procedures defined in advance by the operations team.

In this step, you'll build a declarative procedure engine. Procedures are defined as JSON data (not hard-coded logic), and the engine executes each step with timeout enforcement so a hanging command never blocks the entire system.

In this step, get ready to:

• Define 5 satellite procedures as JSON configuration.
• Build the procedure engine with async execution and timeout enforcement.
• Test procedure execution via the REST API.

Create procedure definitions as JSON data

In satellite operations, procedures are pre-validated sequences of commands that the team approves before launch. Storing them as data means you can update operational behavior without touching application code.

• Create a new file called procedures.json inside the config/ directory.
• Add the following 5 procedure definitions:

{
    "procedures": [
        {
            "name": "battery_recovery",
            "description": "Reset battery controller to recover nominal voltage",
            "steps": [
                {
                    "name": "verify_adcs_safe",
                    "action": "switch_adcs_nominal",
                    "precondition": "adcs_mode != NOMINAL",
                    "timeout_seconds": 10.0
                },
                {
                    "name": "reset_battery",
                    "action": "reset_battery_controller",
                    "timeout_seconds": 15.0
                }
            ]
        },
        {
            "name": "thermal_protection",
            "description": "Activate heater to recover OBC temperature",
            "steps": [
                {
                    "name": "activate_heater",
                    "action": "activate_heater",
                    "timeout_seconds": 10.0
                }
            ]
        },
        {
            "name": "comms_recovery",
            "description": "Reset communications subsystem",
            "steps": [
                {
                    "name": "reset_comms_module",
                    "action": "reset_comms",
                    "timeout_seconds": 10.0
                }
            ]
        },
        {
            "name": "full_health_check",
            "description": "Run complete satellite health verification",
            "steps": [
                {
                    "name": "reset_battery",
                    "action": "reset_battery_controller",
                    "timeout_seconds": 10.0
                },
                {
                    "name": "stabilize_thermal",
                    "action": "activate_heater",
                    "timeout_seconds": 10.0
                },
                {
                    "name": "verify_comms",
                    "action": "reset_comms",
                    "timeout_seconds": 10.0
                }
            ]
        },
        {
            "name": "emergency_power_cycle",
            "description": "Emergency OBC power cycle for critical failures",
            "steps": [
                {
                    "name": "power_cycle",
                    "action": "power_cycle_obc",
                    "timeout_seconds": 20.0
                },
                {
                    "name": "verify_nominal",
                    "action": "switch_adcs_nominal",
                    "timeout_seconds": 10.0
                }
            ]
        }
    ]
}

What does this configuration define?

Each procedure has a name, a description, and an ordered list of steps. Each step specifies an action (a command to send to the satellite), an optional precondition, and a timeout in seconds.

• battery_recovery first ensures ADCS is safe, then resets the battery controller.
• emergency_power_cycle is reserved for critical failures and gets a longer 20-second timeout on the power cycle step.
• full_health_check exercises all subsystems in sequence.

• Save config/procedures.json.
• Confirm the file exists by running:

ls config/

You should see procedures.json listed in the output.

Don't see the file?

Make sure you created the file inside the config/ directory, not in the project root. The path should be config/procedures.json.

Help me find where my procedures.json ended up

Build the procedure engine class

The engine loads procedures from JSON at startup and executes them asynchronously. The critical design choice is timeout enforcement. A satellite command that hangs should never block the entire operations loop.

• Create a new file called procedures.py in the project root (groundops-console/).
• Add the imports, class definition, and procedure loading logic:

import asyncio
import datetime
import json

from models import (
    Procedure,
    ProcedureOutcome,
    ProcedureStepResult,
    StepOutcome,
)


class ProcedureEngine:
    def __init__(self, config_path: str = "config/procedures.json"):
        self.procedures = self._load_procedures(config_path)
        self.history: list[ProcedureOutcome] = []

    def _load_procedures(self, config_path: str) -> dict[str, Procedure]:
        with open(config_path) as f:
            data = json.load(f)
        return {p["name"]: Procedure(**p) for p in data["procedures"]}

    def get_procedure(self, name: str):
        return self.procedures.get(name)

What does this code do?

• _load_procedures() reads the JSON file and builds a dictionary mapping procedure names to Procedure model instances.
• The history list stores every procedure execution result for audit purposes.
• get_procedure() provides safe dictionary lookup, returning None if the procedure name doesn't exist.

• Now add the execute() method below get_procedure():

    async def execute(self, procedure_name: str, command_handler) -> ProcedureOutcome:
        procedure = self.procedures[procedure_name]
        started_at = datetime.datetime.now(datetime.timezone.utc)
        step_outcomes = []

        for step in procedure.steps:
            outcome = await self._execute_step(step, command_handler)
            step_outcomes.append(outcome)
            if outcome.result != ProcedureStepResult.SUCCESS:
                break

        completed_at = datetime.datetime.now(datetime.timezone.utc)
        success = all(
            o.result == ProcedureStepResult.SUCCESS for o in step_outcomes
        )

        result = ProcedureOutcome(
            procedure_name=procedure_name,
            started_at=started_at,
            completed_at=completed_at,
            step_outcomes=step_outcomes,
            success=success,
        )
        self.history.append(result)
        return result

What does execute() do?

The method iterates through the procedure's steps in order. If any step returns a result other than SUCCESS, it breaks immediately. This is the "fail fast" pattern used in real spacecraft operations.

The command_handler parameter is a callable (the simulator's command() method) that actually sends commands to the satellite. This decoupled design means the engine can work with any command interface.

• Add the _execute_step() method below execute():

    async def _execute_step(self, step, command_handler) -> StepOutcome:
        timestamp = datetime.datetime.now(datetime.timezone.utc)
        try:
            loop = asyncio.get_running_loop()
            result = await asyncio.wait_for(
                loop.run_in_executor(None, command_handler, step.action),
                timeout=step.timeout_seconds,
            )
            if result:
                return StepOutcome(
                    step_name=step.name,
                    result=ProcedureStepResult.SUCCESS,
                    message=f"Action '{step.action}' completed successfully",
                    timestamp=timestamp,
                )
            else:
                return StepOutcome(
                    step_name=step.name,
                    result=ProcedureStepResult.FAILURE,
                    message=f"Action '{step.action}' returned failure",
                    timestamp=timestamp,
                )
        except asyncio.TimeoutError:
            return StepOutcome(
                step_name=step.name,
                result=ProcedureStepResult.TIMEOUT,
                message=f"Action '{step.action}' timed out after {step.timeout_seconds}s",
                timestamp=timestamp,
            )

Why use run_in_executor with wait_for?

The simulator's command() method is synchronous. run_in_executor() offloads it to a thread pool so it becomes awaitable.

asyncio.wait_for() wraps that awaitable with a timeout. If the command takes longer than timeout_seconds, it raises asyncio.TimeoutError and the step is recorded as TIMEOUT. This prevents a single hanging command from blocking the entire telemetry loop.

• Save procedures.py.
• Confirm the module loads correctly by running:

python -c "from procedures import ProcedureEngine; e = ProcedureEngine(); print(list(e.procedures.keys()))"

You should see the 5 procedure names printed: battery_recovery, thermal_protection, comms_recovery, full_health_check, and emergency_power_cycle.

Seeing a FileNotFoundError?

Make sure you're running the command from the groundops-console/ directory (where config/ lives). The engine expects config/procedures.json as a relative path.

Also check that your JSON is valid. A missing comma or bracket will cause a json.JSONDecodeError.

Help me debug my ProcedureEngine import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved procedures.py before moving on.

ⓧ I'd like to double check the full code

Here's the complete procedures.py file:

import asyncio
import datetime
import json

from models import (
    Procedure,
    ProcedureOutcome,
    ProcedureStepResult,
    StepOutcome,
)


class ProcedureEngine:
    def __init__(self, config_path: str = "config/procedures.json"):
        self.procedures = self._load_procedures(config_path)
        self.history: list[ProcedureOutcome] = []

    def _load_procedures(self, config_path: str) -> dict[str, Procedure]:
        with open(config_path) as f:
            data = json.load(f)
        return {p["name"]: Procedure(**p) for p in data["procedures"]}

    def get_procedure(self, name: str):
        return self.procedures.get(name)

    async def execute(self, procedure_name: str, command_handler) -> ProcedureOutcome:
        procedure = self.procedures[procedure_name]
        started_at = datetime.datetime.now(datetime.timezone.utc)
        step_outcomes = []

        for step in procedure.steps:
            outcome = await self._execute_step(step, command_handler)
            step_outcomes.append(outcome)
            if outcome.result != ProcedureStepResult.SUCCESS:
                break

        completed_at = datetime.datetime.now(datetime.timezone.utc)
        success = all(
            o.result == ProcedureStepResult.SUCCESS for o in step_outcomes
        )

        result = ProcedureOutcome(
            procedure_name=procedure_name,
            started_at=started_at,
            completed_at=completed_at,
            step_outcomes=step_outcomes,
            success=success,
        )
        self.history.append(result)
        return result

    async def _execute_step(self, step, command_handler) -> StepOutcome:
        timestamp = datetime.datetime.now(datetime.timezone.utc)
        try:
            loop = asyncio.get_running_loop()
            result = await asyncio.wait_for(
                loop.run_in_executor(None, command_handler, step.action),
                timeout=step.timeout_seconds,
            )
            if result:
                return StepOutcome(
                    step_name=step.name,
                    result=ProcedureStepResult.SUCCESS,
                    message=f"Action '{step.action}' completed successfully",
                    timestamp=timestamp,
                )
            else:
                return StepOutcome(
                    step_name=step.name,
                    result=ProcedureStepResult.FAILURE,
                    message=f"Action '{step.action}' returned failure",
                    timestamp=timestamp,
                )
        except asyncio.TimeoutError:
            return StepOutcome(
                step_name=step.name,
                result=ProcedureStepResult.TIMEOUT,
                message=f"Action '{step.action}' timed out after {step.timeout_seconds}s",
                timestamp=timestamp,
            )

Test procedure execution via the REST API

Your main.py already has GET /api/procedures and POST /api/procedures/{name}/execute endpoints wired up from the previous step. Now that the engine exists, these endpoints are functional.

• Start (or restart) the server by running:

uvicorn main:app --reload

• Open a second terminal tab. Make sure your virtual environment is activated in it.
• List available procedures by running:

curl http://localhost:8000/api/procedures

You should see a JSON response listing all 5 procedure names.

• Execute the battery recovery procedure by running:

curl -X POST http://localhost:8000/api/procedures/battery_recovery/execute

You should see a JSON response with step_outcomes showing each step's result as SUCCESS and "success": true at the top level. The simulator's battery voltage has been reset to nominal.

Seeing a connection error or 500 response?

• If the server didn't start, check the terminal running uvicorn for import errors. Make sure procedures.py exists in the project root (not inside config/).
• If you see a 500 error, check the uvicorn terminal output for a traceback. A common cause is a typo in config/procedures.json that prevents the JSON from parsing.
• Make sure you're running curl from a separate terminal tab while uvicorn is still running in the first.

Help me debug my procedure execution endpoint

Your procedure engine is operational. It can load procedure definitions from JSON and execute them with timeout enforcement against the simulator. Next up, you'll build the telemetry monitor and resolver system that automatically triggers these procedures when anomalies are detected.

Build the Telemetry Monitor and Resolver System

Your simulator produces telemetry and the procedure engine executes corrective steps. But nothing connects the two yet. The system needs a decision layer that watches every telemetry frame, decides when values cross safety boundaries, and fires the correct procedure automatically.

In this step, you'll build the two components that close this automation loop. The telemetry monitor evaluates each frame against JSON-configurable alarm limits. The resolver system maps detected anomalies to corrective procedures.

In this step, get ready to:

• Create JSON-configurable alarm limits and a telemetry monitor.
• Build a resolver system that maps alerts to corrective procedures.
• Verify the full automation chain detects anomalies and resolves them.

Create the alarm limits and telemetry monitor

In satellite operations, every telemetry parameter has defined safe operating bounds. Crossing a warning threshold triggers attention. Crossing a critical threshold triggers immediate intervention.

Why JSON for alarm limits?

Storing alarm thresholds in JSON keeps them separate from code. Operations teams can adjust limits between passes without deploying new software. This is standard practice in ground-segment systems where configuration is data, not code.

• Create the alarm limits configuration by adding this to a new file called config/alarm_limits.json:

{
    "battery_voltage": {
        "warning_low": 24.0,
        "warning_high": 30.0,
        "critical_low": 22.0,
        "critical_high": 31.0
    },
    "obc_temperature": {
        "warning_low": 5.0,
        "warning_high": 45.0,
        "critical_low": 0.0,
        "critical_high": 50.0
    },
    "comms_signal_strength": {
        "warning_low": -100.0,
        "critical_low": -110.0
    }
}

What does this configuration define?

Each parameter has warning and critical thresholds. Battery voltage is safe between 24V and 30V (warning bounds). Below 22V or above 31V is critical. The comms signal only has low thresholds because signal strength can only degrade, not spike.

• Save config/alarm_limits.json.

Now you need the monitor class that loads these limits and evaluates each telemetry frame against them.

• Create a new file called monitor.py in the project root and add the following code:

import json

from models import Alert, AlertSeverity, TelemetryFrame


class TelemetryMonitor:
    def __init__(self, config_path: str = "config/alarm_limits.json"):
        self.limits = self._load_limits(config_path)
        self.alerts: list[Alert] = []
        self.active_alerts: dict[str, Alert] = {}

    def _load_limits(self, config_path: str) -> dict:
        with open(config_path) as f:
            return json.load(f)

    def evaluate(self, frame: TelemetryFrame) -> list[Alert]:
        new_alerts = []
        for param_name, limits in self.limits.items():
            value = getattr(frame, param_name, None)
            if value is None or isinstance(value, str):
                continue

            alert_key = f"{frame.satellite_id}:{param_name}"
            severity = self._check_limits(value, limits)

            if severity != AlertSeverity.NOMINAL:
                alert = Alert(
                    timestamp=frame.timestamp,
                    satellite_id=frame.satellite_id,
                    parameter=param_name,
                    value=value,
                    severity=severity,
                    message=f"{param_name} is {severity.value}: {value}",
                )
                self.alerts.append(alert)
                self.active_alerts[alert_key] = alert
                new_alerts.append(alert)
            else:
                if alert_key in self.active_alerts:
                    self.active_alerts[alert_key].resolved = True
                    del self.active_alerts[alert_key]

        return new_alerts

    def _check_limits(self, value: float, limits: dict) -> AlertSeverity:
        if "critical_low" in limits and value <= limits["critical_low"]:
            return AlertSeverity.CRITICAL
        if "critical_high" in limits and value >= limits["critical_high"]:
            return AlertSeverity.CRITICAL
        if "warning_low" in limits and value <= limits["warning_low"]:
            return AlertSeverity.WARNING
        if "warning_high" in limits and value >= limits["warning_high"]:
            return AlertSeverity.WARNING
        return AlertSeverity.NOMINAL

What does this code do?

The TelemetryMonitor loads alarm limits from JSON and evaluates each telemetry frame. For every parameter, it checks whether the current value exceeds warning or critical bounds using _check_limits().

The alert lifecycle is key: when a parameter goes out of bounds, an alert is created and stored in active_alerts keyed by satellite_id:parameter. When the value returns to nominal, the alert is marked resolved and removed from active tracking.

• Save monitor.py.
• Confirm the module loads correctly by running:

python -c "from monitor import TelemetryMonitor; m = TelemetryMonitor(); print(f'Loaded {len(m.limits)} parameters')"

You should see Loaded 3 parameters confirming the monitor reads all three alarm limit entries.

Seeing a FileNotFoundError?

Make sure you're running the command from the groundops-console/ directory (the same folder that contains config/). The monitor expects config/alarm_limits.json relative to where you run the command.

Also check the JSON file for trailing commas or syntax errors. JSON does not allow trailing commas after the last item in an object.

Help me debug my TelemetryMonitor import error.

✔️ Awesome, I've got everything!

Great. Make sure you've saved both config/alarm_limits.json and monitor.py.

ⓧ I'd like to double check the full code

Here is config/alarm_limits.json:

{
    "battery_voltage": {
        "warning_low": 24.0,
        "warning_high": 30.0,
        "critical_low": 22.0,
        "critical_high": 31.0
    },
    "obc_temperature": {
        "warning_low": 5.0,
        "warning_high": 45.0,
        "critical_low": 0.0,
        "critical_high": 50.0
    },
    "comms_signal_strength": {
        "warning_low": -100.0,
        "critical_low": -110.0
    }
}

Here is monitor.py:

import json

from models import Alert, AlertSeverity, TelemetryFrame


class TelemetryMonitor:
    def __init__(self, config_path: str = "config/alarm_limits.json"):
        self.limits = self._load_limits(config_path)
        self.alerts: list[Alert] = []
        self.active_alerts: dict[str, Alert] = {}

    def _load_limits(self, config_path: str) -> dict:
        with open(config_path) as f:
            return json.load(f)

    def evaluate(self, frame: TelemetryFrame) -> list[Alert]:
        new_alerts = []
        for param_name, limits in self.limits.items():
            value = getattr(frame, param_name, None)
            if value is None or isinstance(value, str):
                continue

            alert_key = f"{frame.satellite_id}:{param_name}"
            severity = self._check_limits(value, limits)

            if severity != AlertSeverity.NOMINAL:
                alert = Alert(
                    timestamp=frame.timestamp,
                    satellite_id=frame.satellite_id,
                    parameter=param_name,
                    value=value,
                    severity=severity,
                    message=f"{param_name} is {severity.value}: {value}",
                )
                self.alerts.append(alert)
                self.active_alerts[alert_key] = alert
                new_alerts.append(alert)
            else:
                if alert_key in self.active_alerts:
                    self.active_alerts[alert_key].resolved = True
                    del self.active_alerts[alert_key]

        return new_alerts

    def _check_limits(self, value: float, limits: dict) -> AlertSeverity:
        if "critical_low" in limits and value <= limits["critical_low"]:
            return AlertSeverity.CRITICAL
        if "critical_high" in limits and value >= limits["critical_high"]:
            return AlertSeverity.CRITICAL
        if "warning_low" in limits and value <= limits["warning_low"]:
            return AlertSeverity.WARNING
        if "warning_high" in limits and value >= limits["warning_high"]:
            return AlertSeverity.WARNING
        return AlertSeverity.NOMINAL

Create the resolver rules and resolver system

The monitor detects anomalies, but it does not know what to do about them. The resolver system bridges this gap by mapping each alert condition (parameter + severity) to a corrective procedure name.

• Create the resolver rules configuration by adding this to a new file called config/resolvers.json:

{
    "rules": [
        {
            "name": "low_battery_resolver",
            "parameter": "battery_voltage",
            "severity": "WARNING",
            "procedure_name": "battery_recovery"
        },
        {
            "name": "critical_battery_resolver",
            "parameter": "battery_voltage",
            "severity": "CRITICAL",
            "procedure_name": "emergency_power_cycle"
        },
        {
            "name": "thermal_warning_resolver",
            "parameter": "obc_temperature",
            "severity": "WARNING",
            "procedure_name": "thermal_protection"
        },
        {
            "name": "comms_degraded_resolver",
            "parameter": "comms_signal_strength",
            "severity": "WARNING",
            "procedure_name": "comms_recovery"
        }
    ]
}

What do these rules do?

Each rule maps a specific alert condition to the procedure that corrects it. A WARNING on battery voltage triggers battery_recovery (gentle reset). A CRITICAL battery alert triggers emergency_power_cycle (aggressive intervention).

This decoupled design means operations teams can change which procedure responds to which alert by editing JSON. No code changes required.

• Save config/resolvers.json.

Now build the resolver class that loads these rules and matches alerts to procedures.

• Create a new file called resolvers.py in the project root and add the following code:

import datetime
import json

from models import Alert, ProcedureOutcome, ResolverRule


class ResolverSystem:
    def __init__(self, config_path: str = "config/resolvers.json"):
        self.rules = self._load_rules(config_path)
        self.history: list[dict] = []

    def _load_rules(self, config_path: str) -> list[ResolverRule]:
        with open(config_path) as f:
            data = json.load(f)
        return [ResolverRule(**r) for r in data["rules"]]

    def match(self, alert: Alert) -> str | None:
        for rule in self.rules:
            if rule.parameter == alert.parameter and rule.severity == alert.severity:
                return rule.procedure_name
        return None

    def log_action(self, alert: Alert, outcome: ProcedureOutcome):
        self.history.append({
            "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "alert_parameter": alert.parameter,
            "alert_severity": alert.severity.value,
            "alert_value": alert.value,
            "procedure_executed": outcome.procedure_name,
            "procedure_success": outcome.success,
        })

What does this code do?

The match() method iterates through loaded rules and returns the procedure name when it finds a rule matching the alert's parameter and severity. If no rule matches, it returns None (the system ignores the alert).

The log_action() method records every resolver action in a history list. This creates an audit trail showing which alerts triggered which procedures and whether they succeeded.

• Save resolvers.py.
• Confirm both modules load by running:

python -c "from resolvers import ResolverSystem; r = ResolverSystem(); print(f'Loaded {len(r.rules)} resolver rules')"

You should see Loaded 4 resolver rules confirming all four rules parsed correctly.

Seeing an import error on ResolverRule?

Check that models.py has the ResolverRule class defined. It should have fields name, parameter, severity, and procedure_name. Also verify your JSON severity values are uppercase strings (WARNING, CRITICAL) matching the AlertSeverity enum.

Help me fix my ResolverSystem import.

✔️ Awesome, I've got everything!

Great. Make sure you've saved both config/resolvers.json and resolvers.py.

ⓧ I'd like to double check the full code

Here is config/resolvers.json:

{
    "rules": [
        {
            "name": "low_battery_resolver",
            "parameter": "battery_voltage",
            "severity": "WARNING",
            "procedure_name": "battery_recovery"
        },
        {
            "name": "critical_battery_resolver",
            "parameter": "battery_voltage",
            "severity": "CRITICAL",
            "procedure_name": "emergency_power_cycle"
        },
        {
            "name": "thermal_warning_resolver",
            "parameter": "obc_temperature",
            "severity": "WARNING",
            "procedure_name": "thermal_protection"
        },
        {
            "name": "comms_degraded_resolver",
            "parameter": "comms_signal_strength",
            "severity": "WARNING",
            "procedure_name": "comms_recovery"
        }
    ]
}

Here is resolvers.py:

import datetime
import json

from models import Alert, ProcedureOutcome, ResolverRule


class ResolverSystem:
    def __init__(self, config_path: str = "config/resolvers.json"):
        self.rules = self._load_rules(config_path)
        self.history: list[dict] = []

    def _load_rules(self, config_path: str) -> list[ResolverRule]:
        with open(config_path) as f:
            data = json.load(f)
        return [ResolverRule(**r) for r in data["rules"]]

    def match(self, alert: Alert) -> str | None:
        for rule in self.rules:
            if rule.parameter == alert.parameter and rule.severity == alert.severity:
                return rule.procedure_name
        return None

    def log_action(self, alert: Alert, outcome: ProcedureOutcome):
        self.history.append({
            "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "alert_parameter": alert.parameter,
            "alert_severity": alert.severity.value,
            "alert_value": alert.value,
            "procedure_executed": outcome.procedure_name,
            "procedure_success": outcome.success,
        })

Verify the automation chain end-to-end

The monitor and resolver are built. Your main.py already contains the integration code that connects them to the telemetry loop. Time to prove the full chain works: simulator produces degraded telemetry, monitor detects it, resolver matches it to a procedure.

• Verify the automation chain by running this test script:

python -c "
from simulator import SatelliteSimulator
from monitor import TelemetryMonitor
from resolvers import ResolverSystem

sim = SatelliteSimulator()
sim.battery_voltage = 23.0
frame = sim.generate_frame()

mon = TelemetryMonitor()
res = ResolverSystem()
alerts = mon.evaluate(frame)

for a in alerts:
    proc = res.match(a)
    if proc:
        print(f'Alert: {a.parameter} {a.severity.value} -> Procedure: {proc}')
"

What does this test do?

This script simulates a battery voltage drop to 23.0V (below the 24.0V warning threshold). The monitor evaluates the frame and generates a WARNING alert for battery_voltage. The resolver matches that alert to battery_recovery per the rules you configured.

You should see output similar to:

Alert: battery_voltage WARNING -> Procedure: battery_recovery

This confirms the full detection-to-resolution chain works. When the server runs, this same logic fires automatically every second inside telemetry_loop().

No output printed?

The simulator's generate_frame() applies random drift to the battery voltage you set. If drift pushes it back above 24.0V, no alert fires. Try setting the voltage lower (e.g. 21.0) to guarantee a CRITICAL alert.

Also check that config/alarm_limits.json has the correct warning_low value of 24.0 for battery_voltage.

Help me debug why the monitor isn't generating alerts.

Your monitor detects anomalies and the resolver maps them to procedures. The automation loop in main.py already ties these together with the procedure engine. Next up, you'll build the autopilot sequencer that runs entire procedure sequences unattended, then prove the whole system works with a test-as-you-fly pytest suite.

Build the Autopilot Sequencer and Test Suite

Your telemetry monitor now detects anomalies and the resolver system automatically fires corrective procedures.

The final piece of the automation engine is the autopilot sequencer. This component runs multiple procedures in sequence without human intervention. Think of it as the unattended overnight operator: you hand it a checklist of procedures, it executes them one by one, and reports back whether everything succeeded.

In this step, get ready to:

• Build the autopilot sequencer module with abort capability.
• Write a test-as-you-fly pytest suite exercising the full automation chain.
• Run all tests to prove the system works the same way it runs operationally.

Build the autopilot sequencer

If you look at main.py, it already imports from autopilot and wires up the REST endpoints. The module just doesn't exist yet.

• Create a new file called autopilot.py in the root of your groundops-console folder and add the following code:

import datetime

from models import AutopilotResult, ProcedureOutcome
from procedures import ProcedureEngine


class AutopilotSequencer:
    def __init__(self, procedure_engine: ProcedureEngine):
        self.engine = procedure_engine
        self.history: list[AutopilotResult] = []
        self._running = False

    async def execute_sequence(
        self, sequence_name: str, procedure_names: list[str], command_handler
    ) -> AutopilotResult:
        self._running = True
        started_at = datetime.datetime.now(datetime.timezone.utc)
        outcomes: list[ProcedureOutcome] = []

        for proc_name in procedure_names:
            if not self._running:
                break
            outcome = await self.engine.execute(proc_name, command_handler)
            outcomes.append(outcome)

        completed_at = datetime.datetime.now(datetime.timezone.utc)
        all_succeeded = all(o.success for o in outcomes)

        result = AutopilotResult(
            sequence_name=sequence_name,
            started_at=started_at,
            completed_at=completed_at,
            procedure_outcomes=outcomes,
            all_succeeded=all_succeeded,
        )
        self.history.append(result)
        self._running = False
        return result

    def abort(self):
        self._running = False

What does this code do?

• execute_sequence() takes a named sequence of procedures and runs them one at a time using the existing ProcedureEngine.
• Before each procedure, it checks self._running. If another coroutine calls abort(), the loop stops early.
• The method returns an AutopilotResult containing per-procedure outcomes and an overall all_succeeded flag.

• Save autopilot.py.
• Confirm the module loads by running this in your terminal:

python -c "from autopilot import AutopilotSequencer; print('Autopilot OK')"

You should see Autopilot OK printed with no import errors.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still activated. You should see (venv) in your terminal prompt.

Check that autopilot.py is in the root of groundops-console/ (same level as models.py and procedures.py), not inside a subfolder.

help me debug this ModuleNotFoundError for autopilot

✔️ Awesome, I've got everything!

Great. Make sure you've saved autopilot.py before moving on.

ⓧ I'd like to double check the full code

import datetime

from models import AutopilotResult, ProcedureOutcome
from procedures import ProcedureEngine


class AutopilotSequencer:
    def __init__(self, procedure_engine: ProcedureEngine):
        self.engine = procedure_engine
        self.history: list[AutopilotResult] = []
        self._running = False

    async def execute_sequence(
        self, sequence_name: str, procedure_names: list[str], command_handler
    ) -> AutopilotResult:
        self._running = True
        started_at = datetime.datetime.now(datetime.timezone.utc)
        outcomes: list[ProcedureOutcome] = []

        for proc_name in procedure_names:
            if not self._running:
                break
            outcome = await self.engine.execute(proc_name, command_handler)
            outcomes.append(outcome)

        completed_at = datetime.datetime.now(datetime.timezone.utc)
        all_succeeded = all(o.success for o in outcomes)

        result = AutopilotResult(
            sequence_name=sequence_name,
            started_at=started_at,
            completed_at=completed_at,
            procedure_outcomes=outcomes,
            all_succeeded=all_succeeded,
        )
        self.history.append(result)
        self._running = False
        return result

    def abort(self):
        self._running = False

Write the test-as-you-fly pytest suite

In satellite operations, the testing methodology is called test-as-you-fly. Tests must exercise the identical code paths that run operationally. No mocks of core logic are allowed.

What is test-as-you-fly?

Real satellite ground systems run acceptance tests against the actual flight software, not simulated stand-ins. If a test passes, the team has confidence the procedure will work during a live contact window.

Your tests will instantiate the real SatelliteSimulator, TelemetryMonitor, ResolverSystem, and ProcedureEngine. No mocking.

• Create a new file called tests/test_autopilot.py and add the imports and first test:

import asyncio

from autopilot import AutopilotSequencer
from models import AlertSeverity
from monitor import TelemetryMonitor
from procedures import ProcedureEngine
from resolvers import ResolverSystem
from simulator import SatelliteSimulator


class TestAutopilotEndToEnd:
    """Test-as-you-fly: exercises the full automation chain."""

    def test_autopilot_executes_sequence_successfully(self):
        result = asyncio.run(self._run_sequence_test())
        assert result.all_succeeded is True
        assert len(result.procedure_outcomes) == 3
        for outcome in result.procedure_outcomes:
            assert outcome.success is True

    async def _run_sequence_test(self):
        sim = SatelliteSimulator()
        engine = ProcedureEngine()
        autopilot = AutopilotSequencer(engine)
        return await autopilot.execute_sequence(
            sequence_name="health_check_sequence",
            procedure_names=["battery_recovery", "thermal_protection", "comms_recovery"],
            command_handler=sim.command,
        )

What does this test do?

• It creates a real simulator, a real procedure engine, and a real autopilot sequencer.
• It asks the autopilot to run three procedures in sequence: battery recovery, thermal protection, and comms recovery.
• It asserts all three succeeded. If any step timed out or failed, all_succeeded would be False and the test would fail.
• The pattern asyncio.run(self._run_sequence_test()) lets you call async code from a regular sync test method without needing extra plugins.

You won't run the tests yet. The next chunk adds two more integration tests that complete the suite.

• Still in tests/test_autopilot.py, add the resolver-fires-on-alert test and the full-chain integration test below the first test's helper method:

    def test_resolver_fires_on_alert(self):
        result = asyncio.run(self._run_resolver_test())
        assert result["alert_detected"] is True
        assert result["procedure_fired"] is True
        assert result["procedure_success"] is True

    async def _run_resolver_test(self):
        sim = SatelliteSimulator()
        monitor = TelemetryMonitor()
        engine = ProcedureEngine()
        resolver = ResolverSystem()
        sim.battery_voltage = 23.0
        frame = sim.generate_frame()
        alerts = monitor.evaluate(frame)
        alert_detected = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.WARNING
            for a in alerts
        )
        procedure_fired = False
        procedure_success = False
        for alert in alerts:
            proc_name = resolver.match(alert)
            if proc_name:
                outcome = await engine.execute(proc_name, sim.command)
                resolver.log_action(alert, outcome)
                procedure_fired = True
                procedure_success = outcome.success
        return {"alert_detected": alert_detected, "procedure_fired": procedure_fired, "procedure_success": procedure_success}

    def test_full_automation_chain(self):
        result = asyncio.run(self._run_full_chain_test())
        assert result["initial_alert"] is True
        assert result["resolver_acted"] is True
        assert result["recovery_confirmed"] is True

    async def _run_full_chain_test(self):
        sim = SatelliteSimulator()
        monitor = TelemetryMonitor()
        engine = ProcedureEngine()
        resolver = ResolverSystem()
        sim.battery_voltage = 21.5
        frame = sim.generate_frame()
        alerts = monitor.evaluate(frame)
        initial_alert = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.CRITICAL
            for a in alerts
        )
        resolver_acted = False
        for alert in alerts:
            proc_name = resolver.match(alert)
            if proc_name:
                outcome = await engine.execute(proc_name, sim.command)
                resolver.log_action(alert, outcome)
                resolver_acted = outcome.success
        recovery_frame = sim.generate_frame()
        recovery_alerts = monitor.evaluate(recovery_frame)
        battery_still_critical = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.CRITICAL
            for a in recovery_alerts
        )
        return {"initial_alert": initial_alert, "resolver_acted": resolver_acted, "recovery_confirmed": not battery_still_critical}

What do these tests prove?

test_resolver_fires_on_alert forces the simulator's battery to 23.0V (below the 24.0V warning threshold), generates a telemetry frame, and verifies the monitor raises an alert. It then checks that the resolver matches the alert to battery_recovery and the procedure succeeds.

test_full_automation_chain is the most demanding. It degrades battery to 21.5V (below the 22.0V critical threshold), verifies the resolver fires emergency_power_cycle, and then confirms the satellite recovers by checking the next telemetry frame is no longer critical.

• Save tests/test_autopilot.py.

Seeing indentation errors?

Both new test methods and their helpers must be indented inside the TestAutopilotEndToEnd class (4 spaces). If your editor auto-indented them at the module level, the tests won't be discovered by pytest.

help me fix indentation issues in my test file

✔️ Awesome, I've got everything!

Great. Double check you've saved tests/test_autopilot.py before running the tests.

ⓧ I'd like to double check the full code

import asyncio

from autopilot import AutopilotSequencer
from models import AlertSeverity
from monitor import TelemetryMonitor
from procedures import ProcedureEngine
from resolvers import ResolverSystem
from simulator import SatelliteSimulator


class TestAutopilotEndToEnd:
    """Test-as-you-fly: exercises the full automation chain."""

    def test_autopilot_executes_sequence_successfully(self):
        result = asyncio.run(self._run_sequence_test())
        assert result.all_succeeded is True
        assert len(result.procedure_outcomes) == 3
        for outcome in result.procedure_outcomes:
            assert outcome.success is True

    async def _run_sequence_test(self):
        sim = SatelliteSimulator()
        engine = ProcedureEngine()
        autopilot = AutopilotSequencer(engine)
        return await autopilot.execute_sequence(
            sequence_name="health_check_sequence",
            procedure_names=["battery_recovery", "thermal_protection", "comms_recovery"],
            command_handler=sim.command,
        )

    def test_resolver_fires_on_alert(self):
        result = asyncio.run(self._run_resolver_test())
        assert result["alert_detected"] is True
        assert result["procedure_fired"] is True
        assert result["procedure_success"] is True

    async def _run_resolver_test(self):
        sim = SatelliteSimulator()
        monitor = TelemetryMonitor()
        engine = ProcedureEngine()
        resolver = ResolverSystem()
        sim.battery_voltage = 23.0
        frame = sim.generate_frame()
        alerts = monitor.evaluate(frame)
        alert_detected = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.WARNING
            for a in alerts
        )
        procedure_fired = False
        procedure_success = False
        for alert in alerts:
            proc_name = resolver.match(alert)
            if proc_name:
                outcome = await engine.execute(proc_name, sim.command)
                resolver.log_action(alert, outcome)
                procedure_fired = True
                procedure_success = outcome.success
        return {"alert_detected": alert_detected, "procedure_fired": procedure_fired, "procedure_success": procedure_success}

    def test_full_automation_chain(self):
        result = asyncio.run(self._run_full_chain_test())
        assert result["initial_alert"] is True
        assert result["resolver_acted"] is True
        assert result["recovery_confirmed"] is True

    async def _run_full_chain_test(self):
        sim = SatelliteSimulator()
        monitor = TelemetryMonitor()
        engine = ProcedureEngine()
        resolver = ResolverSystem()
        sim.battery_voltage = 21.5
        frame = sim.generate_frame()
        alerts = monitor.evaluate(frame)
        initial_alert = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.CRITICAL
            for a in alerts
        )
        resolver_acted = False
        for alert in alerts:
            proc_name = resolver.match(alert)
            if proc_name:
                outcome = await engine.execute(proc_name, sim.command)
                resolver.log_action(alert, outcome)
                resolver_acted = outcome.success
        recovery_frame = sim.generate_frame()
        recovery_alerts = monitor.evaluate(recovery_frame)
        battery_still_critical = any(
            a.parameter == "battery_voltage" and a.severity == AlertSeverity.CRITICAL
            for a in recovery_alerts
        )
        return {"initial_alert": initial_alert, "resolver_acted": resolver_acted, "recovery_confirmed": not battery_still_critical}

Run and verify the full automation chain

Every module in the automation engine is now in place. The test suite exercises the same code paths the live system uses during a satellite contact window.

• Run the test suite from your project root by running this command:

pytest tests/ -v

What should I see?

You should see all 3 tests pass with output showing each test name followed by PASSED:

• test_autopilot_executes_sequence_successfully PASSED
• test_resolver_fires_on_alert PASSED
• test_full_automation_chain PASSED

The final summary line should read 3 passed.

Tests failing or not discovered?

• If you see no tests ran, check that your pyproject.toml has pythonpath = ["."] and testpaths = ["tests"] set under [tool.pytest.ini_options].
• If you see ModuleNotFoundError for any of the project modules, make sure you are running pytest from the groundops-console/ directory (not from inside tests/).
• If test_full_automation_chain fails on recovery_confirmed, the random walk in generate_frame() may have drifted the value back below critical. This is extremely rare since emergency_power_cycle resets battery to 27.5V. Re-run the tests.

help me debug my failing pytest tests

All three tests pass. Your automation engine is now proven end-to-end: simulator to monitor to resolver to procedure engine to autopilot. The same code that just passed your tests is the code that runs live when the server is up.

Secret mission

Add Fault-Injection Testing

Your automation engine handles the happy path. But what happens when a command hangs indefinitely, or an operator hits the abort button mid-sequence? Add fault-injection tests that deliberately break things to prove the system degrades gracefully instead of crashing.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about. Decide whether to keep your project files for reference, or delete them to free up disk space.

Resources you used:

• The groundops-console/ project directory on your Desktop (source code, config files, tests, static assets).
• Python virtual environment (venv/ folder inside the project directory).
• The uvicorn development server process (if still running).

Keep everything running

No action needed. Choose this if you want to keep experimenting with the automation engine, adding new procedures, or extending the resolver rules.

• Your project files stay on your Desktop inside groundops-console/.
• The virtual environment preserves all installed packages.
• To restart the server later, activate the virtual environment and run uvicorn main:app --reload.

Pause - I'll come back to this later

Stop the running server process to free up system resources. Your files remain on disk for later.

• Switch to the terminal running the uvicorn server.
• Press Ctrl+C to stop the server.
• Deactivate the virtual environment by running:

deactivate

Your project folder stays intact. To resume later, navigate back into groundops-console/, activate the virtual environment, and start the server again.

Delete - I don't want to use this again

Remove all project files and the virtual environment. This frees up disk space and cleans up completely.

• If the uvicorn server is still running, press Ctrl+C in its terminal to stop it.
• Deactivate the virtual environment by running:

deactivate

• Navigate to your Desktop and delete the project folder.

macOS

cd ~/Desktop
rm -rf groundops-console

Windows

cd ~\Desktop
rmdir /s /q groundops-console

This removes all source code, configuration files, tests, the virtual environment, and all installed packages. Nothing else on your system is affected.

Nice Work!

Nice Work!

You just built a satellite operations automation engine from scratch. This is the kind of system that keeps spacecraft safe during unattended contact windows.

You've learned how to:

• Build a satellite telemetry simulator with realistic parameter drift and a JSON-configurable alarm monitor that evaluates telemetry frames against warning and critical thresholds.
• Build a declarative procedure engine with timeout enforcement and an automated resolver system that maps anomalies to corrective procedures without human intervention.
• Prove the full automation chain with a test-as-you-fly pytest suite that exercises the identical code path used in production, with no mocks of core logic.
• Secret Mission: Added fault-injection tests proving graceful degradation under timeout and abort conditions.

Ready to quiz yourself?