Working on adsb

Makefile

@@ -38,10 +38,7 @@ format-adsb: ## Format code
 	@cd adsb && cargo fmt
 
 build-adsb: ## Build the ADS-B project
-	@cd adsb && cargo build
+	@cd adsb && cargo build --release
-
-run-adsb: ## Run the ADS-B Receiver
-	@cd adsb && cargo run -- -c -D
 
 #################
 #  UI Commands  #

adsb/src/constants.rs

@@ -0,0 +1,45 @@
+use std::fmt::Display;
+use std::time::Duration;
+
+#[derive(Debug)]
+pub struct DeviceInfo {
+  /// Vendor ID
+  pub vid: u16,
+  /// Product ID
+  pub pid: u16,
+}
+
+impl Display for DeviceInfo {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    write!(f, "VID: 0x{:04X} PID: 0x{:04X}", self.vid, self.pid)
+  }
+}
+
+// Devices
+pub const DEVICE_RTL2832U: DeviceInfo = DeviceInfo {
+  vid: 0x0BDA,
+  pid: 0x2832,
+};
+
+// Timeout
+pub const TIMEOUT: Duration = Duration::from_secs(1);
+
+// pub const DEFAULT_BUFFER_LENGTH: usize = 4096;
+pub const DEFAULT_BUFFER_LENGTH: usize = 64;
+
+// Request Types
+pub const REQ_CTRL_OUT: u8 =
+  rusb::constants::LIBUSB_ENDPOINT_OUT | rusb::constants::LIBUSB_REQUEST_TYPE_VENDOR;
+
+// Blocks
+pub const BLOCK_USB: u16 = 1;
+
+// USB
+pub const USB_EPA_CTL: u16 = 0x2148;
+pub const USB_SYSCTL: u16 = 0x2000;
+
+/// ADS-B downlink frequency (1090 MHz)
+pub const ADSB_FREQUENCY_HZ: u32 = 1_090_000_000;
+
+/// RTL-SDR sample rate in samples/second.
+pub const SAMPLE_RATE_HZ: u32 = 2_048_000;

adsb/src/device.rs

@@ -1,23 +1,27 @@
+use std::borrow::Cow;
+use std::fmt::Display;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use rusb::{
   Context, Device, DeviceDescriptor, DeviceHandle, DeviceList, Direction, TransferType, UsbContext,
 };
 use crate::error::{Error, Result};
-use std::time::Duration;
+use crate::constants::{
-
+  ADSB_FREQUENCY_HZ, BLOCK_USB, DEFAULT_BUFFER_LENGTH, REQ_CTRL_OUT, SAMPLE_RATE_HZ, TIMEOUT,
-const TIMEOUT: Duration = Duration::from_secs(1);
+  USB_EPA_CTL, USB_SYSCTL,
+};
 
 /// rusb/libusb implementation of `RtlSdrDevice`
 pub struct RtlSdrDevice {
   /// Device handle
   handle: DeviceHandle<Context>,
-  device_desc: DeviceDescriptor,
+  endpoint: Endpoint,
-  device: Device<Context>,
+  frequency: u32,
+  rate: u32,
 }
 
 impl RtlSdrDevice {
-  /// Display dongle information
+  /// Display RTL SDR information
   pub fn info(vid: u16, pid: u16) {
     let device_list = match DeviceList::new() {
       Ok(d) => d,
@@ -29,7 +33,7 @@ impl RtlSdrDevice {
     for device in device_list.iter() {
       match device.device_descriptor() {
         Ok(device_desc) => {
-          if vid != device_desc.vendor_id() && pid != device_desc.product_id() {
+          if vid != device_desc.vendor_id() || pid != device_desc.product_id() {
             continue;
           }
           println!(
@@ -43,7 +47,7 @@ impl RtlSdrDevice {
           match device.open() {
             Ok(handle) => {
               println!("{}", device_info(&handle, &device_desc, "  ", true));
-            },
+            }
             Err(err) => {
               eprintln!("  Unable to open device: {:?}", err);
               continue;
@@ -58,7 +62,7 @@ impl RtlSdrDevice {
     }
   }
 
-  /// Open the USB device and return a wrapper
+  /// Open the RTL SDR device and return a wrapper
   pub fn open(vid: u16, pid: u16) -> Result<Self> {
     // Create a new libusb context
     let ctx = Context::new().map_err(|_| Error::new("Unable to create libusb context"))?;
@@ -71,33 +75,226 @@ impl RtlSdrDevice {
 
       if device_desc.vendor_id() == vid && device_desc.product_id() == pid {
         let handle = device.open()?;
-        return Ok(Self {
+
-          handle,
+        log::debug!("{}", device_info(&handle, &device_desc, "", false));
-          device_desc,
+
-          device,
+        // Find the endpoint
-        });
+        let endpoint = match Endpoint::find(&device_desc, &device, TransferType::Bulk) {
+          Some(e) => e,
+          None => return Err(Error::new("Unable to find endpoint on device")),
+        };
+        log::debug!("Found readable endpoint: {:?}", endpoint.to_string());
+
+        let mut sdr = Self::new(handle, endpoint);
+
+        sdr.initialize()?;
+
+        return Ok(sdr);
       }
     }
     Err(Error::new("No valid device found"))
   }
 
-  pub fn read(&mut self, transfer_type: TransferType) -> Result<()> {
+  /// Close the RTL SDR device
+  pub fn close(&self) -> Result<()> {
+    log::debug!("Closing device...");
+    self.attach_kernel_driver(self.endpoint.interface);
+    self.handle.release_interface(self.endpoint.interface)?;
+    Ok(())
+  }
+
+  /// Process the USB data
+  pub fn process(&mut self, running: Arc<AtomicBool>) -> Result<()> {
     log::debug!(
-      "Reading active configuration: {} ({:?})",
+      "Reading from active configuration: {}",
-      self.handle.active_configuration()?,
+      self.handle.active_configuration()?
-      transfer_type
     );
 
-    log::debug!("{}", device_info(&self.handle, &self.device_desc, "", false));
-
     // Read endpoint
-    match Endpoint::find_readable(&self.device, &self.device_desc, transfer_type) {
+    let mut buffer = [0u8; DEFAULT_BUFFER_LENGTH];
-      Some(endpoint) => endpoint.read(&mut self.handle)?,
+    while running.load(Ordering::SeqCst) {
-      None => log::warn!("No readable {:?} endpoint", transfer_type),
+      let s = self.read(&mut buffer)?;
+      log::debug!("Read: {}", s);
     }
 
+    self.close()
+  }
+
+  fn new(handle: DeviceHandle<Context>, endpoint: Endpoint) -> Self {
+    Self {
+      handle,
+      endpoint,
+      frequency: 0,
+      rate: 0,
+    }
+  }
+
+  fn read(&self, buffer: &mut [u8; DEFAULT_BUFFER_LENGTH]) -> Result<String> {
+    let length = match self.endpoint.transfer_type {
+      TransferType::Interrupt => self
+        .handle
+        .read_interrupt(self.endpoint.address, buffer, TIMEOUT)
+        .map_err(|err| Error::new(format!("Unable to read interrupt from endpoint: {:?}", err)))?,
+      TransferType::Bulk => self
+        .handle
+        .read_bulk(self.endpoint.address, buffer, TIMEOUT)
+        .map_err(|err| Error::new(format!("Unable to read bulk from endpoint: {:?}", err)))?,
+      _ => 0,
+    };
+    log::trace!("Received {} bytes", length);
+    let s = match String::from_utf8_lossy(&buffer[..length]) {
+      Cow::Borrowed(s) => s.to_string(),
+      Cow::Owned(s) => s,
+    };
+    Ok(s.to_string())
+  }
+
+  fn initialize(&mut self) -> Result<()> {
+    // Configure the device for the endpoint
+    self.set_active_configuration(self.endpoint.config)?;
+    self.claim_interface(self.endpoint.interface)?;
+    self.set_alternate_setting(self.endpoint.interface, self.endpoint.setting)?;
+    self.detach_kernel_driver(self.endpoint.interface);
+
+    self.test_write()?;
+
+    // Reset the internal USB buffer
+    self.reset_buffer()?;
+
+    // Set the center-frequency in Hz
+    self.set_center_frequency(ADSB_FREQUENCY_HZ)?;
+
+    // Set the sample rate
+    self.set_sample_rate(SAMPLE_RATE_HZ)?;
+
     Ok(())
   }
+
+  fn set_active_configuration(&self, configuration: u8) -> Result<()> {
+    self
+      .handle
+      .set_active_configuration(configuration)
+      .map_err(|err| Error::new(format!("Failed to set active configuration: {:?}", err)))
+  }
+
+  fn claim_interface(&self, interface: u8) -> Result<()> {
+    self
+      .handle
+      .claim_interface(interface)
+      .map_err(|err| Error::new(format!("Failed to claim interface: {:?}", err)))
+  }
+
+  fn set_alternate_setting(&self, interface: u8, setting: u8) -> Result<()> {
+    self
+      .handle
+      .set_alternate_setting(interface, setting)
+      .map_err(|err| Error::new(format!("Failed to set alternate setting: {:?}", err)))
+  }
+
+  fn test_write(&self) -> Result<()> {
+    log::trace!("Testing write to device...");
+    let length = ctrl_write_register(&self.handle, BLOCK_USB, USB_SYSCTL, 0x89, 1)?;
+    if length == 0 {
+      log::info!("Resetting device");
+      self
+        .handle
+        .reset()
+        .map_err(|err| Error::new(format!("Failed to reset device: {:?}", err)))?;
+    } else {
+      log::trace!("Test write was successful");
+    }
+    Ok(())
+  }
+
+  fn reset_buffer(&self) -> Result<()> {
+    log::trace!("Resetting buffer...");
+    ctrl_write_register(&self.handle, BLOCK_USB, USB_EPA_CTL, 0x1002, 2)
+      .map_err(|err| Error::new(format!("Failed to reset the internal buffer: {:?}", err)))?;
+    ctrl_write_register(&self.handle, BLOCK_USB, USB_EPA_CTL, 0x0000, 2)
+      .map_err(|err| Error::new(format!("Failed to reset the internal buffer: {:?}", err)))?;
+    Ok(())
+  }
+
+  fn set_center_frequency(&mut self, frequency: u32) -> Result<()> {
+    log::trace!("Setting center_frequency to {}Hz", frequency);
+    self.frequency = frequency;
+    Ok(())
+  }
+
+  fn set_sample_rate(&mut self, rate: u32) -> Result<()> {
+    log::trace!("Setting sample_rate to {}Hz", rate);
+    self.rate = rate;
+    Ok(())
+  }
+
+  fn detach_kernel_driver(&self, interface: u8) {
+    // Detach the kernel driver if applicable
+    if let Ok(true) = self.handle.kernel_driver_active(interface) {
+      log::trace!("Detaching active kernel driver");
+      self.handle.detach_kernel_driver(interface).ok();
+    }
+  }
+
+  fn attach_kernel_driver(&self, interface: u8) {
+    // Attach the kernel driver if applicable
+    if let Ok(true) = self.handle.kernel_driver_active(interface) {
+      log::trace!("Attaching active kernel driver");
+      self.handle.attach_kernel_driver(interface).ok();
+    }
+  }
+}
+
+#[derive(Debug)]
+pub struct Endpoint {
+  config: u8,
+  interface: u8,
+  setting: u8,
+  address: u8,
+  transfer_type: TransferType,
+}
+
+impl Display for Endpoint {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    write!(
+      f,
+      "Config: {}, Interface: {}, Setting: {}, Address: 0x{:04X}, Transfer Type: {:?}",
+      self.config, self.interface, self.setting, self.address, self.transfer_type
+    )
+  }
+}
+
+impl Endpoint {
+  fn find<T: UsbContext>(
+    device_desc: &DeviceDescriptor,
+    device: &Device<T>,
+    transfer_type: TransferType,
+  ) -> Option<Self> {
+    for n in 0..device_desc.num_configurations() {
+      let config_desc = match device.config_descriptor(n) {
+        Ok(c) => c,
+        Err(_) => continue,
+      };
+
+      for interface in config_desc.interfaces() {
+        for interface_desc in interface.descriptors() {
+          for endpoint_desc in interface_desc.endpoint_descriptors() {
+            if endpoint_desc.direction() == Direction::In
+              && endpoint_desc.transfer_type() == transfer_type
+            {
+              return Some(Endpoint {
+                config: config_desc.number(),
+                interface: interface_desc.interface_number(),
+                setting: interface_desc.setting_number(),
+                address: endpoint_desc.address(),
+                transfer_type,
+              });
+            }
+          }
+        }
+      }
+    }
+    None
+  }
 }
 
 fn device_info<T: UsbContext>(
@@ -110,20 +307,23 @@ fn device_info<T: UsbContext>(
     Ok(l) => l,
     Err(err) => {
       return format!("{} Unable to get languages: {:?}", offset, err);
-    },
+    }
   };
   let descriptor_type = device_desc.descriptor_type();
   let mut output = String::new();
   if full {
-    output = format!("{}Device Descriptor ({})\n", offset, descriptor_type).to_string();
+    output = format!("{}Device Descriptor ({})\n", offset, descriptor_type);
   }
   if !languages.is_empty() {
     for language in languages {
-      let manufacturer = handle.read_manufacturer_string(language, device_desc, TIMEOUT)
+      let manufacturer = handle
+        .read_manufacturer_string(language, device_desc, TIMEOUT)
         .unwrap_or_else(|err| err.to_string());
-      let product = handle.read_product_string(language, device_desc, TIMEOUT)
+      let product = handle
+        .read_product_string(language, device_desc, TIMEOUT)
         .unwrap_or_else(|err| err.to_string());
-      let serial_number = handle.read_serial_number_string(language, device_desc, TIMEOUT)
+      let serial_number = handle
+        .read_serial_number_string(language, device_desc, TIMEOUT)
         .unwrap_or_else(|err| err.to_string());
       output.push_str(&format!(
         "{}{}Manufacturer: {}, Product: {}, Serial Number: {}",
@@ -147,7 +347,7 @@ fn device_info<T: UsbContext>(
         let protocol = device_desc.protocol_code();
         let max_packet_size = device_desc.max_packet_size();
         output.push_str(&format!(
-          "{}{}Class: {:#04x}, Subclass: {:#04x}, Protocol: {:#04x}, Max Packet Size: {}\n",
+          "{}{}Class: {:#04x}, Subclass: {:#04x}, Protocol: {:#04x}, Max Packet Size: {}",
           offset, offset, class, sub_class, protocol, max_packet_size
         ))
       }
@@ -156,119 +356,46 @@ fn device_info<T: UsbContext>(
   output
 }
 
-#[derive(Debug)]
+fn ctrl_write_register<T: UsbContext>(
-struct Endpoint {
+  handle: &DeviceHandle<T>,
-  config: u8,
+  block: u16,
-  interface: u8,
+  address: u16,
-  setting: u8,
+  value: u16,
-  address: u8,
+  length: usize,
-  transfer_type: TransferType,
+) -> rusb::Result<usize> {
+  assert!(length == 1 || length == 2);
+
+  let data: [u8; 2] = value.to_be_bytes();
+  let buffer = if length == 1 { &data[1..2] } else { &data };
+  let index = (block << 8) | 0x10;
+  log::trace!(
+    "Received block {}, address 0x{:04X}, value 0x{:04X}, length {} \
+    - writing control register: {} 0x{:04X} 0x{:04X} {:?}",
+    block,
+    address,
+    value,
+    length,
+    REQ_CTRL_OUT,
+    address,
+    index,
+    buffer
+  );
+
+  handle.write_control(REQ_CTRL_OUT, 0x00, address, index, buffer, TIMEOUT)
 }
 
-impl Endpoint {
+fn demod_ctrl_write_register<T: UsbContext>(
-  pub fn find_readable<T: UsbContext>(
+  handle: &DeviceHandle<T>,
-    device: &Device<T>,
+  page: u16,
-    device_desc: &DeviceDescriptor,
+  address: u16,
-    transfer_type: TransferType,
+  value: u16,
-  ) -> Option<Self> {
+  length: usize,
-    for n in 0..device_desc.num_configurations() {
+) -> rusb::Result<usize> {
-      let config_desc = match device.config_descriptor(n) {
+  assert!(length == 1 || length == 2);
-        Ok(c) => c,
-        Err(_) => continue,
-      };
 
-      for interface in config_desc.interfaces() {
+  let data: [u8; 2] = value.to_be_bytes();
-        for interface_desc in interface.descriptors() {
+  let buffer = if length == 1 { &data[1..2] } else { &data };
-          for endpoint_desc in interface_desc.endpoint_descriptors() {
+  let index = 0x10 | page;
-            if endpoint_desc.direction() == Direction::In
+  let address = (address << 8) | 0x20;
-              && endpoint_desc.transfer_type() == transfer_type
+  handle.write_control(REQ_CTRL_OUT, 0x00, address, index, buffer, TIMEOUT)
-            {
-              return Some(Self {
-                config: config_desc.number(),
-                interface: interface_desc.interface_number(),
-                setting: interface_desc.setting_number(),
-                address: endpoint_desc.address(),
-                transfer_type,
-              });
-            }
-          }
-        }
-      }
-    }
-    None
-  }
-
-  fn read<T: UsbContext>(&self, handle: &mut DeviceHandle<T>) -> Result<()> {
-    log::debug!("Reading from endpoint: {:?}", self);
-    let running = Arc::new(AtomicBool::new(true));
-    {
-      let running = running.clone();
-      ctrlc::set_handler(move || {
-        running.store(false, Ordering::SeqCst);
-      })?;
-    }
-
-    // Detach the kernel driver if applicable
-    let has_kernel_driver = match handle.kernel_driver_active(self.interface) {
-      Ok(true) => {
-        log::debug!("Detaching active kernel driver");
-        handle.detach_kernel_driver(self.interface).ok();
-        true
-      }
-      _ => false,
-    };
-
-    self.configure_endpoint(handle)?;
-
-    let mut buffer = [0u8; 4096];
-    while running.load(Ordering::SeqCst) {
-      let length = match self.transfer_type {
-        TransferType::Interrupt => handle
-          .read_interrupt(self.address, &mut buffer, TIMEOUT)
-          .map_err(|err| {
-            Error::new(format!("Unable to read interrupt from endpoint: {:?}", err))
-          })?,
-        TransferType::Bulk => handle
-          .read_bulk(self.address, &mut buffer, TIMEOUT)
-          .map_err(|err| Error::new(format!("Unable to read bulk from endpoint: {:?}", err)))?,
-        _ => 0,
-      };
-      log::debug!("Received: {:?}", &buffer[..length]);
-    }
-
-    // Attach the kernel driver if applicable
-    if has_kernel_driver {
-      log::debug!("Attaching active kernel driver");
-      handle.attach_kernel_driver(self.interface).ok();
-    }
-
-    log::debug!("Exiting USB read");
-    Ok(())
-  }
-
-  fn configure_endpoint<T: UsbContext>(&self, handle: &mut DeviceHandle<T>) -> Result<()> {
-    log::debug!("Configuring endpoint: {:?}", self);
-
-    // Switch to ADS-B mode
-    // let request_type = request_type(Direction::Out, RequestType::Vendor, Recipient::Interface);
-    // handle.write_control(
-    //  request_type,
-    //   0x42,
-    //   0x0002,
-    //   0,
-    //   &[],
-    //   TIMEOUT,
-    // )?;
-
-    handle
-      .set_active_configuration(self.config)
-      .map_err(|err| Error::new(format!("Failed to set active configuration: {:?}", err)))?;
-    handle
-      .claim_interface(self.interface)
-      .map_err(|err| Error::new(format!("Failed to claim interface: {:?}", err)))?;
-    handle
-      .set_alternate_setting(self.interface, self.setting)
-      .map_err(|err| Error::new(format!("Failed to set alternate setting: {:?}", err)))?;
-    Ok(())
-  }
 }

adsb/src/error.rs

@@ -36,3 +36,9 @@ impl From<ctrlc::Error> for Error {
     Error::Other(err.to_string())
   }
 }
+
+impl From<std::str::Utf8Error> for Error {
+  fn from(err: std::str::Utf8Error) -> Self {
+    Error::Other(err.to_string())
+  }
+}

adsb/src/main.rs

@@ -1,12 +1,14 @@
+mod constants;
 mod device;
 mod error;
 mod frame;
 mod hex;
 
-use error::Result;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use crate::device::RtlSdrDevice;
 use clap::Parser;
-use rusb::TransferType;
+use crate::constants::DEVICE_RTL2832U;
 use crate::frame::ADSBFrame;
 use crate::hex::hex_to_bytes;
 
@@ -26,47 +28,71 @@ struct ReceiverArgs {
   info: bool,
 
   /// Enable debug logging
-  #[arg(short = 'D', long, action)]
+  #[arg(short = 'D', long, action = clap::ArgAction::Count)]
-  debug: bool,
+  debug: u8,
 }
 
-fn main() -> Result<()> {
+fn main() {
   let args = ReceiverArgs::parse();
 
-  let default_filter = if args.debug {
+  let default_filter = match args.debug {
-    "warn,adsb=debug"
+    0 => "warn,adsb=info",   // no -D
-  } else {
+    1 => "warn,adsb=debug",  // -D
-    "warn,adsb=info"
+    _ => "trace,adsb=trace", // -DD or more
   };
 
   env_logger::init_from_env(env_logger::Env::default().filter_or("RUST_LOG", default_filter));
 
-  let vid = 0x0BDA;
+  let device_info = DEVICE_RTL2832U;
-  let pid = 0x2832;
 
   // Handle connection
   if args.connect {
-    log::info!("Connecting to device");
+    log::info!("Connecting to {:?}", device_info);
-    let mut device = RtlSdrDevice::open(vid, pid)?;
+    let mut device = match RtlSdrDevice::open(device_info.vid, device_info.pid) {
-    device.read(TransferType::Bulk)
+      Ok(d) => d,
+      Err(err) => {
+        log::error!("Unable to open RTL SDR device: {:?}", err);
+        return;
+      }
+    };
+    log::debug!("Connected to {:?}", device_info.to_string());
+
+    let running = Arc::new(AtomicBool::new(true));
+    if let Err(err) = ctrlc::set_handler({
+      let running = running.clone();
+      move || running.store(false, Ordering::SeqCst)
+    }) {
+      log::error!("Error setting Ctrl-C handler: {}", err);
+      running.store(false, Ordering::SeqCst);
+    };
+
+    if let Err(err) = device.process(running) {
+      log::error!("Failed to read from device: {}", err);
+      if let Err(err) = device.close() {
+        log::error!("Failed to close device: {}", err);
+      };
+    };
   }
   // Display dongle info
   else if args.info {
-    RtlSdrDevice::info(vid, pid);
+    RtlSdrDevice::info(device_info.vid, device_info.pid);
-    Ok(())
   }
   // Handle decode mode
   else if let Some(mut hex_string) = args.decode {
     if let Some(stripped) = hex_string.strip_prefix("0x") {
       hex_string = stripped.to_string();
     }
-    let buf = hex_to_bytes(&hex_string)?;
+    let buffer = match hex_to_bytes(&hex_string) {
-    let frame = ADSBFrame::decode(&buf)?;
+      Ok(buffer) => buffer,
-
+      Err(err) => {
-    log::info!("{}", frame);
+        eprintln!("Unable to convert hex to bytes: {:?}", err);
-    Ok(())
+        return;
+      }
+    };
+    if let Ok(frame) = ADSBFrame::decode(&buffer) {
+      println!("{:?}", frame);
+    };
   } else {
-    log::warn!("No connection specified");
+    eprintln!("No connection specified");
-    Ok(())
   }
 }

api/src/main.rs

@@ -4,7 +4,6 @@ use actix_cors::Cors;
 use actix_web::{App, HttpServer, middleware::Logger, web};
 use dotenv::from_filename;
 use reqwest::Certificate;
-use uuid::Uuid;
 use crate::account::hash;
 use crate::users::{User, ADMIN_ROLE};