Notifications

To monitor the state of the system, especially during flight, it is strongly recommended to subscribe to system notifications via WebSockets.

For more details, see Notifications or the GitHub repository.

Subscribe notifications

The example demonstrates how to subscribe to live notifications to receive important system events, such as:

• Safety warnings

• System errors

def notifications_telemetry(
    on_message_callback: Callable[[Notification], None],
    access_token: str,
    max_messages: int | None = None,
) -> None:
    """Subscribes to notifications 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}/notifications/subscribe"
    try:
        with connect(
            ws_url, additional_headers={"Authorization": f"Bearer {access_token}"}
        ) as websocket:
            logging.info("Connected to notifications 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 notifications telemetry: {e}")

An overview of important notifications can be found in the Fotokite Sigma User Manual under https://manual.fotokite.com/failure-modes/messages-and-warnings

Get notification dictionary

A dictionary of all available notifications is provided in the following format:

• Code

• Severity

• Message (a human-readable interpretation of the notification for display in interfaces)

def dictionary(access_token: str) -> dict[str, object]:
    """Fetches the notifications dictionary.

    Args:
        access_token: The authentication token to use in the request.

    Returns:
        A dictionary containing notification definitions with their codes and descriptions.
        If the request fails or an exception occurs, logs the error and returns an empty dictionary.

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

    return {}