diff --git a/src/device.rs b/src/device.rs index 583a509..38d5f1f 100644 --- a/src/device.rs +++ b/src/device.rs @@ -31,6 +31,14 @@ pub struct DeviceSocket { pub ip: IpAddr, } +#[derive(Debug, Component, Default, Clone, Copy)] +pub struct DeviceDetector { + pub blip_threshold: usize, + pub blip_size: usize, + pub max_duty: u16, + pub duty: u16, +} + fn on_remove_device(mut world: DeferredWorld, context: HookContext) { let component = world .entity(context.entity) @@ -110,8 +118,28 @@ pub struct StormLevels { levels: Vec, } +#[derive(Debug)] +pub enum ConnectionState { + Disconnected, + Connecting, + Connected, +} + +impl core::fmt::Display for ConnectionState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Disconnected => write!(f, "Disconnected"), + Self::Connecting => write!(f, "Connecting"), + Self::Connected => write!(f, "Connected"), + } + } +} + #[derive(Debug, Resource)] -pub struct ConnectedDevice(pub Entity); +pub struct ConnectedDevice { + pub device: Entity, + pub connection_state: ConnectionState, +} fn register_devices( incoming: Res, diff --git a/src/net.rs b/src/net.rs index cb8cead..0458644 100644 --- a/src/net.rs +++ b/src/net.rs @@ -1,6 +1,6 @@ use core::net::IpAddr; use std::{ - io::Read, + io::{Read, Write}, net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpStream, UdpSocket}, time::Duration, }; @@ -27,13 +27,23 @@ use socket2::{Domain, Protocol, Socket, Type}; #[derive(Debug, Resource)] pub struct DiscoverResponse(pub Receiver); +pub enum StrikeUpdateState { + Disconnected, + Connecting, + Connected, + Updating(striker_proto::StrikerResponse), +} + #[derive(Debug, Resource)] -pub struct StrikeUpdates(pub Receiver); +pub struct StrikeUpdates(pub Receiver); #[derive(Debug, Resource)] -pub struct StrikeConnect(pub Sender); +pub struct StrikeRequests(pub Sender); -pub enum StrikeConnection { +#[derive(Debug, Resource)] +pub struct StrikeActions(pub Sender); + +pub enum StrikeAction { Connect(SocketAddr), Disconnect, } @@ -136,28 +146,60 @@ pub fn setup_mdns_task(mut commands: Commands) -> Result { pub fn setup_strike_connection(mut commands: Commands) { let io = IoTaskPool::get(); - let (signal_tx, signal_rx) = async_channel::bounded(1); + let (signal_tx, signal_rx) = async_channel::bounded(2); + let (req_tx, req_rx) = async_channel::bounded(1); let (resp_tx, resp_rx) = async_channel::bounded(64); io.spawn(async move { let mut read_buf = vec![0u8; 4096]; + let mut write_buf = vec![0u8; 4096]; - while let Ok(StrikeConnection::Connect(addr)) = signal_rx.recv().await { + while let Ok(StrikeAction::Connect(addr)) = signal_rx.recv().await { let net_fut = async { loop { + resp_tx.send(StrikeUpdateState::Connecting).await.ok(); let Ok(stream) = Async::::connect(addr).await else { Timer::after(Duration::from_secs(1)).await; continue; }; - while let Ok(read) = stream.read_with(|mut a| a.read(&mut read_buf)).await { - let Ok(data) = striker_proto::receive_response(&mut read_buf[..read]) else { continue }; + resp_tx.send(StrikeUpdateState::Connected).await.ok(); + stream.write_with(|s| s.set_nodelay(true)).await.ok(); - resp_tx - .send(data) - .await - .ok(); - } + let read_fut = async { + while let Ok(read) = stream.read_with(|mut a| a.read(&mut read_buf)).await { + let Ok(data) = striker_proto::receive_response(&mut read_buf[..read]) + else { + continue; + }; + + if resp_tx.send(StrikeUpdateState::Updating(data)).await.is_err() { + break; + } + } + }; + + let write_fut = async { + while let Ok(req) = req_rx.recv().await { + let Ok(payload) = striker_proto::send_request(req, &mut write_buf) + else { + continue; + }; + + if stream.write_with(|mut s| s.write(payload)).await.is_err() { + break; + } + } + }; + + (read_fut, write_fut).race().await; + + stream + .write_with(|s| s.shutdown(std::net::Shutdown::Both)) + .await + .ok(); + + break; } }; @@ -165,18 +207,20 @@ pub fn setup_strike_connection(mut commands: Commands) { while signal_rx .recv() .await - .is_ok_and(|strike| !matches!(strike, StrikeConnection::Disconnect)) + .is_ok_and(|strike| !matches!(strike, StrikeAction::Disconnect)) { } }; (net_fut, cancel_fut).race().await; + resp_tx.send(StrikeUpdateState::Disconnected).await.ok(); } }) .detach(); - commands.insert_resource(StrikeConnect(signal_tx)); + commands.insert_resource(StrikeActions(signal_tx)); commands.insert_resource(StrikeUpdates(resp_rx)); + commands.insert_resource(StrikeRequests(req_tx)); } pub struct NetPlugin; diff --git a/src/views/home.rs b/src/views/home.rs index 14fd10e..a0ede41 100644 --- a/src/views/home.rs +++ b/src/views/home.rs @@ -15,12 +15,12 @@ use bevy::{ use bevy_ratatui::RatatuiContext; use ratatui::{ layout::{Constraint, HorizontalAlignment, Layout}, - style::{Color, Style}, + style::{Color, Style, Stylize}, widgets::{Block, List, ListDirection, ListItem, ListState, Padding, Paragraph}, }; use crate::{ - device::{ConnectedDevice, Device, DeviceSocket}, + device::{ConnectedDevice, ConnectionState, Device, DeviceSocket}, messages::StrikeMessage, net::MdnsSignaler, state::AppState, @@ -77,7 +77,10 @@ pub fn home_message_handler( .iter(world) .nth(offset) .unwrap(); - world.insert_resource(ConnectedDevice(device)); + world.insert_resource(ConnectedDevice { + device, + connection_state: ConnectionState::Disconnected, + }); let mut next_state = world.resource_mut::>(); next_state.set(AppState::Monitoring); @@ -114,7 +117,7 @@ pub fn home_view( .title("Striker") .title_alignment(HorizontalAlignment::Center) .border_style(Color::LightBlue), - ); + ).white(); let items = q_devices.iter().map(|(name, addr)| { ListItem::new(format!( @@ -125,16 +128,17 @@ pub fn home_view( let help_text = "Keys: 'q'/ESC Quit, 's' Toggle Search, UP/DOWN Choose Device(s), ENTER/SPACE Connect to Device"; - let help = Paragraph::new(help_text).block( - Block::bordered() - .padding(Padding::horizontal(2)) - .border_style(Color::LightBlue), - ); + let help = Paragraph::new(help_text) + .block( + Block::bordered() + .padding(Padding::horizontal(2)) + .border_style(Color::LightBlue), + ).white(); let list = List::new(items) .highlight_symbol(">> ") .highlight_spacing(ratatui::widgets::HighlightSpacing::Always) - .highlight_style(Style::new().bg(Color::DarkGray)) + .highlight_style(Style::new().on_dark_gray().black()) .direction(ListDirection::TopToBottom) .block( Block::bordered() @@ -142,7 +146,8 @@ pub fn home_view( .title_alignment(HorizontalAlignment::Center) .padding(Padding::new(2, 2, 1, 1)) .border_style(Color::LightBlue), - ); + ) + .white(); frame.render_widget(paragraph, top); frame.render_stateful_widget(list, mid, &mut list_state.0); diff --git a/src/views/monitoring.rs b/src/views/monitoring.rs index 69b4f1e..d9bf08a 100644 --- a/src/views/monitoring.rs +++ b/src/views/monitoring.rs @@ -22,15 +22,16 @@ use ratatui::{ text::{Line, Span}, widgets::{Axis, Block, Chart, Dataset, Padding, Paragraph}, }; -use striker_proto::{StrikerResponse, Update}; +use striker_proto::{Response, StrikerRequest, StrikerResponse, Update}; use crate::{ device::{ - ConnectedDevice, Device, DeviceSocket, SignalAverage, SignalPeaks, SignalSource, Signals, - StormLevel, StormLevels, StormSignal, StormSource, Timestamp, + ConnectedDevice, ConnectionState, Device, DeviceDetector, DeviceSocket, SignalAverage, + SignalPeaks, SignalSource, Signals, StormLevel, StormLevels, StormSignal, StormSource, + Timestamp, }, messages::StrikeMessage, - net::{StrikeConnect, StrikeConnection, StrikeUpdates}, + net::{StrikeAction, StrikeActions, StrikeRequests, StrikeUpdateState, StrikeUpdates}, state::AppState, }; @@ -53,17 +54,20 @@ pub fn monitoring_message_handler( pub fn enter_monitoring( connected: Res, - signal: Res, + signal: Res, + request: Res, q_devices: Query<&DeviceSocket, With>, mut commands: Commands, ) -> Result { - let addr = q_devices.get(connected.0)?; + let addr = q_devices.get(connected.device)?; signal .0 - .try_send(StrikeConnection::Connect(SocketAddr::new( - addr.ip, addr.port, - )))?; + .try_send(StrikeAction::Connect(SocketAddr::new(addr.ip, addr.port)))?; + + request.0.force_send(StrikerRequest { + request: striker_proto::Request::DetectorInfo, + })?; commands.insert_resource(DataClear(Timer::from_seconds( 60.0 * 1.0, @@ -73,8 +77,8 @@ pub fn enter_monitoring( Ok(()) } -pub fn exit_monitoring(signal: Res, mut commands: Commands) -> Result { - signal.0.try_send(StrikeConnection::Disconnect)?; +pub fn exit_monitoring(signal: Res, mut commands: Commands) -> Result { + signal.0.try_send(StrikeAction::Disconnect)?; commands.remove_resource::(); Ok(()) @@ -107,31 +111,56 @@ pub fn clear_device_data( pub fn update_device_data( updates: Res, - connected: Res, + mut connected: ResMut, mut commands: Commands, ) -> Result { - let mut entity = commands.entity(connected.0); + let mut entity = commands.entity(connected.device); - while let Ok(StrikerResponse::Update(update)) = updates.0.try_recv() { + while let Ok(update) = updates.0.try_recv() { match update { - Update::Warning { timestamp, level } => { - entity.with_related::(( - Timestamp::from_seconds(timestamp)?, - StormLevel(level), - )); + StrikeUpdateState::Connected => { + connected.connection_state = ConnectionState::Connected; } - Update::Strike { - timestamp, - peaks, - samples, - average, - } => { - entity.with_related::(( - Timestamp::from_microseconds(timestamp)?, - SignalPeaks::new(peaks), - StormSignal::new(samples), - SignalAverage::new(average), - )); + StrikeUpdateState::Connecting => { + connected.connection_state = ConnectionState::Connecting; + } + StrikeUpdateState::Updating(response) => match response { + StrikerResponse::Response(Response::DetectorInfo { + blip_threshold, + blip_size, + max_duty, + duty, + }) => { + entity.insert(DeviceDetector { + blip_threshold, + blip_size, + max_duty, + duty, + }); + } + StrikerResponse::Update(Update::Warning { timestamp, level }) => { + entity.with_related::(( + Timestamp::from_seconds(timestamp)?, + StormLevel(level), + )); + } + StrikerResponse::Update(Update::Strike { + timestamp, + peaks, + samples, + average, + }) => { + entity.with_related::(( + Timestamp::from_microseconds(timestamp)?, + SignalPeaks::new(peaks), + StormSignal::new(samples), + SignalAverage::new(average), + )); + } + _ => {} + }, + StrikeUpdateState::Disconnected => { + connected.connection_state = ConnectionState::Disconnected; } } } @@ -142,8 +171,16 @@ pub fn update_device_data( pub fn monitoring_view( mut context: ResMut, connected: Res, - q_devices: Query<(&Name, Option<&StormLevels>, Option<&Signals>), With>, - q_levels: Query<(&Timestamp, &StormLevel)>, + q_devices: Query< + ( + &Name, + Option<&DeviceDetector>, + Option<&StormLevels>, + Option<&Signals>, + ), + With, + >, + q_levels: Query<&StormLevel>, q_signals: Query<(&Timestamp, &StormSignal)>, ) -> Result { context.draw(|frame| { @@ -154,14 +191,18 @@ pub fn monitoring_view( ]) .areas(frame.area()); - let [left, center, right] = Layout::horizontal([ + let [top_left, top_mid_left, top_mid_right, top_right] = Layout::horizontal([ + Constraint::Fill(1), Constraint::Fill(1), Constraint::Fill(1), Constraint::Fill(1), ]) .areas(top); - let (device, levels, signals) = q_devices.get(connected.0).unwrap(); + let [chart_block, detector_block] = + Layout::vertical([Constraint::Fill(1), Constraint::Length(3)]).areas(mid); + + let (device, detector, levels, signals) = q_devices.get(connected.device).unwrap(); let latest_level = levels.and_then(|c| { c.last() @@ -172,30 +213,22 @@ pub fn monitoring_view( .and_then(|s| s.last().copied()) .and_then(|entity| q_signals.get(entity).ok()); - let name = Paragraph::new(device.as_str()).block( - Block::bordered() - .padding(Padding::horizontal(2)) - .title("Device") - .title_alignment(HorizontalAlignment::Center) - .border_style(Color::LightGreen), - ); - - let timestamp_signal = latest_signal.map(|(t, _)| t); - let timestamp_level = latest_level.map(|(t, _)| t); + let name = Paragraph::new(device.as_str()) + .block( + Block::bordered() + .padding(Padding::horizontal(2)) + .title("Device") + .title_alignment(HorizontalAlignment::Center) + .border_style(Color::LightGreen), + ) + .white(); - let timestamp = match (timestamp_signal, timestamp_level) { - (Some(signal), Some(level)) => signal - .duration_since(**level) - .is_positive() - .then_some(signal) - .or(Some(level)), - (Some(time), None) | (None, Some(time)) => Some(time), - _ => None, - }; + let timestamp = latest_signal.map(|(t, _)| t); let timestamp = Paragraph::new(Line::from_iter( timestamp.map(|t| Span::from(format!("{}", t.0))), )) + .white() .block( Block::bordered() .padding(Padding::horizontal(2)) @@ -207,8 +240,9 @@ pub fn monitoring_view( let warn_level = Paragraph::new(Line::from_iter( Some(Span::raw("Level: ")) .into_iter() - .chain(latest_level.map(|(_, s)| Span::from(format!("{}", s.0)))), + .chain(latest_level.map(|s| Span::from(format!("{}", s.0)))), )) + .white() .block( Block::bordered() .padding(Padding::horizontal(2)) @@ -217,6 +251,16 @@ pub fn monitoring_view( .border_style(Color::LightGreen), ); + let connection = Paragraph::new(connected.connection_state.to_string()) + .white() + .block( + Block::bordered() + .padding(Padding::horizontal(2)) + .title("State") + .title_alignment(HorizontalAlignment::Center) + .border_style(Color::LightGreen), + ); + let data = latest_signal .map(|(_, samples)| { samples @@ -250,16 +294,33 @@ pub fn monitoring_view( .y_axis(Axis::default().bounds([-y_bounds, y_bounds])) .block(block); - let help = Paragraph::new("Keys: 'q'/ESC Quit, BACKSPACE Return to Device select").block( + let help = Paragraph::new("Keys: 'q'/ESC Quit, BACKSPACE Return to Device select") + .block( + Block::bordered() + .padding(Padding::horizontal(2)) + .border_style(Color::LightGreen), + ) + .white(); + + let detector = detector.copied().unwrap_or_default(); + + let detector_info = Paragraph::new(format!( + "Threshold: {}, Blip Size: {}, Max Duty {}, Current Duty: {}", + detector.blip_threshold, detector.blip_size, detector.max_duty, detector.duty + )) + .block( Block::bordered() .padding(Padding::horizontal(2)) .border_style(Color::LightGreen), - ); - - frame.render_widget(name, left); - frame.render_widget(timestamp, center); - frame.render_widget(warn_level, right); - frame.render_widget(chart, mid); + ) + .white(); + + frame.render_widget(name, top_left); + frame.render_widget(timestamp, top_mid_left); + frame.render_widget(warn_level, top_mid_right); + frame.render_widget(connection, top_right); + frame.render_widget(chart, chart_block); + frame.render_widget(detector_info, detector_block); frame.render_widget(help, bottom); })?;