System Information

This section explains how to use system endpoints to identify the connected system, access component information, and subscribe to real-time updates via WebSockets.

For more details, see System or the GitHub repository.

Get system info

Retrieve basic system information, such as:

• System ID

• Timezone

• Hardware configuration (main board components)

def system_info(access_token: str = "") -> SystemMessage:
    """Fetches system information.

    Args:
        access_token: Bearer token for API authentication. Defaults to "".

    Returns:
        Parsed system information message if the request is successful.
        Empty dictionary if an error occurs.

    Raises:
        requests.HTTPError: If the HTTP request fails with a non-200 status code.
        Exception: For any other exceptions encountered during the request.
    """
    try:
        response = requests.get(
            f"{BASE_REST_API_URL}/system/info",
            headers={"Authorization": f"Bearer {access_token}"},
        )
        if response.status_code == 200:
            return cast(SystemMessage, response.json())
        else:
            response.raise_for_status()
    except Exception as e:
        logging.error(f"Error fetching system info: {e}")

    return {}

Subscribe system state

This example shows how to subscribe to frequently changing data, such as:

• Ground station battery level

• Kite battery level

• GPS status

It enables real-time monitoring of the system’s overall health and readiness.

def system_telemetry(
    on_message_callback: Callable[[SystemMessage], None],
    access_token: str = "",
    max_messages: int | None = None,
) -> None:
    """Subscribes to system info telemetry

    Args:
        on_message_callback: A callback function to handle incoming messages.
        access_token: The authentication token to use in the websocket connection.
        max_messages: The maximum number of messages to process. Defaults to None.
    """
    ws_url = f"{BASE_WEBSOCKET_API_URL}/system/state/subscribe?access_token={access_token}"
    try:
        with connect(
            ws_url, additional_headers={"Authorization": f"Bearer {access_token}"}
        ) as websocket:
            logging.info("Connected to system telemetry")
            count = 0
            while True:
                message = websocket.recv()
                data = json.loads(message)

                on_message_callback(data)

                count += 1
                if max_messages is not None and count >= max_messages:
                    break
    except Exception as e:
        logging.error(f"Error in system telemetry: {e}")