client = MavlinkClient(
    ground_control_station_ip="192.168.8.10",
    mavlink_port="14550",
    vehicle_ip="192.168.8.1",
    camera_streams={
        "hires_front_small": {"port": 8900, "path": "/live"},
    },
    arm_enable=False,
    dry_run=False,
    log_level="INFO",
    known_messages=["HEARTBEAT", "LOCAL_POSITION_NED", "COMMAND_ACK"],
)

​
__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)
client = MavlinkClient(
    ground_control_station_ip="192.168.8.10",
    mavlink_port="14550",
    vehicle_ip="192.168.8.1",
    camera_streams={
        "hires_front_small": {"port": 8900, "path": "/live"},
    },
    arm_enable=False,
    dry_run=False,
    log_level="INFO",
    known_messages=["HEARTBEAT", "LOCAL_POSITION_NED", "COMMAND_ACK"],
)

​
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
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
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
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
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
bool
whether all props are spinning above threshold

​
arm

​
MavlinkClient.arm()
Arm the drone if enabled.
​
Returns
​
bool
bool
whether drone was successfully armed

​
disarm

​
MavlinkClient.disarm()
Disarm the drone.
​
Returns
​
bool
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
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
np.ndarray
frame as numpy array if successful, else None