174 lines
5.9 KiB
Rust
174 lines
5.9 KiB
Rust
use std::collections::HashMap;
|
|
use std::sync::{Arc, RwLock};
|
|
use std::time::{Duration, Instant};
|
|
use futures::stream::StreamExt;
|
|
use tokio::sync::mpsc::channel;
|
|
use adsb_deku::ICAO;
|
|
use super::beast;
|
|
|
|
#[derive(Default)]
|
|
pub struct Entry {
|
|
pub category: Option<(adsb_deku::adsb::TypeCoding, u8)>,
|
|
pub callsign: Option<String>,
|
|
pub altitude: Option<u32>,
|
|
cpr1: Option<adsb_deku::Altitude>,
|
|
cpr2: Option<adsb_deku::Altitude>,
|
|
position: Option<adsb_deku::cpr::Position>,
|
|
pub heading: Option<f32>,
|
|
pub speed: Option<f64>,
|
|
pub vertical_rate: Option<i16>,
|
|
last_update: Option<Instant>,
|
|
pub squawk: Option<u32>,
|
|
pub seen: Option<Instant>, // Option wrapper is just for derive Default
|
|
pub seen_pos: Option<Instant>,
|
|
pub rssi: Option<f32>,
|
|
}
|
|
|
|
impl Entry {
|
|
const MAX_AGE: u64 = 300;
|
|
|
|
fn update(&mut self, kind: adsb_deku::adsb::ME, rssi: Option<f32>) {
|
|
self.seen = Some(Instant::now());
|
|
rssi.map(|rssi| self.rssi = Some(rssi));
|
|
|
|
match kind {
|
|
adsb_deku::adsb::ME::AirbornePositionBaroAltitude(altitude) |
|
|
adsb_deku::adsb::ME::AirbornePositionGNSSAltitude(altitude) => {
|
|
self.altitude = altitude.alt.map(|alt| alt.into());
|
|
if self.cpr2.map_or(false, |cpr2| cpr2.odd_flag != altitude.odd_flag) {
|
|
// if last altitude had a different odd flag,
|
|
// shift the entries
|
|
self.cpr1 = self.cpr2.take();
|
|
}
|
|
// add the new entry
|
|
self.cpr2 = Some(altitude);
|
|
|
|
match (&self.cpr1, &self.cpr2) {
|
|
(Some(cpr1), Some(cpr2)) => {
|
|
if let Some(pos) = adsb_deku::cpr::get_position((cpr1, cpr2)) {
|
|
if pos.latitude < -90. || pos.latitude > 90. ||
|
|
pos.longitude < -180. || pos.longitude > 180.
|
|
{
|
|
eprintln!("invalid position: {:?}", pos);
|
|
} else {
|
|
self.position = Some(pos);
|
|
self.seen_pos = Some(Instant::now());
|
|
}
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
adsb_deku::adsb::ME::AirborneVelocity(velocity) => {
|
|
if let Some((heading, speed, vrate)) = velocity.calculate() {
|
|
self.heading = Some(heading);
|
|
self.speed = Some(speed);
|
|
self.vertical_rate = Some(vrate);
|
|
}
|
|
}
|
|
adsb_deku::adsb::ME::AircraftIdentification(ident) => {
|
|
self.category = Some((ident.tc, ident.ca));
|
|
self.callsign = Some(ident.cn);
|
|
}
|
|
adsb_deku::adsb::ME::AircraftStatus(status) => {
|
|
self.squawk = Some(status.squawk);
|
|
}
|
|
adsb_deku::adsb::ME::TargetStateAndStatusInformation(_) => {}
|
|
adsb_deku::adsb::ME::AircraftOperationStatus(_) => {}
|
|
msg => {
|
|
eprintln!("unhandled adsb msg: {:?}", msg);
|
|
}
|
|
}
|
|
|
|
self.last_update = Some(Instant::now());
|
|
}
|
|
|
|
pub fn position(&self) -> Option<&adsb_deku::cpr::Position> {
|
|
self.position.as_ref()
|
|
}
|
|
|
|
pub fn flight(&self) -> Option<&str> {
|
|
self.callsign.as_ref().map(|callsign| {
|
|
callsign.split(char::is_whitespace)
|
|
.next().unwrap()
|
|
})
|
|
}
|
|
|
|
pub fn altitude_m(&self) -> Option<f64> {
|
|
self.altitude.map(|altitude| altitude as f64 * 0.3048)
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct Aircrafts {
|
|
state: Arc<RwLock<HashMap<ICAO, RwLock<Entry>>>>,
|
|
}
|
|
|
|
impl Aircrafts {
|
|
pub fn new() -> Self {
|
|
Aircrafts {
|
|
state: Arc::new(RwLock::new(HashMap::new())),
|
|
}
|
|
}
|
|
|
|
pub fn read(&self) -> std::sync::RwLockReadGuard<HashMap<ICAO, RwLock<Entry>>> {
|
|
self.state.read().unwrap()
|
|
}
|
|
|
|
pub fn connect(&self, host: &'static str, port: u16) {
|
|
// buffering channel because readsb is very sensitive
|
|
let (tx, mut rx) = channel(16 * 1024);
|
|
|
|
// network input
|
|
tokio::spawn(async move {
|
|
loop {
|
|
let mut stream;
|
|
if let Ok(stream_) = beast::connect(host, port).await {
|
|
stream = Box::pin(stream_);
|
|
} else {
|
|
tokio::time::sleep(Duration::from_secs(1)).await;
|
|
// Retry
|
|
continue;
|
|
}
|
|
|
|
while let Some(frame) = stream.next().await {
|
|
let _ = tx.send(frame).await;
|
|
}
|
|
}
|
|
});
|
|
|
|
// state update
|
|
let state = self.state.clone();
|
|
tokio::spawn(async move {
|
|
while let Some(frame) = rx.recv().await {
|
|
match frame.parse_adsb() {
|
|
Some(adsb_deku::Frame { df: adsb_deku::DF::ADSB(adsb), .. }) => {
|
|
state.write().unwrap()
|
|
.entry(adsb.icao)
|
|
.or_default()
|
|
.write().unwrap()
|
|
.update(adsb.me, frame.rssi());
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
});
|
|
|
|
// discard old states
|
|
let state = self.state.clone();
|
|
tokio::spawn(async move {
|
|
loop {
|
|
state.write().unwrap().
|
|
retain(|_, entry| {
|
|
entry.read().unwrap()
|
|
.last_update.map(|last_update| {
|
|
last_update + Duration::from_secs(Entry::MAX_AGE) > Instant::now()
|
|
})
|
|
.unwrap_or(false)
|
|
});
|
|
tokio::time::sleep(Duration::from_secs(1)).await;
|
|
}
|
|
});
|
|
}
|
|
}
|