MavlinkClient

init

MavlinkClient.__init__( ground_control_station_ip: str = "192.168.8.10", mavlink_port: str = "14550", vehicle_ip: str = "192.168.8.1", communication_protocol: str = "udp", camera_streams: Optional[Dict[str, Dict[str, Any]]] = None, arm_enable: bool = False, dry_run: bool = False, log_level: str = "INFO", known_messages: List[str] = None, )

Instantiates a Mavlink client to handle message passing and command sending to/from a mavlink drone. Additionally sets up video stream manager for handling RTSP streams from the drone.

Arguments

ground_control_station_ip

str

IP address of the ground control station (default: ‘192.168.8.10’)

mavlink_port

str

port number for mavlink connection (default: ‘14550’)

vehicle_ip

str

IP address of the vehicle (default: ‘192.168.8.1’)

communication_protocol

str

communication protocol to use, one of (‘udp’|‘tcp’) (default: ‘udp’)

camera_streams

Optional[Dict[str, Dict[str, Any]]]

Dictionary of dictionaries containing the

arm_enable

bool

whether to enable arming the drone (default: False)

dry_run

bool

whether to run in dry run mode which bypasses arming exceptions (default: False)

log_level

str

logging level to be displayed to the user, one of (‘DEBUG’|‘INFO’|‘WARNING’|‘ERROR’),

known_messages

List[str]

list of expected message types to handle without warning (default: None)

get_msg

MavlinkClient.get_msg(msg_type: str)

Thread-safe method to get a message from msg_dict.

Arguments

msg_type

str

name of the message type to be retrieved.

Returns

msg

returns message if found, else None

update_setpoint

MavlinkClient.update_setpoint(controls, **kwargs)

Update the current setpoint of the drone as well as the currently active set of controls.

Arguments

controls

List[str]

controls

attempt_command

MavlinkClient.attempt_command( command_name: str, command_id: int, params: List[Any], confirmation: int = 0, max_attempts: int = 3, retry_delay: float = 1.0, )

Attempt to perform a given command with name and id. Includes retry logic with delays.

Arguments

command_name

str

arbitrary command name for logging/display

command_id

int

Mavlink ID of the command to be attempted

confirmation

int

whether this is first transmission of command (0, default), or a confirmation (1-255)

max_attempts

int

maximum number of attempts to send the command (default=3)

retry_delay

float

delay in seconds between attempts (default=1.0)

Returns

bool

whether command was successfully completed

set_custom_msg_rates

MavlinkClient.set_custom_msg_rates(msg_id: int, interval_us: int)

Set custom message publishing rates for a given message ID.

Arguments

msg_id

int

mavlink message ID to set custom rate for

interval_us

int

interval in microseconds to publish message

request_msg

MavlinkClient.request_msg(msg_id: int)

Request a specific message from the drone.

Arguments

msg_id

int

mavlink message ID to request

Returns

bool

whether command was successful

reset_ned_origin

MavlinkClient.reset_ned_origin()

Saves current position as the arming waypoint and resets mavlink origin.

Returns

bool

whether mavlink origin reset was successful

check_props_spinning

MavlinkClient.check_props_spinning(pwm_min_threshold: float = 1100)

Check if props are spinning by requesting and monitoring servo outputs. NOTE: this is the commanded PWM value, not the actual RPM.

Arguments

pwm_min_threshold

float

minimum PWM threshold for prop spinning

Returns

bool

whether all props are spinning above threshold

arm

MavlinkClient.arm()

Arm the drone if enabled.

Returns

bool

whether drone was successfully armed

disarm

MavlinkClient.disarm()

Disarm the drone.

Returns

bool

whether drone was successfully disarmed

set_offboard_control

MavlinkClient.set_offboard_control()

Sets drone control mode to OFFBOARD. This enables us to send position, velocity, or acceleration setpoints to the drone. NOTE: Drone may reject offboard mode if certain conditions are not met (recent setpoint message received, armed, etc.)

Returns

bool

whether offboard control mode was successfully set

start

MavlinkClient.start()

Start the drone by resetting NED origin, arming the drone, and setting offboard control mode.

terminate

MavlinkClient.terminate()

Terminate the drone client and shut down all threads.

get_single_frame

MavlinkClient.get_single_frame(camera_name: str)

Get a single frame from the named video stream.

Arguments

camera_name

str

name of the camera stream to capture frame from

Returns

np.ndarray

frame as numpy array if successful, else None

Get Started

Open GRID

GRID Enterprise

Robot API

Simulation

AI Layer

Deployment

Data Generation Pipelines

FAQ

init

get_msg

update_setpoint

attempt_command

set_custom_msg_rates

request_msg

reset_ned_origin

check_props_spinning

arm

disarm

set_offboard_control

start

terminate

get_single_frame

Get Started

Open GRID

GRID Enterprise

Robot API

Simulation

AI Layer

Deployment

Data Generation Pipelines

FAQ

​__init__

​get_msg

​update_setpoint

​attempt_command

​set_custom_msg_rates

​request_msg

​reset_ned_origin

​check_props_spinning

​arm

​disarm

​set_offboard_control

​start

​terminate

​get_single_frame

init

get_msg

update_setpoint

attempt_command

set_custom_msg_rates

request_msg

reset_ned_origin

check_props_spinning

arm

disarm

set_offboard_control

start

terminate

get_single_frame