Mercurial > templog

comparison rust/src/sensor.rs @ 603:b45b8b4cf0f5 rust

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
get rid of lazy_static, config is passed around better use of threadpool for sensors readings are no longer options
author Matt Johnston <matt@ucc.asn.au>
date Thu, 16 Feb 2017 23:19:12 +0800
parents 8c21df3711e2
children 278f1002b5c7
comparison
equal deleted inserted replaced
602:613f114feb4b 603:b45b8b4cf0f5
6 use std::io; 6 use std::io;
7 use std::fs::File; 7 use std::fs::File;
8 use std::io::{Read,BufReader,BufRead}; 8 use std::io::{Read,BufReader,BufRead};
9 use std::path::PathBuf; 9 use std::path::PathBuf;
10 use std::error::Error; 10 use std::error::Error;
11 use std::sync::Arc;
11 12
12 use tokio_core::reactor::Interval; 13 use tokio_core::reactor::Interval;
13 use tokio_core::reactor::Handle; 14 use tokio_core::reactor::Handle;
14 use futures::Stream; 15 use futures::Stream;
15 use types::*; 16 use types::*;
17 use ::Config;
16 18
17 use ::rigid_config; 19
18 20
19 pub trait Sensor { 21 pub trait Sensor {
20 fn stream(handle: &Handle) 22 fn stream(&self, handle: &Handle)
21 -> Box<Stream<Item=Readings, Error=io::Error>>; 23 -> Box<Stream<Item=Readings, Error=io::Error>>;
22 } 24 }
23 25
26 #[derive(Clone)]
24 pub struct OneWireSensor { 27 pub struct OneWireSensor {
28 config: Arc<Config>,
25 } 29 }
26 30
27 impl OneWireSensor { 31 impl OneWireSensor {
28 fn new() -> OneWireSensor { 32 pub fn new(config: &Config) -> Self {
29 OneWireSensor { 33 OneWireSensor {
34 config: Arc::new(config.clone()),
30 } 35 }
31 } 36 }
32 37
33 fn step() -> Readings { 38 fn step(&self) -> Readings {
34 let mut r = Readings::new(); 39 let mut r = Readings::new();
35 40
41 if let Ok(names) = self.sensor_names() {
42 for n in &names {
43 match self.read_sensor(n) {
44 Ok(s) => r.add(n, s),
45 Err(e) => debug!("Error reading sensors {}: {}", n, e)
46 }
47 }
48 }
36 49
37 r.add("ambient", Some(31.2));
38 r.add("wort_todo", Some(8.0));
39 debug!("sensor step {:?}", r); 50 debug!("sensor step {:?}", r);
40 r 51 r
41 } 52 }
42 53
54 fn read_sensor(&self, n: &str) -> Result<f32, Box<Error>> {
55 let mut path = PathBuf::from(&self.config.SENSOR_BASE_DIR);
56 path.push(n);
57 path.push("w1_slave");
58 let f = BufReader::new(File::open(path)?);
59 Ok(3.4)
60 }
61
43 fn sensor_names(&self) -> Result<Vec<String>, Box<Error>> { 62 fn sensor_names(&self) -> Result<Vec<String>, Box<Error>> {
44 let mut path = PathBuf::from(&rigid_config.SENSOR_BASE_DIR); 63 let mut path = PathBuf::from(&self.config.SENSOR_BASE_DIR);
45 path.push("w1_master_slaves"); 64 path.push("w1_master_slaves");
46 65
47 let f = BufReader::new(File::open(path)?); 66 let f = BufReader::new(File::open(path)?);
48 let s = f.lines().collect::<Result<Vec<String>, io::Error>>()?; 67 let s = f.lines().collect::<Result<Vec<String>, io::Error>>()?;
49 Ok(s) 68 Ok(s)
50 } 69 }
51 } 70 }
52 71
53 // does this need to be static? 72 impl Sensor for OneWireSensor {
54 lazy_static! {
55 static ref thread_pool : futures_cpupool::CpuPool = futures_cpupool::CpuPool::new(3); // TODO: how many?
56 }
57 73
58 impl Sensor for OneWireSensor { 74 fn stream(&self, handle: &Handle) -> Box<Stream<Item=Readings, Error=io::Error>> {
59 fn stream(handle: &Handle) 75 let pool = futures_cpupool::CpuPool::new(4); // TODO: how many?
60 -> Box<Stream<Item=Readings, Error=io::Error>> {
61 let mut s = OneWireSensor::new();
62 76
63 let dur = Duration::new(rigid_config.SENSOR_SLEEP,0); 77 let dur = Duration::new(self.config.SENSOR_SLEEP,0);
78 let s = Arc::new(self.clone());
64 Interval::new(dur, handle).unwrap() 79 Interval::new(dur, handle).unwrap()
65 .and_then(|()| { 80 .and_then(move |()| {
66 thread_pool.spawn_fn(|| Ok(Self::step())) 81 let a = s.clone();
67 }).boxed() 82 pool.spawn_fn(move || Ok(a.step()))
83 })
84 .boxed()
68 } 85 }
69 } 86 }
70 87
88 #[derive(Clone)]
71 pub struct TestSensor { 89 pub struct TestSensor {
90 config: Config,
72 } 91 }
73 92
74 impl TestSensor { 93 impl TestSensor {
75 pub fn new() -> Self { 94 pub fn new(config: &Config) -> Self {
76 TestSensor {} 95 TestSensor {
96 config: config.clone(),
97 }
77 } 98 }
78 99
79 fn step(&mut self) -> Readings { 100 fn test_step() -> Readings {
80 let mut r = Readings::new(); 101 let mut r = Readings::new();
81 r.add("ambient", Some(31.2)); 102 r.add("ambient", 31.2);
82 r.add("wort", Some(Self::try_read("test_wort.txt").unwrap_or_else(|_| 18.0))); 103 r.add("wort", Self::try_read("test_wort.txt").unwrap_or_else(|_| 18.0));
83 r.add("fridge", Some(Self::try_read("test_fridge.txt").unwrap_or_else(|_| 20.0))); 104 r.add("fridge", Self::try_read("test_fridge.txt").unwrap_or_else(|_| 20.0));
84 r 105 r
85 } 106 }
86 107
87 fn try_read(filename: &str) -> Result<f32, Box<Error>> { 108 fn try_read(filename: &str) -> Result<f32, Box<Error>> {
88 let mut s = String::new(); 109 let mut s = String::new();
90 Ok(s.trim().parse::<f32>()?) 111 Ok(s.trim().parse::<f32>()?)
91 } 112 }
92 } 113 }
93 114
94 impl Sensor for TestSensor { 115 impl Sensor for TestSensor {
95 fn stream(handle: &Handle) 116 fn stream(&self, handle: &Handle)
96 -> Box<Stream<Item=Readings, Error=io::Error>> { 117 -> Box<Stream<Item=Readings, Error=io::Error>> {
97 let mut s = TestSensor::new();
98 118
99 let dur = Duration::new(rigid_config.SENSOR_SLEEP,0); 119 let dur = Duration::new(self.config.SENSOR_SLEEP,0);
100 Interval::new(dur, handle).unwrap().map(move |()| { 120 Interval::new(dur, handle).unwrap()
101 s.step() 121 .and_then(move |()| {
102 }).boxed() 122 Ok(Self::test_step())
123 }).boxed()
103 } 124 }
104 } 125 }
105 126