init commit

src/main.rs

@@ -0,0 +1,246 @@
+use obd2::{commands::Obd2DataRetrieval, device::{Elm327, SerialPort}, Obd2};
+use std::thread;
+use std::thread::JoinHandle;
+use std::time::Duration;
+use anyhow::anyhow;
+use std::sync::{Arc, Mutex};
+use std::collections::VecDeque;
+use gpsd_client::*;
+
+mod auto_events;
+use auto_events::{AutoEvent, AutoEventType};
+
+mod file_format;
+
+
+
+static THREAD_DELAY: Duration = Duration::from_millis(1_000);
+static trip_id: Mutex<u32> = Mutex::new(0);
+
+fn get_speed(device: &mut Obd2::<Elm327<SerialPort>>) -> ::anyhow::Result<AutoEventType> {
+    if let Some(speed_kmph) = device.get_speed()?.first() {
+        let speed_kmph_u32: u32 = speed_kmph.clone().into();
+        let speed_mph: u32 = (speed_kmph_u32 * 621371) / 1000000;
+
+        Ok(auto_events::SpeedEvent::new(speed_mph))
+    } else {
+        Err(anyhow!("Unable to read speed from OBDII"))
+    }
+}
+
+fn get_rpm(device: &mut Obd2::<Elm327<SerialPort>>) -> ::anyhow::Result<AutoEventType> {
+    if let Some(rpm) = device.get_rpm()?.first() {
+        Ok(auto_events::RPMEvent::new(rpm.round() as u32))
+    } else {
+        Err(anyhow!("Unable to read rpm from OBDII"))
+    }
+}
+
+fn get_throttle(device: &mut Obd2::<Elm327<SerialPort>>) -> ::anyhow::Result<AutoEventType> {
+    if let Some(throttle_level) = device.get_throttle_position()?.first() {
+        Ok(auto_events::ThrottleEvent::new(f32::from(throttle_level.clone())/255.0))
+    } else {
+        Err(anyhow!("Unable to read rpm from OBDII"))
+    }
+}
+
+fn get_fuel_level(device: &mut Obd2::<Elm327<SerialPort>>) -> ::anyhow::Result<AutoEventType> {
+    if let Some(fuel_level) = device.get_fuel_level()?.first() {
+        Ok(auto_events::FuelLevelEvent::new(f32::from(fuel_level.clone())/255.0))
+    } else {
+        Err(anyhow!("Unable to read rpm from OBDII"))
+    }
+}
+
+fn wait_for_reconnect(device: &mut Option<Obd2<Elm327<SerialPort>>>, vin: &mut Option<String>) {
+    loop {
+        thread::sleep(THREAD_DELAY);
+
+        *device = None;
+
+        if let Ok(port) = SerialPort::new("/dev/ttyUSB0") {
+            match Elm327::new(port) {
+                Ok(new_device) =>{
+                if let Ok(mut dev) = Obd2::new(new_device) {
+                    if let Err(x) = dev.reset() {
+                        log::error!("OBDII initialization failed, retrying... {}", x);
+                        continue;
+                    }
+
+                    log::info!("Connected to OBDII");
+
+                    match dev.get_vin() {
+                        Ok(new_vin) => {
+                            *vin = Some(new_vin.clone());
+                            match vin_parser::get_info(new_vin.as_str()) {
+                                Ok(vin_result) => {
+                                    let year = vin_result.years().last().map_or("Unknown Year".to_string(), |y| y.to_string());
+                                    log::info!("Vehicle is: {} {}", vin_result.manufacturer, year);
+                                }
+                                Err(_) => log::warn!("Unable to decode VIN"),
+                            }
+                        }
+                        Err(_) => {
+                            *vin = None;
+                            log::warn!("Unable to read VIN");
+                        }
+                    }
+
+                    *device = Some(dev);
+                    return;
+                } else {
+                    log::warn!("Reconnect failed. Ignition may be off.");
+                }
+                },
+                Err(x)=>{
+                    log::warn!("Reconnect failed. Ignition may be off. {}", x);
+                }
+            }
+        }
+    };
+
+}
+
+fn event_handler(events: Arc<Mutex<VecDeque<AutoEvent>>>) {
+    thread::sleep(THREAD_DELAY / 2);
+    let mut cur_tid: u32 = trip_id.lock().unwrap().clone();
+
+    loop {
+
+        let cur_datetime = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap();
+        let mut file = file_format::File::create(format!("./out_{}.obt", cur_datetime.as_secs()).as_str()).unwrap();
+
+        loop {
+            thread::sleep(THREAD_DELAY);
+
+            let tid = trip_id.lock().unwrap();
+
+            if *tid > cur_tid {
+               file.write().unwrap(); 
+               cur_tid = tid.clone();
+               break;
+            }
+
+            let ref mut el = events.lock().unwrap();
+            while !el.is_empty() {
+                if let Some(event) = el.pop_front() {
+                    log::info!("{:?}", event);
+                    file.add_record(event);
+                }
+            }
+
+        };
+    };
+}
+
+fn unix_ts() -> i64 {
+    let time  = chrono::Utc::now();
+    time.timestamp_millis()
+}
+
+fn event_getter(events: Arc<Mutex<VecDeque<AutoEvent>>>, device: Arc<Mutex<Option<Obd2<Elm327<SerialPort>>>>>, vin: Arc<Mutex<Option<String>>>) {
+    let event_getters: Vec<fn(&mut Obd2::<Elm327<SerialPort>>) -> ::anyhow::Result<AutoEventType>> = vec![
+        get_speed, 
+        get_rpm,
+        get_throttle,
+        get_fuel_level
+    ];
+
+    let mut ignition_on: bool = false;
+    let mut gps: GPS = GPS::connect().unwrap();
+
+    'record_loop: loop {
+        thread::sleep(THREAD_DELAY);
+        let mut full_event: AutoEvent = AutoEvent{ content: vec![], timestamp: 0 };
+
+        for event in &event_getters {
+            let ref mut x = device.lock().unwrap();
+            if let Some(ref mut dev) = *(*x) {
+                match event(dev) {
+                    Ok(event) => {
+                        // Derive ignition on/off events from RPM 
+                        if let AutoEventType::RPM(ev) = event.clone() {
+                            if ev.rpm > 0 && !ignition_on{
+                                log::info!("Ignition ON");
+                                ignition_on = true;
+                                continue 'record_loop;
+                            } else if ignition_on && ev.rpm == 0 {
+                                ignition_on = false;
+                                {
+                                    let mut mtx = trip_id.lock().unwrap();
+                                    *mtx += 1;
+                                }
+                                log::info!("Ignition OFF");
+                                continue 'record_loop;
+                            }
+                        }
+
+                        if ignition_on {
+                            full_event.content.push(event);
+                        }
+                    },
+                    Err(err) => {
+                        if let Some(obd2::Error::Device(obd2::error::DeviceError(obd2::device::Error::IO(io_err)))) = err.downcast_ref::<obd2::Error>() {
+                            if io_err.kind() == std::io::ErrorKind::BrokenPipe {
+                                log::warn!("Disconnected from OBDII reader. Ignition may be off. Attempting to reconnect.");
+                                wait_for_reconnect(x, &mut vin.lock().unwrap());
+                                break;
+                            }
+                        } else if let Some(obd2::Error::Device(obd2::error::DeviceError(obd2::device::Error::Communication(_)))) = err.downcast_ref::<obd2::Error>() {
+                                log::error!("OBDII connection was broken. Resetting connection.");
+                                wait_for_reconnect(x, &mut vin.lock().unwrap());
+                            log::error!("{}", err);
+                        } else {
+                            log::error!("{}", err);
+                        }
+                    }
+                } 
+            }
+        }
+
+        if full_event.content.len() == 0 {
+            continue;
+        }
+        
+        let data: GPSData = gps.current_data().unwrap();
+        full_event.content.push(auto_events::GPSLocationEvent::new(data.lat, data.lon));
+        let speed_ms = data.speed;
+        let speed_mph = speed_ms * 2.2369362921; 
+        full_event.content.push(auto_events::GPSSpeedEvent::new(speed_mph.round() as u32));
+        full_event.timestamp = unix_ts();
+
+        events.lock().unwrap().push_back(full_event);
+    }
+}
+
+fn main() -> ::anyhow::Result<()> {
+    file_format::File::open("weis_drive.obt")?.write_json()?;
+    return Ok(());
+    env_logger::init();
+
+    let events: Arc<Mutex<VecDeque<AutoEvent>>> = Arc::new(Mutex::new(vec![].into()));
+    let device: Arc<Mutex<Option<Obd2<Elm327<SerialPort>>>>> = Arc::new(Mutex::new(None));
+    let vin: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
+
+    log::info!("Starting On Board Telematics v1.0.0");
+
+    wait_for_reconnect(&mut *device.lock().unwrap(), &mut vin.lock().unwrap());
+
+
+    // Start worker threads
+    let mut threads: Vec<JoinHandle<()>> = vec![];
+    let mut events_cloned = Arc::clone(&events);
+    /*hreads.push(*//*)*/
+    thread::spawn(move|| {event_handler(events_cloned)});
+
+    events_cloned = Arc::clone(&events);
+    let device_cloned = Arc::clone(&device);
+    let vin_cloned = Arc::clone(&vin);
+    threads.push(thread::spawn(move|| {event_getter(events_cloned, device_cloned, vin_cloned)}));
+
+    for thread in threads {
+        thread.join().unwrap();
+    }
+
+    Ok(())
+}