refactoring with desperado

Author: xoolive

Cargo.lock

@@ -12,6 +12,8 @@ homepage.workspace = true
 
 [features]
 rtlsdr = ['rs1090/rtlsdr']
+pluto = ['rs1090/pluto']
+soapy = ['rs1090/soapy']
 sero = ['rs1090/sero']
 ssh = ['rs1090/ssh']
 
@@ -22,6 +24,7 @@ clap_complete = "4.5.62"
 clap_complete_nushell = "4.5.10"
 crossterm = { version = "0.29.0", features = ["event-stream"] }
 deku = "0.20.2"
+desperado = "0.1.2"
 dirs = "6.0.0"
 dotenv = "0.15.0"
 futures = "0.3.31"

crates/jet1090/Cargo.toml

@@ -114,11 +114,6 @@ struct Options {
     /// More details are available at: <https://mode-s.org/jet1090/sources>
     sources: Vec<source::Source>,
 
-    #[cfg(feature = "rtlsdr")]
-    /// List the detected devices, for now, only --discover rtlsdr is fully supported
-    #[arg(long, value_name = "ARGS")]
-    discover: Option<String>,
-
     /// logging file, use "-" for stdout (only in non-interactive mode)
     #[arg(short, long, value_name = "FILE")]
     log_file: Option<String>,
@@ -262,12 +257,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
-    #[cfg(feature = "rtlsdr")]
-    if let Some(args) = cli_options.discover {
-        rtlsdr::enumerate(&args.to_string());
-        return Ok(());
-    }
-
     if options.sources.is_empty() {
         eprintln!(
             "No source of data specified, use --help for more information"

crates/jet1090/src/main.rs

@@ -34,7 +34,9 @@ pub async fn sensors(value: &Source) -> Vec<Sensor> {
         Address::Tcp(_)
         | Address::Udp(_)
         | Address::Websocket(_)
-        | Address::Rtlsdr(_) => {
+        | Address::Rtlsdr(_)
+        | Address::Pluto(_)
+        | Address::Soapy(_) => {
             vec![Sensor {
                 serial: value.serial(),
                 name: value.name.clone(),

crates/jet1090/src/sensor.rs

@@ -4,18 +4,24 @@ use std::str::FromStr;
 
 use rs1090::prelude::*;
 
-#[cfg(feature = "rtlsdr")]
-use rs1090::source::rtlsdr;
+use rs1090::source::iqread;
 #[cfg(feature = "sero")]
 use rs1090::source::sero;
 #[cfg(feature = "ssh")]
 use rs1090::source::ssh::{TunnelledTcp, TunnelledWebsocket};
 
+use desperado::IqAsyncSource;
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc::Sender;
 use tracing::error;
 use url::Url;
 
+const MODES_FREQ: f64 = 1.09e9;
+const RATE_2_4M: f64 = 2.4e6;
+
+#[cfg(feature = "rtlsdr")]
+const RTLSDR_GAIN: f64 = 49.6;
+
 /**
 * A structure to describe the endpoint to access data.
 *
@@ -73,6 +79,20 @@ pub enum WebsocketPath {
     Long(WebsocketStruct),
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct PlutoConfig {
+    pub uri: String,
+    pub sample_rate: i64,
+    pub gain: f64,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct SoapyConfig {
+    pub args: Option<String>,
+    pub sample_rate: u32,
+    pub gain: Option<f64>,
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Address {
@@ -84,6 +104,10 @@ pub enum Address {
     Websocket(WebsocketPath),
     /// A RTL-SDR dongle (require feature `rtlsdr`): the parameter can be empty, or use other specifiers, e.g. `rtlsdr://serial=00000001`
     Rtlsdr(Option<String>),
+    /// A Pluto-ADAF SDR (require feature `pluto`): the parameter can be empty, or use other specifiers, e.g. `pluto://ip=192.168.2.1`
+    Pluto(PlutoConfig),
+    /// A SoapySDR device (require feature `soapy`): the parameter can be empty, or use other specifiers, e.g. `soapy://driver=rtlsdr`
+    Soapy(SoapyConfig),
     /// A token-based access to Sero Systems (require feature `sero`).
     Sero(SeroParams),
 }
@@ -148,6 +172,16 @@ impl FromStr for Source {
                 url.port_or_known_default().unwrap()
             )),
             "rtlsdr" => Address::Rtlsdr(url.host_str().map(|s| s.to_string())),
+            "pluto" => Address::Pluto(PlutoConfig {
+                uri: format!("ip:{}", url.host_str().unwrap()),
+                sample_rate: RATE_2_4M as i64,
+                gain: 50.,
+            }),
+            "soapy" => Address::Soapy(SoapyConfig {
+                args: url.host_str().map(|s| s.to_string()),
+                sample_rate: RATE_2_4M as u32,
+                gain: Some(49.6),
+            }),
             "ws" => Address::Websocket(WebsocketPath::Short(format!(
                 "ws://{}:{}/{}",
                 url.host_str().unwrap_or("0.0.0.0"),
@@ -200,6 +234,14 @@ impl Source {
                 let name = reference.clone().unwrap_or("rtlsdr".to_string());
                 build_serial(&name)
             }
+            Address::Pluto(config) => {
+                let name = config.uri.clone();
+                build_serial(&name)
+            }
+            Address::Soapy(config) => {
+                let name = config.args.clone().unwrap_or("soapy".to_string());
+                build_serial(&name)
+            }
             Address::Sero(_) => 0,
         }
     }
@@ -226,14 +268,88 @@ impl Source {
                 #[cfg(feature = "rtlsdr")]
                 {
                     let args = args.clone();
+                    // TODO that's temporary fix for now, discussion in PR
+                    let device_index = if let Some(ref args_str) = args {
+                        // Parse args string for "device_index=xx" or "index=xx"
+                        let mut index = 0;
+                        for part in args_str.split(',') {
+                            let part = part.trim();
+                            if let Some(val) =
+                                part.strip_prefix("device_index=")
+                            {
+                                if let Ok(idx) = val.parse::<usize>() {
+                                    index = idx;
+                                    break;
+                                }
+                            } else if let Some(val) =
+                                part.strip_prefix("index=")
+                            {
+                                if let Ok(idx) = val.parse::<usize>() {
+                                    index = idx;
+                                    break;
+                                }
+                            }
+                        }
+                        index
+                    } else {
+                        0
+                    };
+                    tokio::spawn(async move {
+                        let source = IqAsyncSource::from_rtlsdr(
+                            device_index,
+                            MODES_FREQ as u32,
+                            RATE_2_4M as u32,
+                            Some(10 * RTLSDR_GAIN as i32),
+                        )
+                        .await
+                        .expect("Failed to create RTL-SDR source");
+                        iqread::receiver(tx, source, serial, 2.4e6, name).await
+                    });
+                }
+            }
+            Address::Pluto(config) => {
+                #[cfg(not(feature = "pluto"))]
+                {
+                    error!("Compile jet1090 with the pluto feature, {:?} argument ignored", uri);
+                    std::process::exit(127);
+                }
+                #[cfg(feature = "pluto")]
+                {
+                    let config = config.clone();
+                    tokio::spawn(async move {
+                        let source = IqAsyncSource::from_pluto(
+                            &config.uri,
+                            MODES_FREQ as i64,
+                            config.sample_rate,
+                            config.gain,
+                        )
+                        .await
+                        .expect("Failed to create PlutoSDR source");
+                        iqread::receiver(tx, source, serial, 2.4e6, name).await
+                    });
+                }
+            }
+            Address::Soapy(config) => {
+                #[cfg(not(feature = "soapy"))]
+                {
+                    error!("Compile jet1090 with the soapy feature, {:?} argument ignored", args);
+                    std::process::exit(127);
+                }
+                #[cfg(feature = "soapy")]
+                {
+                    let args = config.clone();
                     tokio::spawn(async move {
-                        rtlsdr::receiver::<&str>(
-                            tx,
-                            args.as_deref(),
-                            serial,
-                            name,
+                        let source = IqAsyncSource::from_soapy(
+                            &args.args.unwrap_or("".to_string()),
+                            0,
+                            MODES_FREQ as u32,
+                            args.sample_rate,
+                            args.gain,
+                            "TUNER",
                         )
                         .await
+                        .expect("Failed to create SoapySDR source");
+                        iqread::receiver(tx, source, serial, 2.4e6, name).await
                     });
                 }
             }

crates/jet1090/src/source.rs

@@ -11,7 +11,9 @@ edition.workspace = true
 
 [features]
 aircraft = ['dirs', 'reqwest', 'rusqlite', 'zip']
-rtlsdr = ['soapysdr']
+rtlsdr = ['desperado/rtlsdr']
+pluto = ['desperado/pluto']
+soapy = ['desperado/soapy']
 sero = ['prost', 'tonic', 'tonic-prost', 'dirs', 'reqwest']
 ssh = ['async-recursion', 'bytes', 'dirs', 'http', 'hyper-util', 'makiko', 'ssh2-config', 'tower']
 
@@ -20,6 +22,7 @@ ansi_term = "0.12.1"
 async-recursion = { version = "1.1.1", optional = true }
 async-stream = "0.3.6"
 bytes = { version = "1.11.0", optional = true }
+desperado = "0.1.2"
 deku = { version = "0.20.2", features = ["logging"] }
 dirs = { version = "6.0.0", optional = true }
 futures = "0.3.31"
@@ -39,7 +42,6 @@ reqwest = { version = "0.12.26", optional = true, default-features = false, feat
 rusqlite = { version = "0.37.0", features = ["bundled"], optional = true }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.143"
-soapysdr = { version = "0.4.4", optional = true }
 ssh2-config = { version = "0.6.0", optional = true }
 tonic = { version = "0.14.2", features = ["_tls-any"], optional = true }
 tonic-prost = { version = "0.14.2", optional = true }

crates/rs1090/Cargo.toml

@@ -31,8 +31,7 @@ pub mod prelude {
     #[cfg(not(target_arch = "wasm32"))]
     pub use crate::source::beast;
 
-    #[cfg(feature = "rtlsdr")]
-    pub use crate::source::rtlsdr;
+    pub use crate::source::iqread;
 
     #[cfg(feature = "sero")]
     pub use crate::source::sero;