Mercurial > templog
view rust/src/sensor.rs @ 603:b45b8b4cf0f5 rust
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 |
line wrap: on
line source
extern crate tokio_core; extern crate futures; extern crate futures_cpupool; use std::time::Duration; use std::io; use std::fs::File; use std::io::{Read,BufReader,BufRead}; use std::path::PathBuf; use std::error::Error; use std::sync::Arc; use tokio_core::reactor::Interval; use tokio_core::reactor::Handle; use futures::Stream; use types::*; use ::Config; pub trait Sensor { fn stream(&self, handle: &Handle) -> Box<Stream<Item=Readings, Error=io::Error>>; } #[derive(Clone)] pub struct OneWireSensor { config: Arc<Config>, } impl OneWireSensor { pub fn new(config: &Config) -> Self { OneWireSensor { config: Arc::new(config.clone()), } } fn step(&self) -> Readings { let mut r = Readings::new(); if let Ok(names) = self.sensor_names() { for n in &names { match self.read_sensor(n) { Ok(s) => r.add(n, s), Err(e) => debug!("Error reading sensors {}: {}", n, e) } } } debug!("sensor step {:?}", r); r } fn read_sensor(&self, n: &str) -> Result<f32, Box<Error>> { let mut path = PathBuf::from(&self.config.SENSOR_BASE_DIR); path.push(n); path.push("w1_slave"); let f = BufReader::new(File::open(path)?); Ok(3.4) } fn sensor_names(&self) -> Result<Vec<String>, Box<Error>> { let mut path = PathBuf::from(&self.config.SENSOR_BASE_DIR); path.push("w1_master_slaves"); let f = BufReader::new(File::open(path)?); let s = f.lines().collect::<Result<Vec<String>, io::Error>>()?; Ok(s) } } impl Sensor for OneWireSensor { fn stream(&self, handle: &Handle) -> Box<Stream<Item=Readings, Error=io::Error>> { let pool = futures_cpupool::CpuPool::new(4); // TODO: how many? let dur = Duration::new(self.config.SENSOR_SLEEP,0); let s = Arc::new(self.clone()); Interval::new(dur, handle).unwrap() .and_then(move |()| { let a = s.clone(); pool.spawn_fn(move || Ok(a.step())) }) .boxed() } } #[derive(Clone)] pub struct TestSensor { config: Config, } impl TestSensor { pub fn new(config: &Config) -> Self { TestSensor { config: config.clone(), } } fn test_step() -> Readings { let mut r = Readings::new(); r.add("ambient", 31.2); r.add("wort", Self::try_read("test_wort.txt").unwrap_or_else(|_| 18.0)); r.add("fridge", Self::try_read("test_fridge.txt").unwrap_or_else(|_| 20.0)); r } fn try_read(filename: &str) -> Result<f32, Box<Error>> { let mut s = String::new(); File::open(filename)?.read_to_string(&mut s)?; Ok(s.trim().parse::<f32>()?) } } impl Sensor for TestSensor { fn stream(&self, handle: &Handle) -> Box<Stream<Item=Readings, Error=io::Error>> { let dur = Duration::new(self.config.SENSOR_SLEEP,0); Interval::new(dur, handle).unwrap() .and_then(move |()| { Ok(Self::test_step()) }).boxed() } }