summaryrefslogtreecommitdiff
path: root/src/main.rs
blob: e4f2e7eec6daa4502da5bf8ac30b90a7ac9554ad (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
extern crate futures;
extern crate mio;
extern crate socketcan;
extern crate tokio;
#[macro_use]
extern crate log;
extern crate env_logger;

extern crate serde;
#[macro_use]
extern crate serde_derive;
extern crate serde_yaml;

#[macro_use]
extern crate lazy_static;

mod types;
mod canstream;
mod isotp;
mod definitions;

use tokio::prelude::*;
use std::env::args;
use futures::sync::mpsc;
use tokio::timer::Interval;
use std::time::{Duration, Instant};
use std::sync::Mutex;
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::vec::Vec;

use types::Value;

fn packet_print(id: u16, bytes: &[u8]) -> String {
    let mut ret: String = String::new();

    ret.push_str(&format!(
        "{:03X}/{:03X} (len={:2})",
        id - 8,
        id,
        bytes.len()
    ));

    for b in bytes {
        ret.push_str(&format!(" {:02X}", b));
    }

    ret
}

fn main() {
    env_logger::init();

    let raw_defs = definitions::RawECUEntry::get_all();

    info!("Read {} raw ecu entries", raw_defs.len());

    let world = futures::future::lazy(move || {
        let args: Vec<String> = args().collect();
        let iface = if args.len() > 1 {
            args[1].clone()
        } else {
            "vcan0".into()
        };
        info!("Connecting Canopied to {}", iface);
        let (sink, stream) = isotp::ISOTP::new(&iface).split();
        let (sender, receiver) = mpsc::channel(100);

        trace!("Spawning ISOTP forwarder");

        tokio::spawn(
            receiver
                .map(|f| {
                    trace!("Forwarding into ISOTP {:?}", f);
                    f
                })
                .forward(sink.sink_map_err(|_| ()))
                .map(|_| ()),
        );

        trace!("Preparing an Interval to run the requests");
        let always_defs = futures::stream::iter_ok(raw_defs.iter().cycle());
        let every_100ms = Interval::new(Instant::now(), Duration::from_millis(100));
        let zipped = every_100ms.zip(always_defs);

        tokio::spawn(zipped.map_err(|_| ()).for_each(move |(_, def)| {
            let req = def.request();
            let dest = req.sendto();
            let packet = req.packet().to_vec();
            let targ = sender.clone();
            tokio::spawn(futures::future::lazy(move || {
                targ.send((dest, packet)).wait().map(|_| ()).map_err(|_| ())
            }))
        }));

        let mut current: Mutex<HashMap<String, Value>> = Mutex::new(HashMap::new());

        tokio::spawn(
            stream
                .map_err(|e| error!("error = {:?}", e))
                .for_each(move |frame| {
                    trace!(
                        "Received ISO-TP packet: {}",
                        packet_print(frame.0, &frame.1)
                    );
                    let ecu = definitions::RawECUEntry::ecu_for(frame.0, &frame.1);
                    if let Some(ecu) = ecu {
                        if let Some(decoded) = ecu.decode(&frame.1) {
                            trace!("Decoded: {:?}", decoded);
                            let curmap = current.get_mut().unwrap();
                            for entry in decoded {
                                let name = entry.get_name().to_owned();
                                let value = entry.get_value().unwrap();
                                match curmap.entry(name.clone()) {
                                    Entry::Occupied(o) => if value != *o.get() {
                                        info!("{} changed {:?} -> {:?}", name, o.get(), value);
                                    },
                                    Entry::Vacant(_) => {
                                        info!("{} changed None -> {:?}", name, value);
                                    }
                                };
                                curmap.insert(name, value);
                            }
                        }
                    }

                    Ok(())
                }),
        )
    });
    tokio::run(world);
}