summaryrefslogtreecommitdiff
path: root/src/main.rs
blob: 8daac6c2b2a216ed4b58b2256f4406b58cba9518 (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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
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;

#[macro_use]
extern crate nom;

mod canstream;
mod definitions;
mod exprast;
mod exprparser;
mod isotp;
mod qexpr;
mod qvalue;
mod types;

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

use definitions::CadencedECUs;
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();
    let units = definitions::Units::get();
    let derived = definitions::Derived::get();

    info!("Read {} raw ecu entries", raw_defs.len());
    info!("Read {} unit definitions", units.count());
    info!("Read {} derived definitions", derived.count());

    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(CadencedECUs::new());
        let every_50ms = Interval::new(Instant::now(), Duration::from_millis(50));
        let zipped = every_50ms.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();
                                if !name.contains("speed") {
                                    continue;
                                }
                                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);
                            }
                        }
                    } else {
                        warn!(
                            "Cannot locate Raw RCU Entry for {}",
                            packet_print(frame.0, &frame.1)
                        );
                    }

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