__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. ground_control_station_ip IP address of the ground control station (default: ‘192.168.8.10’)
port number for mavlink connection (default: ‘14550’)
IP address of the vehicle (default: ‘192.168.8.1’)
communication protocol to use, one of (‘udp’|‘tcp’) (default: ‘udp’)
camera_streams Optional[Dict[str, Dict[str, Any]]]
Dictionary of dictionaries containing the
whether to enable arming the drone (default: False)
whether to run in dry run mode which bypasses arming exceptions (default: False)
logging level to be displayed to the user, one of (‘DEBUG’|‘INFO’|‘WARNING’|‘ERROR’),
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. name of the message type to be retrieved.
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. 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. arbitrary command name for logging/display
Mavlink ID of the command to be attempted
whether this is first transmission of command (0, default), or a confirmation (1-255)
maximum number of attempts to send the command (default=3)
delay in seconds between attempts (default=1.0)
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. mavlink message ID to set custom rate for
interval in microseconds to publish message
request_msg MavlinkClient.request_msg(msg_id: int)
Request a specific message from the drone. mavlink message ID to request
whether command was successful
reset_ned_origin MavlinkClient.reset_ned_origin()
Saves current position as the arming waypoint and resets mavlink origin. 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. minimum PWM threshold for prop spinning
whether all props are spinning above threshold
arm Arm the drone if enabled. whether drone was successfully armed
disarm Disarm the drone. 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.) whether offboard control mode was successfully set
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. name of the camera stream to capture frame from
frame as numpy array if successful, else None
