use std::collections::BTreeMap;
use std::sync::Arc;
use ados_protocol::camera_control::{CameraControlRequest, CameraControlResult};
use rmpv::Value;
use crate::client::{ClientError, EventCallback, PluginIpcClient};
use crate::vision::VisionClient;
mod media;
mod network;
pub use media::{SensorsClient, VideoClient};
pub use network::{CloudClient, MdnsClient, NodeClient};
#[derive(Clone)]
pub struct EventsClient {
ipc: Arc<PluginIpcClient>,
}
impl EventsClient {
pub async fn publish(&self, topic: &str, payload: Value) -> Result<i64, ClientError> {
self.ipc.event_publish(topic, payload).await
}
pub async fn subscribe(
&self,
topic_pattern: &str,
callback: EventCallback,
) -> Result<(), ClientError> {
self.ipc.event_subscribe(topic_pattern, callback).await
}
}
#[derive(Clone)]
pub struct MavlinkClient {
ipc: Arc<PluginIpcClient>,
}
impl MavlinkClient {
pub async fn send(
&self,
msg_bytes: &[u8],
component_id: Option<i64>,
) -> Result<Value, ClientError> {
self.ipc.mavlink_send(msg_bytes, component_id).await
}
pub async fn subscribe(
&self,
msg_name: &str,
callback: EventCallback,
) -> Result<(), ClientError> {
self.ipc.mavlink_subscribe(msg_name, callback).await
}
pub async fn register_component(&self, comp_id: i64, kind: &str) -> Result<Value, ClientError> {
self.ipc.mavlink_register_component(comp_id, kind).await
}
}
fn sticks_to_channels(roll: f32, pitch: f32, yaw: f32, throttle: f32) -> [u16; 4] {
use ados_protocol::msp::{bipolar_to_pwm, throttle_to_pwm};
[
bipolar_to_pwm(roll),
bipolar_to_pwm(pitch),
throttle_to_pwm(throttle),
bipolar_to_pwm(yaw),
]
}
#[derive(Clone)]
pub struct MspClient {
ipc: Arc<PluginIpcClient>,
}
impl MspClient {
pub async fn send(&self, msg_bytes: &[u8]) -> Result<Value, ClientError> {
self.ipc.msp_send(msg_bytes).await
}
pub async fn send_raw_rc(&self, channels: &[u16]) -> Result<Value, ClientError> {
let frame = ados_protocol::msp::set_raw_rc(channels, true)
.ok_or_else(|| ClientError::Rpc("too many RC channels for the frame".to_string()))?;
self.ipc.msp_send(&frame).await
}
pub async fn send_sticks(
&self,
roll: f32,
pitch: f32,
yaw: f32,
throttle: f32,
) -> Result<Value, ClientError> {
self.send_raw_rc(&sticks_to_channels(roll, pitch, yaw, throttle))
.await
}
pub async fn subscribe(&self, callback: EventCallback) -> Result<(), ClientError> {
self.ipc.msp_subscribe(callback).await
}
}
#[derive(Clone)]
pub struct TelemetryClient {
ipc: Arc<PluginIpcClient>,
}
impl TelemetryClient {
pub async fn subscribe(&self, callback: EventCallback) -> Result<(), ClientError> {
let on_deliver = move |args: Value| {
let Value::Map(entries) = args else {
return;
};
if let Some((_, state)) = entries
.into_iter()
.find(|(k, _)| k.as_str() == Some("payload"))
{
callback(state);
}
};
self.ipc.telemetry_subscribe(Arc::new(on_deliver)).await
}
pub async fn extend(&self, channel: &str, payload: Value) -> Result<Value, ClientError> {
self.ipc.telemetry_extend(channel, payload).await
}
}
#[derive(Clone)]
pub struct PeripheralClient {
ipc: Arc<PluginIpcClient>,
}
impl PeripheralClient {
pub async fn register_camera_driver(&self, driver_ref: &str) -> Result<Value, ClientError> {
self.ipc
.peripheral_register_driver("camera", driver_ref)
.await
}
pub async fn register_lidar_driver(&self, driver_ref: &str) -> Result<Value, ClientError> {
self.ipc
.peripheral_register_driver("lidar", driver_ref)
.await
}
pub async fn register_gimbal_driver(&self, driver_ref: &str) -> Result<Value, ClientError> {
self.ipc
.peripheral_register_driver("gimbal", driver_ref)
.await
}
pub async fn register_gps_driver(&self, driver_ref: &str) -> Result<Value, ClientError> {
self.ipc.peripheral_register_driver("gps", driver_ref).await
}
pub async fn register_esc_driver(&self, driver_ref: &str) -> Result<Value, ClientError> {
self.ipc.peripheral_register_driver("esc", driver_ref).await
}
pub async fn register_payload_actuator_driver(
&self,
driver_ref: &str,
) -> Result<Value, ClientError> {
self.ipc
.peripheral_register_driver("payload-actuator", driver_ref)
.await
}
pub async fn unregister(&self, handle_id: &str) -> Result<Value, ClientError> {
self.ipc.peripheral_unregister_driver(handle_id).await
}
pub async fn claim_camera(
&self,
device_path: &str,
exclusive: bool,
) -> Result<Value, ClientError> {
self.ipc.camera_claim(device_path, exclusive).await
}
}
#[derive(Clone)]
pub struct CameraClient {
ipc: Arc<PluginIpcClient>,
}
impl CameraClient {
pub async fn claim(&self, device_path: &str, exclusive: bool) -> Result<Value, ClientError> {
self.ipc.camera_claim(device_path, exclusive).await
}
pub async fn release(&self, device_path: &str) -> Result<Value, ClientError> {
self.ipc.camera_release(device_path).await
}
pub async fn get_frame(
&self,
device_path: &str,
format: &str,
timeout_ms: i64,
) -> Result<Value, ClientError> {
self.ipc
.camera_get_frame(device_path, format, timeout_ms)
.await
}
pub async fn set_controls(
&self,
req: &CameraControlRequest,
) -> Result<CameraControlResult, ClientError> {
self.ipc.camera_set_controls(req).await
}
}
#[derive(Clone)]
pub struct GpioClient {
ipc: Arc<PluginIpcClient>,
}
impl GpioClient {
pub async fn set(&self, chip: i64, pin: i64, high: bool) -> Result<Value, ClientError> {
self.ipc.gpio_output_set(chip, pin, high).await
}
pub async fn beep(
&self,
chip: i64,
pin: i64,
on_ms: i64,
cycles: i64,
off_ms: Option<i64>,
freq_hz: Option<i64>,
) -> Result<Value, ClientError> {
self.ipc
.gpio_buzzer_beep(chip, pin, on_ms, cycles, off_ms, freq_hz)
.await
}
}
#[derive(Clone)]
pub struct DisplayClient {
ipc: Arc<PluginIpcClient>,
}
impl DisplayClient {
pub async fn set_page(
&self,
title: &str,
rows: &[(String, String)],
zones: &[(i64, i64, i64, i64, String, String)],
) -> Result<Value, ClientError> {
self.ipc.display_page_set(title, rows, zones).await
}
}
#[derive(Clone)]
pub struct RadioClient {
ipc: Arc<PluginIpcClient>,
}
impl RadioClient {
pub async fn open_aux_stream(&self) -> Result<Value, ClientError> {
self.ipc.radio_aux_stream_open().await
}
pub async fn close_aux_stream(&self) -> Result<Value, ClientError> {
self.ipc.radio_aux_stream_close().await
}
pub async fn send_aux(&self, channel: u8, payload: &[u8]) -> Result<Value, ClientError> {
self.ipc.radio_aux_stream_send(channel, payload).await
}
pub async fn subscribe_aux<F: Fn(u8, Vec<u8>) + Send + Sync + 'static>(
&self,
cb: F,
) -> Result<(), ClientError> {
let on_deliver = move |args: Value| {
let Some(m) = args.as_map() else { return };
let ch = m
.iter()
.find(|(k, _)| k.as_str() == Some("channel"))
.and_then(|(_, v)| v.as_u64())
.unwrap_or(0) as u8;
let pl = match m
.iter()
.find(|(k, _)| k.as_str() == Some("payload"))
.map(|(_, v)| v)
{
Some(Value::Binary(b)) => b.clone(),
_ => Vec::new(),
};
cb(ch, pl);
};
self.ipc.register_aux_callback(Arc::new(on_deliver));
self.ipc.radio_aux_stream_subscribe().await.map(|_| ())
}
}
#[derive(Clone)]
pub struct ButtonClient {
ipc: Arc<PluginIpcClient>,
}
impl ButtonClient {
pub async fn subscribe(
&self,
callback: Arc<dyn Fn(ados_protocol::buttons::ButtonPress) + Send + Sync>,
) -> Result<(), ClientError> {
let on_deliver = move |args: Value| {
let Some(press) = args
.as_map()
.and_then(|m| m.iter().find(|(k, _)| k.as_str() == Some("press")))
.map(|(_, v)| v.clone())
else {
return;
};
let Ok(blob) = rmp_serde::to_vec_named(&press) else {
return;
};
let Ok(decoded) = rmp_serde::from_slice::<ados_protocol::buttons::ButtonPress>(&blob)
else {
return;
};
callback(decoded);
};
self.ipc.register_button_callback(Arc::new(on_deliver));
self.ipc.button_subscribe().await?;
Ok(())
}
}
#[derive(Clone)]
pub struct ToolsClient {
ipc: Arc<PluginIpcClient>,
}
impl ToolsClient {
pub fn serve(&self, handler: crate::client::ToolHandler) {
self.ipc.set_tool_handler(handler);
}
}
#[derive(Clone)]
pub struct FlightClient {
ipc: Arc<PluginIpcClient>,
}
impl FlightClient {
pub async fn guided_setpoint(&self, args: Value) -> Result<Value, ClientError> {
self.ipc.flight_guided_setpoint(args).await
}
pub async fn rate_setpoint(&self, args: Value) -> Result<Value, ClientError> {
self.ipc.flight_rate_setpoint(args).await
}
}
#[derive(Clone)]
pub struct ConfigClient {
ipc: Arc<PluginIpcClient>,
static_config: Arc<BTreeMap<String, Value>>,
}
impl ConfigClient {
pub fn static_get(&self, key: &str) -> Option<&Value> {
self.static_config.get(key)
}
pub async fn get(&self, key: &str, default: Value) -> Result<Value, ClientError> {
self.ipc.config_get(key, default).await
}
pub async fn set(&self, key: &str, value: Value, scope: &str) -> Result<Value, ClientError> {
self.ipc.config_set(key, value, scope).await
}
}
#[derive(Clone)]
pub struct ProcessClient {
ipc: Arc<PluginIpcClient>,
}
impl ProcessClient {
pub async fn spawn(
&self,
basename: &str,
args: Vec<String>,
env: Vec<(String, String)>,
) -> Result<Value, ClientError> {
self.ipc.process_spawn(basename, args, env).await
}
}
#[derive(Clone)]
pub struct LifecycleClient {
ipc: Arc<PluginIpcClient>,
}
impl LifecycleClient {
pub async fn on_pause(&self, handler: EventCallback) -> Result<(), ClientError> {
let topic = format!("plugin.{}.lifecycle.pause", self.ipc.plugin_id());
self.ipc.event_subscribe(&topic, handler).await
}
pub async fn on_resume(&self, handler: EventCallback) -> Result<(), ClientError> {
let topic = format!("plugin.{}.lifecycle.resume", self.ipc.plugin_id());
self.ipc.event_subscribe(&topic, handler).await
}
}
#[derive(Clone)]
pub struct PluginContext {
pub plugin_id: String,
pub plugin_version: String,
pub agent_id: String,
pub data_dir: Option<std::path::PathBuf>,
pub events: EventsClient,
pub mavlink: MavlinkClient,
pub msp: MspClient,
pub telemetry: TelemetryClient,
pub peripheral_manager: PeripheralClient,
pub peripherals: PeripheralClient,
pub camera: CameraClient,
pub vision: VisionClient,
pub config: ConfigClient,
pub buttons: ButtonClient,
pub flight: FlightClient,
pub gpio: GpioClient,
pub display: DisplayClient,
pub radio: RadioClient,
pub cloud: CloudClient,
pub node: NodeClient,
pub mdns: MdnsClient,
pub video: VideoClient,
pub sensors: SensorsClient,
pub process: ProcessClient,
pub lifecycle: LifecycleClient,
pub tools: ToolsClient,
ipc: Arc<PluginIpcClient>,
}
impl PluginContext {
pub fn new(
ipc: Arc<PluginIpcClient>,
plugin_version: impl Into<String>,
agent_id: impl Into<String>,
data_dir: Option<String>,
static_config: BTreeMap<String, Value>,
) -> Self {
let plugin_id = ipc.plugin_id().to_string();
let peripheral_manager = PeripheralClient { ipc: ipc.clone() };
Self {
plugin_id,
plugin_version: plugin_version.into(),
agent_id: agent_id.into(),
data_dir: data_dir.map(std::path::PathBuf::from),
events: EventsClient { ipc: ipc.clone() },
mavlink: MavlinkClient { ipc: ipc.clone() },
msp: MspClient { ipc: ipc.clone() },
telemetry: TelemetryClient { ipc: ipc.clone() },
peripherals: peripheral_manager.clone(),
peripheral_manager,
camera: CameraClient { ipc: ipc.clone() },
vision: VisionClient::new(ipc.clone()),
buttons: ButtonClient { ipc: ipc.clone() },
flight: FlightClient { ipc: ipc.clone() },
gpio: GpioClient { ipc: ipc.clone() },
display: DisplayClient { ipc: ipc.clone() },
radio: RadioClient { ipc: ipc.clone() },
cloud: CloudClient { ipc: ipc.clone() },
node: NodeClient { ipc: ipc.clone() },
mdns: MdnsClient { ipc: ipc.clone() },
video: VideoClient { ipc: ipc.clone() },
sensors: SensorsClient { ipc: ipc.clone() },
config: ConfigClient {
ipc: ipc.clone(),
static_config: Arc::new(static_config),
},
process: ProcessClient { ipc: ipc.clone() },
lifecycle: LifecycleClient { ipc: ipc.clone() },
tools: ToolsClient { ipc: ipc.clone() },
ipc,
}
}
pub async fn ping_supervisor(&self) -> Result<Value, ClientError> {
self.ipc.ping().await
}
pub fn client(&self) -> &Arc<PluginIpcClient> {
&self.ipc
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn static_config_is_read_synchronously() {
let ipc = Arc::new(PluginIpcClient::new(
"com.example.demo",
"tok",
"/tmp/x.sock",
));
let mut cfg = BTreeMap::new();
cfg.insert("palette".to_string(), Value::from("ironbow"));
let ctx = PluginContext::new(ipc, "1.0.0", "agent-1", None, cfg);
assert_eq!(
ctx.config.static_get("palette"),
Some(&Value::from("ironbow"))
);
assert!(ctx.config.static_get("missing").is_none());
assert_eq!(ctx.plugin_id, "com.example.demo");
assert_eq!(ctx.agent_id, "agent-1");
assert_eq!(ctx.plugin_version, "1.0.0");
}
#[test]
fn send_sticks_scales_and_orders_aetr() {
assert_eq!(
sticks_to_channels(0.0, 0.0, 0.0, 0.0),
[1500, 1500, 1000, 1500]
);
assert_eq!(
sticks_to_channels(1.0, -1.0, 1.0, 1.0),
[2000, 1000, 2000, 2000]
);
assert_eq!(
sticks_to_channels(5.0, -5.0, 0.0, 2.0),
[2000, 1000, 2000, 1500]
);
}
}