Mercurial > templog
view rust/src/sensor.rs @ 617:87a78343140e rust
Duration::sum works now
author | Matt Johnston <matt@ucc.asn.au> |
---|---|
date | Tue, 21 Mar 2017 19:13:49 +0800 |
parents | f3e39e2107fd |
children | c57821a60e51 |
line wrap: on
line source
extern crate tokio_core; extern crate futures; extern crate futures_cpupool; extern crate regex; 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 std::str::FromStr; 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=TemplogError>>; } #[derive(Clone)] pub struct OneWireSensor { config: Config, } impl OneWireSensor { pub fn new(config: &Config) -> Self { OneWireSensor { config: 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, TemplogError> { lazy_static! { // multiline static ref THERM_RE: regex::Regex = regex::Regex::new("(?m).* YES\n.*t=(.*)\n").unwrap(); } let mut path = PathBuf::from(&self.config.SENSOR_BASE_DIR); path.push(n); path.push("w1_slave"); let mut s = String::new(); File::open(path)?.read_to_string(&mut s)?; let caps = THERM_RE.captures(&s).ok_or_else(|| { TemplogError::new("Bad sensor contents match") })?; let v = caps.get(1).ok_or_else(|| { TemplogError::new("Bad field contents match") })?.as_str(); Ok(f32::from_str(v)?) } fn sensor_names(&self) -> Result<Vec<String>, TemplogError> { // TODO: needs to handle multiple busses. 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=TemplogError>> { 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()); let i = Interval::new(dur, handle).unwrap() .map_err(|e| TemplogError::new_io("Interval failed. Should not happen", e)) .and_then(move |()| { let a = s.clone(); pool.spawn_fn(move || Ok(a.step())) }); consume_errors(i) } } #[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, TemplogError> { 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=TemplogError>> { let dur = Duration::new(self.config.SENSOR_SLEEP,0); let i = Interval::new(dur, handle).unwrap() .map_err(|e| TemplogError::new_io("Interval failed. Should not happen", e)) .and_then(move |()| { Ok(Self::test_step()) }); consume_errors(i) } }