Headless runner for console[CPP-318]. (#123)

* Headless runner for console.

* Update console_backend/Cargo.toml

Co-authored-by: Jason Mobarak <[email protected]>

* Log messages.

Co-authored-by: Jason Mobarak <[email protected]>

Author: john-michaelburke

Cargo.lock

@@ -152,6 +152,18 @@ dependencies = [
 ]
 run_task = "start-console"
 
+[tasks.headless-run]
+command = "cargo"
+args = [
+  "run",
+  "--bin",
+  "headless-console",
+  "--features",
+  "headless-console",
+  "--no-default-features",
+  "${@}",
+]
+
 [tasks.store-version]
 script_runner = "@shell"
 script = '''

Makefile.toml

@@ -40,6 +40,7 @@ minreq = { version = "2.4.2", features = ["https"] }
 regex = { version = "1.5.4" }
 semver = { version = "1" }
 rust-ini = "0.17.0"
+snap = { version = "1", optional = true }
 
 [dependencies.sbp]
 git = "https://github.com/swift-nav/libsbp.git"
@@ -80,8 +81,15 @@ path = "src/bin/fft_monitor.rs"
 bench = false
 required-features = ["fft"]
 
+[[bin]]
+name = "headless-console"
+path = "src/bin/headless-console.rs"
+bench = false
+required-features = ["headless-console"]
+
 [features]
 default = ["pyo3"]
 benches = []
 tests = []
 fft = ["serde-pickle"]
+headless-console = ["snap"]

console_backend/Cargo.toml

@@ -14,7 +14,8 @@ extern crate console_backend;
 use console_backend::{
     connection::Connection,
     process_messages,
-    types::{ClientSender, RealtimeDelay, SharedState},
+    shared_state::SharedState,
+    types::{ClientSender, RealtimeDelay},
 };
 
 const BENCH_FILEPATH: &str = "./tests/data/piksi-relay.sbp";

console_backend/benches/cpu_benches.rs

@@ -6,7 +6,8 @@ use capnp::message::Builder;
 use crate::constants::*;
 use crate::errors::GET_MUT_OBJECT_FAILURE;
 use crate::fusion_status_flags::FusionStatusFlags;
-use crate::types::{CapnProtoSender, Deque, SharedState};
+use crate::shared_state::SharedState;
+use crate::types::{CapnProtoSender, Deque};
 use crate::utils::serialize_capnproto_builder;
 
 /// AdvancedInsTab struct.

console_backend/src/advanced_ins_tab.rs

@@ -4,7 +4,8 @@ use capnp::message::Builder;
 
 use crate::constants::{MAGNETOMETER_Y_AXIS_PADDING_MULTIPLIER, NUM_POINTS};
 use crate::errors::GET_MUT_OBJECT_FAILURE;
-use crate::types::{CapnProtoSender, Deque, SharedState};
+use crate::shared_state::SharedState;
+use crate::types::{CapnProtoSender, Deque};
 use crate::utils::serialize_capnproto_builder;
 
 /// AdvancedMagnetometerTab struct.

console_backend/src/advanced_magnetometer_tab.rs

@@ -3,7 +3,8 @@ use log::error;
 use crate::constants::{AMPLITUDES, CHANNELS, FREQUENCIES};
 use crate::errors::SHARED_STATE_LOCK_MUTEX_FAILURE;
 use crate::fft_monitor::FftMonitor;
-use crate::types::{CapnProtoSender, SharedState, Specan};
+use crate::shared_state::SharedState;
+use crate::types::{CapnProtoSender, Specan};
 use crate::utils::serialize_capnproto_builder;
 use capnp::message::Builder;
 

console_backend/src/advanced_spectrum_analyzer_tab.rs

@@ -12,9 +12,9 @@ use crate::constants::*;
 use crate::date_conv::*;
 use crate::output::BaselineLog;
 use crate::piksi_tools_constants::EMPTY_STR;
+use crate::shared_state::SharedState;
 use crate::types::{
-    BaselineNED, CapnProtoSender, Deque, GnssModes, GpsTime, MsgSender, Result, SharedState,
-    UtcDateTime,
+    BaselineNED, CapnProtoSender, Deque, GnssModes, GpsTime, MsgSender, Result, UtcDateTime,
 };
 use crate::utils::*;
 

console_backend/src/baseline_tab.rs

@@ -0,0 +1,62 @@
+use anyhow::Result;
+use chrono::prelude::*;
+use console_backend::{
+    cli_options::{handle_cli, CliOptions},
+    connection::ConnectionState,
+    log_panel::setup_logging,
+    server_recv_thread::server_recv_thread,
+    shared_state::SharedState,
+    types::ClientSender,
+    utils::{refresh_loggingbar, refresh_navbar},
+};
+use crossbeam::channel;
+use std::fs::File;
+use std::io::{prelude::*, BufWriter};
+
+fn main() -> Result<()> {
+    let opt = CliOptions::from_filtered_cli();
+    if opt.input.is_none() {
+        eprintln!(
+            r#"
+Running in headless mode only command line options work.
+Help:
+./headless-console --help
+Usage:
+./headless-console tcp piksi-relay-bb9f2b10e53143f4a816a11884e679cf.ce.swiftnav.com --port=55555
+"#
+        );
+        return Ok(());
+    }
+    let (client_send_, client_recv) = channel::unbounded::<Vec<u8>>();
+    let (_server_send, server_recv) = channel::unbounded::<Vec<u8>>();
+    let client_send = ClientSender::new(client_send_);
+    setup_logging(client_send.clone(), true);
+    let shared_state = SharedState::new();
+    let connection_state = ConnectionState::new(client_send.clone(), shared_state.clone());
+    handle_cli(opt, &connection_state, shared_state.clone());
+    refresh_navbar(&mut client_send.clone(), shared_state.clone());
+    refresh_loggingbar(&mut client_send.clone(), shared_state.clone());
+    server_recv_thread(connection_state, client_send, server_recv, shared_state);
+
+    let local: DateTime<Local> = Local::now();
+    let mut filename = format!("headless-console-{}.data", local);
+    filename.retain(|c| !c.is_whitespace());
+    let file_out = File::create(filename)?;
+    let buf_out = BufWriter::new(file_out);
+    let mut msg_count: usize = 0;
+    let mut file_out = snap::write::FrameEncoder::new(buf_out);
+    while let Ok(msg) = client_recv.recv() {
+        match file_out.write_all(&msg) {
+            Ok(_) => {
+                msg_count += 1;
+                if msg_count % 100 == 0 {
+                    println!("Messages received: {}", msg_count);
+                }
+            }
+            Err(err) => {
+                eprintln!("{}", err);
+            }
+        }
+    }
+    Ok(())
+}

console_backend/src/bin/headless-console.rs

@@ -7,11 +7,14 @@ use std::{
 use strum::VariantNames;
 
 use crate::constants::{AVAILABLE_BAUDRATES, AVAILABLE_REFRESH_RATES};
+use crate::errors::{CONVERT_TO_STR_FAILURE, SHARED_STATE_LOCK_MUTEX_FAILURE};
 use crate::log_panel::LogLevel;
-use crate::types::FlowControl;
+use crate::output::CsvLogging;
+use crate::shared_state::SharedState;
+use crate::types::{FlowControl, RealtimeDelay};
 use crate::{
     common_constants::{SbpLogging, Tabs},
-    connection::Connection,
+    connection::{Connection, ConnectionState},
 };
 
 #[derive(Debug)]
@@ -233,3 +236,48 @@ fn is_baudrate(br: &str) -> Result<(), String> {
         AVAILABLE_BAUDRATES
     ))
 }
+
+/// Start connections based on CLI options.
+///
+/// # Parameters
+/// - `opt`: CLI Options to start specific connection type.
+/// - `connection_state`: The Server state to start a specific connection.
+/// - `client_send`: Client Sender channel for communication from backend to frontend.
+/// - `shared_state`: The shared state for validating another connection is not already running.
+pub fn handle_cli(opt: CliOptions, connection_state: &ConnectionState, shared_state: SharedState) {
+    if let Some(opt_input) = opt.input {
+        match opt_input {
+            Input::Tcp { host, port } => {
+                connection_state.connect_to_host(host, port);
+            }
+            Input::File { file_in } => {
+                let filename = file_in.display().to_string();
+                connection_state.connect_to_file(filename, RealtimeDelay::On, opt.exit_after);
+            }
+            Input::Serial {
+                serialport,
+                baudrate,
+                flow_control,
+            } => {
+                let serialport = serialport.display().to_string();
+                connection_state.connect_to_serial(serialport, baudrate, flow_control);
+            }
+        }
+    }
+    if let Some(folder) = opt.dirname {
+        shared_state.set_logging_directory(PathBuf::from(folder));
+    }
+    let log_level = if let Some(log_level_) = opt.log_level {
+        (*log_level_).clone()
+    } else {
+        LogLevel::INFO
+    };
+    shared_state.set_log_level(log_level);
+    let mut shared_data = shared_state.lock().expect(SHARED_STATE_LOCK_MUTEX_FAILURE);
+    (*shared_data).logging_bar.csv_logging = CsvLogging::from(opt.csv_log);
+    if let Some(sbp_log) = opt.sbp_log {
+        (*shared_data).logging_bar.sbp_logging =
+            SbpLogging::from_str(&sbp_log.to_string()).expect(CONVERT_TO_STR_FAILURE);
+    }
+    log::logger().flush();
+}

console_backend/src/cli_options.rs

@@ -1,6 +1,7 @@
 use crate::constants::*;
 use crate::errors::*;
 use crate::process_messages::process_messages;
+use crate::shared_state::SharedState;
 use crate::types::*;
 use anyhow::anyhow;
 use crossbeam::channel::{unbounded, Receiver, Sender};

console_backend/src/connection.rs

@@ -1,11 +1,10 @@
 pub(crate) const HEARTBEAT_LOCK_MUTEX_FAILURE: &str = "unable to lock heartbeat mutex";
-pub(crate) const SHARED_STATE_LOCK_MUTEX_FAILURE: &str = "unable to lock shared_state mutex";
+pub const SHARED_STATE_LOCK_MUTEX_FAILURE: &str = "unable to lock shared_state mutex";
 pub(crate) const UPDATE_STATUS_LOCK_MUTEX_FAILURE: &str = "unable to lock update status mutex";
 pub(crate) const CAP_N_PROTO_SERIALIZATION_FAILURE: &str = "unable to serialize capnproto message";
 #[allow(dead_code)]
-pub(crate) const CAP_N_PROTO_DESERIALIZATION_FAILURE: &str =
-    "unable to deserialize capnproto message";
-pub(crate) const CONVERT_TO_STR_FAILURE: &str = "error converting to str";
+pub const CAP_N_PROTO_DESERIALIZATION_FAILURE: &str = "unable to deserialize capnproto message";
+pub const CONVERT_TO_STR_FAILURE: &str = "error converting to str";
 pub(crate) const GET_MUT_OBJECT_FAILURE: &str = "error trying to get mut object";
 pub(crate) const UNABLE_TO_STOP_TIMER_THREAD_FAILURE: &str = "unable to kill running timer thread";
 pub(crate) const UNABLE_TO_SEND_INS_UPDATE_FAILURE: &str = "unable to send an ins status update";

console_backend/src/errors.rs

@@ -14,8 +14,8 @@ use crate::errors::{
     THREAD_JOIN_FAILURE, UNABLE_TO_SEND_INS_UPDATE_FAILURE, UNABLE_TO_STOP_TIMER_THREAD_FAILURE,
     UPDATE_STATUS_LOCK_MUTEX_FAILURE,
 };
-use crate::types::ArcBool;
-use crate::types::{CapnProtoSender, SharedState};
+use crate::shared_state::SharedState;
+use crate::types::{ArcBool, CapnProtoSender};
 use crate::utils::serialize_capnproto_builder;
 
 const STATUS_PERIOD: f64 = 1.0;

console_backend/src/fusion_status_flags.rs

@@ -25,7 +25,9 @@ pub mod process_messages;
 #[cfg(not(test))]
 #[cfg(all(not(feature = "benches"), not(feature = "tests"), feature = "pyo3"))]
 pub mod server;
+pub mod server_recv_thread;
 pub mod settings_tab;
+pub mod shared_state;
 pub mod solution_tab;
 pub mod solution_velocity_tab;
 pub mod status_bar;
@@ -62,7 +64,7 @@ struct Tabs<'link, S: types::CapnProtoSender> {
 
 impl<'link, S: types::CapnProtoSender> Tabs<'link, S> {
     fn new(
-        shared_state: types::SharedState,
+        shared_state: shared_state::SharedState,
         client_sender: S,
         msg_sender: types::MsgSender,
         link: sbp::link::Link<'link, ()>,

console_backend/src/lib.rs

@@ -6,7 +6,7 @@ use capnp::message::Builder;
 use crate::common_constants as cc;
 use crate::constants::LOG_WRITER_BUFFER_MESSAGE_COUNT;
 use crate::errors::CONSOLE_LOG_JSON_TO_STRING_FAILURE;
-use crate::types::*;
+use crate::types::{CapnProtoSender, ClientSender};
 use crate::utils::serialize_capnproto_builder;
 
 use async_logger::Writer;
@@ -96,8 +96,8 @@ pub fn handle_log_msg(msg: MsgLog) {
     }
 }
 
-pub fn setup_logging(client_sender: ClientSender) {
-    let log_panel = LogPanelWriter::new(client_sender);
+pub fn setup_logging(client_sender: ClientSender, debug: bool) {
+    let log_panel = LogPanelWriter::new(client_sender, debug);
     let logger = Logger::builder()
         .buf_size(LOG_WRITER_BUFFER_MESSAGE_COUNT)
         .formatter(splitable_log_formatter)
@@ -111,11 +111,15 @@ pub fn setup_logging(client_sender: ClientSender) {
 #[derive(Debug)]
 pub struct LogPanelWriter<S: CapnProtoSender> {
     pub client_sender: S,
+    pub debug: bool,
 }
 
 impl<S: CapnProtoSender> LogPanelWriter<S> {
-    pub fn new(client_sender: S) -> LogPanelWriter<S> {
-        LogPanelWriter { client_sender }
+    pub fn new(client_sender: S, debug: bool) -> LogPanelWriter<S> {
+        LogPanelWriter {
+            client_sender,
+            debug,
+        }
     }
 }
 
@@ -132,6 +136,9 @@ impl<S: CapnProtoSender> Writer<Box<String>> for LogPanelWriter<S> {
         let mut entries = log_update.init_entries(slice.len() as u32);
 
         for (idx, item) in slice.iter().enumerate() {
+            if self.debug {
+                eprintln!("{}", item);
+            }
             let mut entry = entries.reborrow().get(idx as u32);
 
             entry.set_line(&**item);

console_backend/src/log_panel.rs

@@ -5,9 +5,13 @@ use log::{debug, error};
 use sbp::{messages::SBP, time::GpsTime};
 
 use crate::common_constants::SbpLogging;
-use crate::constants::*;
-use crate::output::*;
-use crate::types::*;
+use crate::constants::{
+    BASELINE_TIME_STR_FILEPATH, POS_LLH_TIME_STR_FILEPATH, SBP_FILEPATH, SBP_JSON_FILEPATH,
+    VEL_TIME_STR_FILEPATH,
+};
+use crate::output::{CsvLogging, SbpLogger};
+use crate::shared_state::{create_directory, SharedState};
+use crate::types::CapnProtoSender;
 use crate::utils::refresh_loggingbar;
 
 pub struct MainTab<S: CapnProtoSender> {
@@ -191,7 +195,7 @@ mod tests {
     use super::*;
     use crate::baseline_tab::BaselineTab;
     use crate::solution_tab::SolutionTab;
-    use crate::types::{PosLLH, VelNED};
+    use crate::types::{BaselineNED, MsgSender, PosLLH, TestSender, VelNED};
     use crate::utils::{mm_to_m, ms_to_sec};
     use glob::glob;
     use sbp::messages::navigation::{MsgBaselineNED, MsgPosLLH, MsgVelNED};

console_backend/src/main_tab.rs

@@ -3,7 +3,8 @@ use capnp::message::Builder;
 use log::warn;
 use std::collections::{BTreeMap, HashMap};
 
-use crate::types::*;
+use crate::shared_state::SharedState;
+use crate::types::{CapnProtoSender, ObservationMsg, SignalCodes};
 use crate::utils::{compute_doppler, sec_to_ns};
 
 use crate::utils::serialize_capnproto_builder;