use std::time::Duration;
use pamoja_core::{Actuator, Device, Error as CoreError, Result as CoreResult, Telemetry};
use tokio::time::timeout;
use crate::dialect::{
mav_autopilot, mav_cmd, mav_mission_result, mav_mission_type, mav_state, mav_type, Attitude,
BatteryStatus, CommandAck, CommandLong, GlobalPositionInt, GpsRawInt, Heartbeat, Message,
MissionAck, MissionCount, MissionItemInt, MissionRequest, MissionRequestInt,
SetPositionTargetGlobalInt, SetPositionTargetLocalNed, Statustext, SysStatus, VfrHud,
};
use crate::frame::Frame;
use crate::link::{ByteLink, Connection};
use crate::protocol::command::{AckOutcome, CommandProtocol};
use crate::protocol::mission::{MissionReceiver, MissionSender, ReceiverAction};
use crate::protocol::MAX_RETRIES;
use crate::signing::{Signer, Verifier};
use crate::MavlinkError;
const RESPONSE_TIMEOUT: Duration = Duration::from_millis(1500);
const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(5);
pub const GCS_COMPONENT: u8 = 190;
#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum Report {
Heartbeat(Heartbeat),
SysStatus(SysStatus),
GpsRawInt(GpsRawInt),
Attitude(Attitude),
GlobalPositionInt(GlobalPositionInt),
VfrHud(VfrHud),
BatteryStatus(BatteryStatus),
Statustext(Statustext),
Other(Box<Frame>),
}
impl Report {
fn from_frame(frame: &Frame) -> Report {
let raw = || Report::Other(Box::new(*frame));
match frame.message_id() {
Heartbeat::ID => Heartbeat::decode(frame.payload())
.map(Report::Heartbeat)
.unwrap_or_else(|_| raw()),
SysStatus::ID => SysStatus::decode(frame.payload())
.map(Report::SysStatus)
.unwrap_or_else(|_| raw()),
GpsRawInt::ID => GpsRawInt::decode(frame.payload())
.map(Report::GpsRawInt)
.unwrap_or_else(|_| raw()),
Attitude::ID => Attitude::decode(frame.payload())
.map(Report::Attitude)
.unwrap_or_else(|_| raw()),
GlobalPositionInt::ID => GlobalPositionInt::decode(frame.payload())
.map(Report::GlobalPositionInt)
.unwrap_or_else(|_| raw()),
VfrHud::ID => VfrHud::decode(frame.payload())
.map(Report::VfrHud)
.unwrap_or_else(|_| raw()),
BatteryStatus::ID => BatteryStatus::decode(frame.payload())
.map(Report::BatteryStatus)
.unwrap_or_else(|_| raw()),
Statustext::ID => Statustext::decode(frame.payload())
.map(Report::Statustext)
.unwrap_or_else(|_| raw()),
_ => raw(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum Setpoint {
Local(SetPositionTargetLocalNed),
Global(SetPositionTargetGlobalInt),
}
pub struct Vehicle<L> {
connection: Connection<L>,
id: String,
target_system: u8,
target_component: u8,
autodetect_target: bool,
}
impl<L: ByteLink> Vehicle<L> {
pub fn new(link: L, system_id: u8, component_id: u8) -> Self {
Vehicle {
connection: Connection::new(link, system_id, component_id),
id: Self::format_id(1, 1),
target_system: 1,
target_component: 1,
autodetect_target: true,
}
}
pub fn with_target(mut self, system_id: u8, component_id: u8) -> Self {
self.target_system = system_id;
self.target_component = component_id;
self.autodetect_target = false;
self.id = Self::format_id(system_id, component_id);
self
}
pub fn with_signer(mut self, signer: Signer) -> Self {
self.connection = self.connection.with_signer(signer);
self
}
pub fn with_verifier(mut self, verifier: Verifier) -> Self {
self.connection = self.connection.with_verifier(verifier);
self
}
pub fn target_system(&self) -> u8 {
self.target_system
}
pub fn target_component(&self) -> u8 {
self.target_component
}
pub async fn send_heartbeat(&mut self) -> CoreResult<()> {
let heartbeat = Heartbeat {
custom_mode: 0,
type_: mav_type::GCS,
autopilot: mav_autopilot::INVALID,
base_mode: 0,
system_status: mav_state::ACTIVE,
mavlink_version: 3,
};
self.tx(&heartbeat).await
}
pub async fn recv(&mut self) -> CoreResult<Report> {
let frame = self.rx().await?;
Ok(Report::from_frame(&frame))
}
pub async fn send_command(&mut self, command: u16, params: [f32; 7]) -> CoreResult<u8> {
let mut protocol = CommandProtocol::new(command, MAX_RETRIES);
loop {
let request = CommandLong {
param1: params[0],
param2: params[1],
param3: params[2],
param4: params[3],
param5: params[4],
param6: params[5],
param7: params[6],
command,
target_system: self.target_system,
target_component: self.target_component,
confirmation: protocol.confirmation(),
};
self.tx(&request).await?;
let resend = loop {
match timeout(RESPONSE_TIMEOUT, self.rx()).await {
Err(_elapsed) => {
if protocol.on_timeout().is_none() {
return Err(CoreError::Transport(
"command was not acknowledged".into(),
));
}
break true;
}
Ok(frame) => {
let frame = frame?;
if frame.message_id() == CommandAck::ID {
let ack = CommandAck::decode(frame.payload()).map_err(map_mav)?;
match protocol.on_ack(&ack) {
AckOutcome::Final(result) => return Ok(result),
AckOutcome::InProgress(_) | AckOutcome::Unrelated => continue,
}
}
}
}
};
debug_assert!(resend);
}
}
pub async fn arm(&mut self, arm: bool) -> CoreResult<u8> {
let flag = if arm { 1.0 } else { 0.0 };
self.send_command(
mav_cmd::COMPONENT_ARM_DISARM,
[flag, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
)
.await
}
pub async fn set_mode(&mut self, base_mode: u8, custom_mode: u32) -> CoreResult<u8> {
self.send_command(
mav_cmd::DO_SET_MODE,
[
base_mode as f32,
custom_mode as f32,
0.0,
0.0,
0.0,
0.0,
0.0,
],
)
.await
}
pub async fn takeoff(&mut self, altitude: f32) -> CoreResult<u8> {
self.send_command(
mav_cmd::NAV_TAKEOFF,
[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, altitude],
)
.await
}
pub async fn request_message(&mut self, message_id: u32) -> CoreResult<u8> {
self.send_command(
mav_cmd::REQUEST_MESSAGE,
[message_id as f32, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
)
.await
}
pub async fn set_message_interval(
&mut self,
message_id: u32,
interval_us: i32,
) -> CoreResult<u8> {
self.send_command(
mav_cmd::SET_MESSAGE_INTERVAL,
[
message_id as f32,
interval_us as f32,
0.0,
0.0,
0.0,
0.0,
0.0,
],
)
.await
}
pub async fn upload_mission(&mut self, items: &[MissionItemInt]) -> CoreResult<()> {
let sender = MissionSender::new(
items,
self.target_system,
self.target_component,
mav_mission_type::MISSION,
);
self.tx(&sender.count()).await?;
let mut last_seq: Option<u16> = None;
let mut retries = MAX_RETRIES;
loop {
match timeout(RESPONSE_TIMEOUT, self.rx()).await {
Err(_elapsed) => {
if retries == 0 {
return Err(CoreError::Transport("mission upload timed out".into()));
}
retries -= 1;
match last_seq {
None => self.tx(&sender.count()).await?,
Some(seq) => {
if let Some(item) = sender.item(seq) {
self.tx(&item).await?;
}
}
}
}
Ok(frame) => {
let frame = frame?;
match frame.message_id() {
MissionRequestInt::ID => {
let request =
MissionRequestInt::decode(frame.payload()).map_err(map_mav)?;
self.answer_item(&sender, request.seq, &mut last_seq, &mut retries)
.await?;
}
MissionRequest::ID => {
let request =
MissionRequest::decode(frame.payload()).map_err(map_mav)?;
self.answer_item(&sender, request.seq, &mut last_seq, &mut retries)
.await?;
}
MissionAck::ID => {
let ack = MissionAck::decode(frame.payload()).map_err(map_mav)?;
if ack.type_ == mav_mission_result::ACCEPTED {
return Ok(());
}
return Err(CoreError::Transport(format!(
"vehicle rejected the mission: result {}",
ack.type_
)));
}
_ => {}
}
}
}
}
}
pub async fn download_mission(&mut self) -> CoreResult<Vec<MissionItemInt>> {
let mut receiver = MissionReceiver::new(
self.target_system,
self.target_component,
mav_mission_type::MISSION,
);
self.tx(&receiver.request_list()).await?;
let mut items: Vec<MissionItemInt> = Vec::new();
let mut last_request: Option<MissionRequestInt> = None;
let mut got_count = false;
let mut retries = MAX_RETRIES;
loop {
match timeout(RESPONSE_TIMEOUT, self.rx()).await {
Err(_elapsed) => {
if retries == 0 {
return Err(CoreError::Transport("mission download timed out".into()));
}
retries -= 1;
match &last_request {
Some(request) => self.tx(request).await?,
None => self.tx(&receiver.request_list()).await?,
}
}
Ok(frame) => {
let frame = frame?;
match frame.message_id() {
MissionCount::ID if !got_count => {
let count = MissionCount::decode(frame.payload())
.map_err(map_mav)?
.count;
got_count = true;
items.reserve(count as usize);
match receiver.on_count(count) {
ReceiverAction::Request(request) => {
self.tx(&request).await?;
last_request = Some(request);
retries = MAX_RETRIES;
}
ReceiverAction::Ack(ack) => {
self.tx(&ack).await?;
return Ok(items);
}
}
}
MissionItemInt::ID => {
let item = MissionItemInt::decode(frame.payload()).map_err(map_mav)?;
let (accepted, action) = receiver.on_item(&item);
if let Some(item) = accepted {
items.push(item);
}
match action {
ReceiverAction::Request(request) => {
self.tx(&request).await?;
last_request = Some(request);
retries = MAX_RETRIES;
}
ReceiverAction::Ack(ack) => {
self.tx(&ack).await?;
return Ok(items);
}
}
}
_ => {}
}
}
}
}
}
async fn answer_item(
&mut self,
sender: &MissionSender<'_>,
seq: u16,
last_seq: &mut Option<u16>,
retries: &mut u8,
) -> CoreResult<()> {
if let Some(item) = sender.item(seq) {
self.tx(&item).await?;
*last_seq = Some(seq);
*retries = MAX_RETRIES;
}
Ok(())
}
async fn wait_for_heartbeat(&mut self) -> CoreResult<()> {
loop {
let frame = timeout(HEARTBEAT_TIMEOUT, self.rx())
.await
.map_err(|_| CoreError::Transport("no heartbeat from the vehicle".into()))??;
if frame.message_id() == Heartbeat::ID {
if self.autodetect_target {
self.target_system = frame.system_id();
self.target_component = frame.component_id();
self.id = Self::format_id(self.target_system, self.target_component);
}
return Ok(());
}
}
}
async fn tx<M: Message>(&mut self, message: &M) -> CoreResult<()> {
self.connection.send(message).await.map_err(map_mav)
}
async fn rx(&mut self) -> CoreResult<Frame> {
self.connection.recv().await.map_err(map_mav)
}
fn format_id(system_id: u8, component_id: u8) -> String {
format!("mavlink:{system_id}.{component_id}")
}
}
impl<L: ByteLink> Device for Vehicle<L> {
fn id(&self) -> &str {
&self.id
}
async fn connect(&mut self) -> CoreResult<()> {
self.wait_for_heartbeat().await
}
async fn disconnect(&mut self) -> CoreResult<()> {
Ok(())
}
}
impl<L: ByteLink> Telemetry for Vehicle<L> {
type Frame = Report;
async fn next_frame(&mut self) -> CoreResult<Option<Report>> {
match self.connection.recv().await {
Ok(frame) => Ok(Some(Report::from_frame(&frame))),
Err(MavlinkError::Closed) => Ok(None),
Err(err) => Err(map_mav(err)),
}
}
}
impl<L: ByteLink> Actuator for Vehicle<L> {
type Command = Setpoint;
async fn apply(&mut self, command: Setpoint) -> CoreResult<()> {
match command {
Setpoint::Local(setpoint) => self.tx(&setpoint).await,
Setpoint::Global(setpoint) => self.tx(&setpoint).await,
}
}
}
fn map_mav(err: MavlinkError) -> CoreError {
match err {
MavlinkError::Closed => CoreError::Closed,
MavlinkError::BadPayload => CoreError::Codec("malformed MAVLink payload".into()),
MavlinkError::Unsigned | MavlinkError::BadSignature | MavlinkError::ReplayedTimestamp => {
CoreError::Auth(err.to_string())
}
other => CoreError::Transport(other.to_string()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dialect::{mav_frame, mav_result};
use crate::link::{MemoryLink, SitlAutopilot};
fn waypoint(seq: u16, lat: i32, lon: i32, alt: f32) -> MissionItemInt {
MissionItemInt {
param1: 0.0,
param2: 0.0,
param3: 0.0,
param4: 0.0,
x: lat,
y: lon,
z: alt,
seq,
command: mav_cmd::NAV_WAYPOINT,
target_system: 0,
target_component: 0,
frame: mav_frame::GLOBAL_RELATIVE_ALT_INT,
current: (seq == 0) as u8,
autocontinue: 1,
mission_type: mav_mission_type::MISSION,
}
}
fn spawn_autopilot(link: MemoryLink) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut autopilot = SitlAutopilot::new(link, 1, 1);
let _ = autopilot.emit_heartbeat().await;
loop {
if autopilot.serve_once().await.is_err() {
break;
}
}
})
}
#[tokio::test]
async fn connect_learns_the_target_from_the_heartbeat() {
let (gcs, vehicle) = MemoryLink::pair();
let handle = spawn_autopilot(vehicle);
let mut client = Vehicle::new(gcs, 255, GCS_COMPONENT);
client.connect().await.unwrap();
assert_eq!(client.target_system(), 1);
assert_eq!(client.id(), "mavlink:1.1");
handle.abort();
}
#[tokio::test]
async fn a_command_is_acknowledged() {
let (gcs, vehicle) = MemoryLink::pair();
let handle = spawn_autopilot(vehicle);
let mut client = Vehicle::new(gcs, 255, GCS_COMPONENT);
client.connect().await.unwrap();
let result = client.arm(true).await.unwrap();
assert_eq!(result, mav_result::ACCEPTED);
handle.abort();
}
#[tokio::test]
async fn a_mission_uploads_and_downloads_unchanged() {
let (gcs, vehicle) = MemoryLink::pair();
let handle = spawn_autopilot(vehicle);
let mut client = Vehicle::new(gcs, 255, GCS_COMPONENT);
client.connect().await.unwrap();
let plan = [
waypoint(0, 473_977_418, 85_455_939, 10.0),
waypoint(1, 473_977_500, 85_456_000, 20.0),
waypoint(2, 473_977_600, 85_456_100, 15.0),
];
client.upload_mission(&plan).await.unwrap();
let downloaded = client.download_mission().await.unwrap();
assert_eq!(downloaded.len(), 3);
for (sent, got) in plan.iter().zip(downloaded.iter()) {
assert_eq!(sent.x, got.x);
assert_eq!(sent.y, got.y);
assert_eq!(sent.z, got.z);
assert_eq!(sent.command, got.command);
}
handle.abort();
}
#[tokio::test]
async fn an_offboard_setpoint_is_accepted_by_the_actuator() {
let (gcs, vehicle) = MemoryLink::pair();
let handle = spawn_autopilot(vehicle);
let mut client = Vehicle::new(gcs, 255, GCS_COMPONENT);
client.connect().await.unwrap();
let setpoint = Setpoint::Local(SetPositionTargetLocalNed::velocity(
0,
mav_frame::LOCAL_NED,
client.target_system(),
client.target_component(),
0.5,
0.0,
-0.2,
));
client.apply(setpoint).await.unwrap();
handle.abort();
}
}