improvements

Cargo.toml

@@ -10,22 +10,15 @@ release-logs = ["tracing/max_level_info"]
 [dependencies]
 serde = "1.0"
 serde_derive = "1.0"
-influxdb2 = { version = "0.3.3", features = [
+tokio = { version = "1", features = ["rt-multi-thread", "io-util"] }
-    "rustls",
-], default-features = false }
-tokio = { version = "1", features = ["rt", "io-util"] }
 clap = { version = "4.0", features = ["derive"] }
-log = "0.4"
 figment = { version = "0.10", features = ["toml"] }
-gethostname = "0.3"
 futures = "0.3.24"
 async-trait = "0.1.58"
 tokio-stream = { version = "0.1.11", features = ["sync"] }
-bitflags = "1.3.2"
 byteorder = "1.4.3"
 tokio-serial = "5.4.4"
 bytes = "1.2.1"
-chrono = "0.4.23"
 libc = "0.2.137"
 async-stream = "0.3.6"
 itertools = "0.14.0"
@@ -35,7 +28,14 @@ serde_json = "1.0.146"
 backoff = { version = "0.4.0", features = ["tokio"] }
 serde_repr = "0.1.20"
 tracing = "0.1.44"
-tracing-subscriber = { version = "0.3.22", features = ["fmt"] }
+tracing-subscriber = { version = "0.3.22", features = ["fmt", "ansi", "time", "env-filter"] }
+serialport = "4.8.1"
+gethostname = "1.1.0"
+bitflags = "2.10.0"
+influxdb2 = { version = "0.5.2" }
+chrono = "0.4.43"
+serde_with = "3.16.1"
+ctrlc = "3.5.1"
 
 [dependencies.chrony-candm]
 git = "https://github.com/aws/chrony-candm"

Dockerfile

@@ -1,6 +1,6 @@
-FROM rust:slim as builder
+FROM rust:slim AS builder
 
-RUN apt-get update && apt-get install -y libssl-dev pkg-config
+RUN apt-get update && apt-get install -y libudev-dev pkg-config
 # build deps only first for build cache
 WORKDIR /usr/src
 RUN USER=root cargo new chimemon
@@ -11,6 +11,7 @@ COPY . .
 RUN cargo build --release
 
 FROM debian:bullseye-slim
+RUN apt-get update && apt-get install -y libudev
 WORKDIR /app
 COPY --from=builder /usr/src/chimemon/target/release/chimemon chimemon
 CMD ["/app/chimemon"]

config.toml.dist

@@ -24,6 +24,13 @@
     status_interval = 10
     measurement = "uccm_gpsdo"
     
+    [sources.prs10]
+    enabled = true
+    port = "/dev/ttyUSB0"
+    status_interval = 10
+    stats_interval = 10
+    measurement = "prs10_gpsdo"
+
 
 [targets]
     [targets.chrony]

src/chrony_refclock.rs

@@ -1,74 +0,0 @@
-use crate::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, ChronySockConfig};
-use async_trait::async_trait;
-use libc::{c_double, c_int, timeval};
-use std::mem;
-use std::os::unix::net::UnixDatagram;
-use std::path::PathBuf;
-use tracing::debug;
-
-const CHRONY_MAGIC: c_int = 0x534f434b;
-
-pub struct ChronySockServer {
-    sock_path: PathBuf,
-}
-
-#[repr(C)]
-#[derive(Debug)]
-pub struct ChronyTimeReport {
-    tv: timeval,
-    offset: c_double,
-    pulse: c_int,
-    leap: c_int,
-    _pad: c_int,
-    magic: c_int,
-}
-
-impl ChronySockServer {
-    pub fn new(config: ChronySockConfig) -> Self {
-        ChronySockServer {
-            sock_path: config.sock.into(),
-        }
-    }
-}
-
-#[async_trait]
-impl ChimemonTarget for ChronySockServer {
-    async fn run(mut self, mut chan: ChimemonTargetChannel) {
-        loop {
-            let msg = chan.recv().await.unwrap();
-            match msg {
-                ChimemonMessage::TimeReport(tr) => {
-                    if tr.valid {
-                        {
-                            let frame = ChronyTimeReport {
-                                tv: timeval {
-                                    tv_sec: TryInto::<libc::time_t>::try_into(
-                                        tr.system_time.timestamp(),
-                                    )
-                                    .unwrap(),
-                                    tv_usec: tr.system_time.timestamp_subsec_micros()
-                                        as libc::suseconds_t,
-                                },
-                                offset: tr.offset.num_nanoseconds().unwrap() as f64 / 1e9,
-                                leap: if tr.leap_flag { 1 } else { 0 },
-                                pulse: 0,
-                                _pad: 0,
-                                magic: CHRONY_MAGIC,
-                            };
-                            let bs = unsafe {
-                                std::slice::from_raw_parts(
-                                    (&frame as *const ChronyTimeReport) as *const u8,
-                                    mem::size_of::<ChronyTimeReport>(),
-                                )
-                            };
-                            debug!("Sending to chrony sock {:#?}", frame);
-                            let sock = UnixDatagram::unbound().unwrap();
-                            sock.send_to(bs, &self.sock_path).unwrap();
-                        }
-                    }
-                }
-                _ => continue,
-            }
-        }
-    }
-}

src/config.rs

@@ -0,0 +1,214 @@
+use figment::{Provider, providers::Serialized, util::map, value::Map};
+use gethostname::gethostname;
+use serde_derive::{Deserialize, Serialize};
+use serde_with::{DurationSeconds, serde_as};
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct InfluxConfig {
+    pub enabled: bool,
+    pub timeout: std::time::Duration,
+    pub url: String,
+    pub org: String,
+    pub bucket: String,
+    pub token: String,
+    pub tags: Map<String, String>,
+}
+
+impl Default for InfluxConfig {
+    fn default() -> Self {
+        let host = gethostname().into_string().unwrap();
+        InfluxConfig {
+            enabled: false,
+            timeout: std::time::Duration::from_secs(10),
+            url: "http://localhost:8086".into(),
+            org: "default".into(),
+            bucket: "default".into(),
+            token: "".into(),
+            tags: map! { "host".into() => host },
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct ChronyConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub tracking_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub sources_interval: std::time::Duration,
+    pub measurement_prefix: String,
+    pub tracking_measurement: String,
+    pub sources_measurement: String,
+    pub host: String,
+}
+
+impl Default for ChronyConfig {
+    fn default() -> Self {
+        ChronyConfig {
+            enabled: false,
+            timeout: std::time::Duration::from_secs(5),
+            tracking_interval: std::time::Duration::from_secs(60),
+            sources_interval: std::time::Duration::from_secs(300),
+            measurement_prefix: "chrony.".into(),
+            tracking_measurement: "tracking".into(),
+            sources_measurement: "sources".into(),
+            host: "127.0.0.1:323".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct ChronySockConfig {
+    pub enabled: bool,
+    pub sock: String,
+}
+
+impl Default for ChronySockConfig {
+    fn default() -> Self {
+        ChronySockConfig {
+            enabled: false,
+            sock: "".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct HwmonSensorConfig {
+    pub device: String,
+    pub sensor: String,
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct HwmonConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub interval: std::time::Duration,
+    pub measurement: String,
+    pub sensors: Map<String, HwmonSensorConfig>,
+}
+
+impl Default for HwmonConfig {
+    fn default() -> Self {
+        HwmonConfig {
+            enabled: false,
+            interval: std::time::Duration::from_secs(60),
+            measurement: "hwmon".into(),
+            sensors: map! {},
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct GpsdConfig {
+    pub enabled: bool,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub interval: std::time::Duration,
+    pub host: String,
+}
+
+impl Default for GpsdConfig {
+    fn default() -> Self {
+        GpsdConfig {
+            enabled: false,
+            interval: std::time::Duration::from_secs(60),
+            host: "localhost:2947".into(),
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(default)]
+pub struct Prs10Config {
+    pub enabled: bool,
+    pub port: String,
+    pub baud: u32,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub status_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub stats_interval: std::time::Duration,
+}
+
+impl Default for Prs10Config {
+    fn default() -> Self {
+        Prs10Config {
+            enabled: false,
+            port: "/dev/ttyS0".into(),
+            baud: 9600,
+            timeout: std::time::Duration::from_secs(1),
+            status_interval: std::time::Duration::from_secs(10),
+            stats_interval: std::time::Duration::from_secs(30),
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct UCCMConfig {
+    pub enabled: bool,
+    pub port: String,
+    pub baud: u32,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub status_interval: std::time::Duration,
+    #[serde_as(as = "DurationSeconds<u64>")]
+    pub timeout: std::time::Duration,
+    pub measurement: String,
+}
+
+impl Default for UCCMConfig {
+    fn default() -> Self {
+        UCCMConfig {
+            enabled: false,
+            port: "/dev/ttyS0".into(),
+            baud: 57600,
+            status_interval: std::time::Duration::from_secs(10),
+            timeout: std::time::Duration::from_secs(1),
+            measurement: "uccm_gpsdo".into(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum SourceConfig {
+    Chrony(ChronyConfig),
+    Hwmon(HwmonConfig),
+    Uccm(UCCMConfig),
+    Gpsd(GpsdConfig),
+    Prs10(Prs10Config),
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum TargetConfig {
+    ChronySock(ChronySockConfig),
+    Influxdb(InfluxConfig),
+}
+
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub struct Config {
+    pub sources: Map<String, SourceConfig>,
+    pub targets: Map<String, TargetConfig>,
+}
+
+impl Provider for Config {
+    fn metadata(&self) -> figment::Metadata {
+        figment::Metadata::named("Default config")
+    }
+    fn data(&self) -> Result<Map<figment::Profile, figment::value::Dict>, figment::Error> {
+        Serialized::defaults(Config::default()).data()
+    }
+}

src/hwmon.rs

@@ -1,162 +0,0 @@
-use crate::{
-    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
-    SourceStatus,
-};
-use async_trait::async_trait;
-use futures::{StreamExt, stream};
-use influxdb2::models::DataPoint;
-use std::{
-    fs::File,
-    io::Read,
-    path::PathBuf,
-    sync::Arc,
-    time::{Duration, SystemTime, UNIX_EPOCH},
-};
-use tracing::{Instrument, debug, error, info, info_span, warn};
-
-pub struct HwmonSource {
-    config: Config,
-    sensors: Vec<Arc<HwmonSensor>>,
-}
-
-#[derive(Debug, Clone)]
-struct HwmonSensor {
-    name: String,
-    value_path: PathBuf,
-    device: String,
-    sensor: String,
-    label: Option<String>,
-    tags: Arc<Vec<(String, String)>>,
-}
-
-impl HwmonSensor {
-    fn new(name: &str, device: &str, sensor: &str) -> Self {
-        let value_path = PathBuf::from(HWMON_ROOT)
-            .join(device)
-            .join(sensor.to_owned() + "_input");
-        let label_path_raw = PathBuf::from(HWMON_ROOT)
-            .join(device)
-            .join(sensor.to_owned() + "_label");
-        let label = if label_path_raw.is_file() {
-            let mut f =
-                File::open(&label_path_raw).expect(&format!("Unable to open `{label_path_raw:?}`"));
-            let mut label = String::new();
-            f.read_to_string(&mut label)
-                .expect(&format!("Unable to read from `{label_path_raw:?}"));
-            Some(label.trim().to_owned())
-        } else {
-            None
-        };
-        let mut tags_vec = vec![
-            ("name".to_owned(), name.to_owned()),
-            ("device".to_owned(), device.to_owned()),
-            ("sensor".to_owned(), sensor.to_owned()),
-        ];
-        if let Some(label) = &label {
-            tags_vec.push(("label".to_owned(), label.clone()))
-        }
-        Self {
-            value_path,
-            label,
-            device: device.to_owned(),
-            sensor: sensor.to_owned(),
-            name: name.to_owned(),
-            tags: Arc::new(tags_vec),
-        }
-    }
-}
-
-#[derive(Debug)]
-struct HwmonReport {
-    values: Vec<(Arc<HwmonSensor>, f64)>,
-}
-
-impl SourceReportDetails for HwmonReport {
-    fn is_healthy(&self) -> bool {
-        //self.alarms.iter().any(|(_sensor, alarm)| *alarm)
-        true
-    }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
-        let mut metrics = Vec::new();
-        for (sensor, value) in &self.values {
-            metrics.push(SourceMetric::new_float(
-                "hwmon_value",
-                *value,
-                sensor.tags.clone(),
-            ))
-        }
-        // for (sensor, alarm) in &self.alarms {
-        //     metrics.push(SourceMetric::new_bool(
-        //         "hwmon_alarm",
-        //         *alarm,
-        //         sensor.tags.clone(),
-        //     ))
-        // }
-
-        metrics
-    }
-}
-
-const HWMON_ROOT: &str = "/sys/class/hwmon";
-
-impl HwmonSource {
-    pub fn new(config: Config) -> Self {
-        let sensors = config
-            .sources
-            .hwmon
-            .sensors
-            .iter()
-            .map(|(k, v)| Arc::new(HwmonSensor::new(k, &v.name, &v.sensor)))
-            .collect();
-
-        HwmonSource { config, sensors }
-    }
-
-    async fn get_raw_value(sensor: &HwmonSensor) -> Result<String, std::io::Error> {
-        tokio::fs::read_to_string(&sensor.value_path).await
-    }
-}
-
-#[async_trait]
-impl ChimemonSource for HwmonSource {
-    async fn run(self, chan: ChimemonSourceChannel) {
-        info!("hwmon task started");
-        let mut interval =
-            tokio::time::interval(Duration::from_secs(self.config.sources.hwmon.interval));
-        loop {
-            interval.tick().await;
-            let mut values = Vec::new();
-            for s in &self.sensors {
-                if let Ok(sensor_val) = HwmonSource::get_raw_value(s).await {
-                    debug!(
-                        "hwmon {} raw value {}",
-                        s.value_path.to_string_lossy(),
-                        sensor_val
-                    );
-                    if let Ok(parsed) = sensor_val.trim().parse::<f64>() {
-                        values.push((s.clone(), parsed));
-                    } else {
-                        error!(
-                            "Unable to parse sensor value {sensor_val} at {}",
-                            s.value_path.to_string_lossy()
-                        );
-                    }
-                } else {
-                    error!("Unable to get hwmon sensor value");
-                }
-            }
-            let report = SourceReport {
-                name: "hwmon".to_owned(),
-                status: SourceStatus::Healthy,
-                details: Arc::new(HwmonReport { values }),
-            };
-            info!("Writing hwmon data");
-            match chan.send(report.into()) {
-                Ok(_) => {}
-                Err(e) => {
-                    warn!("Unable to send to message channel ({e})")
-                }
-            }
-        }
-    }
-}

src/lib.rs

@@ -1,131 +1,22 @@
-pub mod chrony;
+pub mod config;
-pub mod chrony_refclock;
+pub mod sources;
-pub mod gpsd;
+pub mod targets;
-pub mod hwmon;
+
-pub mod uccm;
+#[macro_export]
+macro_rules! fatal {
+    ($($arg:tt)*) => {{
+        tracing::error!($($arg)*);
+        std::process::exit(-1);
+    }};
+}
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
-use figment::{
-    Figment,
-    providers::{Format, Serialized, Toml},
-    util::map,
-    value::Map,
-};
-use gethostname::gethostname;
-use influxdb2::models::DataPoint;
-use serde_derive::{Deserialize, Serialize};
-use tokio::sync::broadcast::*;
 
-use std::{fmt::Debug, net::SocketAddr, path::Path, sync::Arc};
+use tokio::sync::broadcast;
+use tokio_util::sync::CancellationToken;
 
-#[derive(Serialize, Deserialize, Clone)]
+use std::{fmt::Debug, sync::Arc};
-pub struct InfluxConfig {
-    pub enabled: bool,
-    pub url: String,
-    pub org: String,
-    pub bucket: String,
-    pub token: String,
-    pub tags: Map<String, String>,
-}
-
-impl Default for InfluxConfig {
-    fn default() -> Self {
-        let host = gethostname().into_string().unwrap();
-        InfluxConfig {
-            enabled: false,
-            url: "http://localhost:8086".into(),
-            org: "default".into(),
-            bucket: "default".into(),
-            token: "".into(),
-            tags: map! { "host".into() => host },
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct ChronyConfig {
-    pub enabled: bool,
-    pub timeout: u64,
-    pub tracking_interval: u64,
-    pub sources_interval: u64,
-    pub measurement_prefix: String,
-    pub tracking_measurement: String,
-    pub sources_measurement: String,
-    pub host: String,
-}
-
-impl Default for ChronyConfig {
-    fn default() -> Self {
-        ChronyConfig {
-            enabled: false,
-            timeout: 5,
-            tracking_interval: 60,
-            sources_interval: 300,
-            measurement_prefix: "chrony.".into(),
-            tracking_measurement: "tracking".into(),
-            sources_measurement: "sources".into(),
-            host: "127.0.0.1:323".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct ChronySockConfig {
-    pub enabled: bool,
-    pub sock: String,
-}
-
-impl Default for ChronySockConfig {
-    fn default() -> Self {
-        ChronySockConfig {
-            enabled: false,
-            sock: "".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct HwmonSensorConfig {
-    pub name: String,
-    pub sensor: String,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct HwmonConfig {
-    pub enabled: bool,
-    pub interval: u64,
-    pub measurement: String,
-    pub sensors: Map<String, HwmonSensorConfig>,
-}
-
-impl Default for HwmonConfig {
-    fn default() -> Self {
-        HwmonConfig {
-            enabled: false,
-            interval: 60,
-            measurement: "hwmon".into(),
-            sensors: map! {},
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct GpsdConfig {
-    pub enabled: bool,
-    pub interval: u64,
-    pub host: String,
-}
-
-impl Default for GpsdConfig {
-    fn default() -> Self {
-        GpsdConfig {
-            enabled: false,
-            interval: 60,
-            host: "localhost:2947".into(),
-        }
-    }
-}
 
 #[derive(Clone, Debug)]
 pub struct TimeReport {
@@ -145,48 +36,53 @@ pub enum SourceStatus {
     Unknown,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Copy, Clone, Debug)]
 pub enum MetricValue {
     Int(i64),
     Float(f64),
     Bool(bool),
 }
 
+type MetricTag = (&'static str, String);
+type MetricTags = Vec<MetricTag>;
+
 #[derive(Clone, Debug)]
 pub struct SourceMetric {
-    name: String,
+    name: &'static str,
     value: MetricValue,
-    tags: Arc<Vec<(String, String)>>,
 }
 
 impl SourceMetric {
-    pub fn new_int(name: &str, value: i64, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_int(name: &'static str, value: i64) -> Self {
         Self {
-            name: name.to_owned(),
+            name: name,
             value: MetricValue::Int(value),
-            tags,
         }
     }
 
-    pub fn new_float(name: &str, value: f64, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_float(name: &'static str, value: f64) -> Self {
         Self {
-            name: name.to_owned(),
+            name: name,
             value: MetricValue::Float(value),
-            tags,
         }
     }
 
-    pub fn new_bool(name: &str, value: bool, tags: Arc<Vec<(String, String)>>) -> Self {
+    pub fn new_bool(name: &'static str, value: bool) -> Self {
         Self {
-            name: name.to_owned(),
+            name: name,
             value: MetricValue::Bool(value),
-            tags,
         }
     }
 }
 
+#[derive(Debug)]
+pub struct SourceMetricSet {
+    metrics: Vec<SourceMetric>,
+    tags: Arc<MetricTags>,
+}
+
 pub trait SourceReportDetails: Debug + Send + Sync {
-    fn to_metrics(&self) -> Vec<SourceMetric>;
+    fn to_metrics(&self) -> Vec<SourceMetricSet>;
     fn is_healthy(&self) -> bool;
 }
 
@@ -194,75 +90,15 @@ pub trait SourceReportDetails: Debug + Send + Sync {
 pub struct SourceReport {
     pub name: String,
     pub status: SourceStatus,
-    pub details: Arc<dyn SourceReportDetails>,
+    pub details: Arc<dyn SourceReportDetails + Send + Sync>,
-}
-
-#[derive(Serialize, Deserialize, Clone)]
-pub struct UCCMConfig {
-    pub enabled: bool,
-    pub port: String,
-    pub baud: u32,
-    pub status_interval: std::time::Duration,
-    pub timeout: std::time::Duration,
-    pub measurement: String,
-}
-
-impl Default for UCCMConfig {
-    fn default() -> Self {
-        UCCMConfig {
-            enabled: false,
-            port: "/dev/ttyS0".into(),
-            baud: 57600,
-            status_interval: std::time::Duration::from_secs(10),
-            timeout: std::time::Duration::from_secs(1),
-            measurement: "uccm_gpsdo".into(),
-        }
-    }
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct SourcesConfig {
-    pub chrony: ChronyConfig,
-    pub hwmon: HwmonConfig,
-    pub uccm: UCCMConfig,
-    pub gpsd: GpsdConfig,
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct TargetsConfig {
-    pub chrony: ChronySockConfig,
-}
-
-#[derive(Serialize, Deserialize, Clone, Default)]
-pub struct Config {
-    pub influxdb: InfluxConfig,
-    pub sources: SourcesConfig,
-    pub targets: TargetsConfig,
-}
-
-pub fn load_config(filename: &Path) -> Figment {
-    Figment::from(Serialized::defaults(Config::default())).merge(Toml::file(filename))
 }
 
 #[derive(Debug, Clone)]
 pub enum ChimemonMessage {
-    DataPoint(DataPoint),
-    DataPoints(Vec<DataPoint>),
     TimeReport(TimeReport),
     SourceReport(SourceReport),
 }
 
-impl From<DataPoint> for ChimemonMessage {
-    fn from(dp: DataPoint) -> Self {
-        ChimemonMessage::DataPoint(dp)
-    }
-}
-impl From<Vec<DataPoint>> for ChimemonMessage {
-    fn from(dps: Vec<DataPoint>) -> Self {
-        ChimemonMessage::DataPoints(dps)
-    }
-}
-
 impl From<TimeReport> for ChimemonMessage {
     fn from(tr: TimeReport) -> Self {
         ChimemonMessage::TimeReport(tr)
@@ -275,15 +111,21 @@ impl From<SourceReport> for ChimemonMessage {
     }
 }
 
-pub type ChimemonSourceChannel = Sender<ChimemonMessage>;
+pub type ChimemonSourceChannel = broadcast::Sender<ChimemonMessage>;
-pub type ChimemonTargetChannel = Receiver<ChimemonMessage>;
+pub type ChimemonTargetChannel = broadcast::Receiver<ChimemonMessage>;
 
 #[async_trait]
 pub trait ChimemonSource {
-    async fn run(self, chan: ChimemonSourceChannel);
+    type Config;
+    const TASK_NAME: &'static str;
+    fn new(name: &str, config: Self::Config) -> Self;
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken);
 }
 
 #[async_trait]
 pub trait ChimemonTarget {
-    async fn run(self, chan: ChimemonTargetChannel);
+    type Config;
+    const TASK_NAME: &'static str;
+    fn new(name: &str, config: Self::Config) -> Self;
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken);
 }

src/main.rs

@@ -1,15 +1,31 @@
-use clap::{Parser, ValueEnum};
+use std::sync::Arc;
-use futures::future::join_all;
-use std::path::Path;
-use tokio::sync::broadcast;
-use tracing::{Instrument, debug, error, info, info_span, warn};
-use tracing_subscriber;
 
-use crate::{chrony::*, chrony_refclock::ChronySockServer, hwmon::HwmonSource, uccm::UCCMMonitor};
+use async_trait::async_trait;
-use chimemon::{gpsd::GpsdSource, *};
+use clap::{Parser, ValueEnum};
+use figment::{
+    Figment,
+    providers::{Format, Toml},
+};
+use futures::future::join_all;
+use tokio::{select, sync::broadcast, task::JoinHandle};
+use tokio_util::sync::CancellationToken;
+use tracing::{Instrument, debug, error, info, info_span, warn};
+use tracing_subscriber::{self, EnvFilter, fmt::format::FmtSpan};
+
+use chimemon::{
+    config::{SourceConfig, TargetConfig},
+    targets::influx::InfluxTarget,
+    *,
+};
+use config::Config;
+use sources::{
+    chrony::ChronyClient, gpsd::GpsdSource, hwmon::HwmonSource, prs10::Prs10Monitor,
+    uccm::UCCMMonitor,
+};
+use targets::chrony_refclock::ChronySockServer;
 
 const PROGRAM_NAME: &str = "chimemon";
-const VERSION: &str = "0.0.1";
+const VERSION: &str = env!("CARGO_PKG_VERSION");
 
 #[derive(ValueEnum, Clone)]
 enum Level {
@@ -19,147 +35,211 @@ enum Level {
     Error,
 }
 
+impl From<Level> for tracing::Level {
+    fn from(level: Level) -> Self {
+        match level {
+            Level::Debug => tracing::Level::DEBUG,
+            Level::Info => tracing::Level::INFO,
+            Level::Warn => tracing::Level::WARN,
+            Level::Error => tracing::Level::ERROR,
+        }
+    }
+}
+
+impl Level {
+    fn level(self) -> tracing::Level {
+        self.into()
+    }
+}
+
 #[derive(Parser)]
 struct Args {
     /// TOML configuration to load
     #[arg(short, long, default_value_t = String::from("config.toml"))]
     config_file: String,
-    #[arg(value_enum, default_value_t = Level::Warn)]
+    #[arg(value_enum, default_value_t = Level::Info)]
     log_level: Level,
+    #[arg(short, long, default_value_t = false)]
+    echo_task: bool,
 }
 
-#[tokio::main(flavor = "current_thread")]
+fn run_source(
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    name: &str,
-    tracing_subscriber::fmt::init();
+    source: SourceConfig,
+    chan: ChimemonSourceChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    match source {
+        SourceConfig::Chrony(cfg) if cfg.enabled => {
+            spawn_source::<ChronyClient>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Gpsd(cfg) if cfg.enabled => {
+            spawn_source::<GpsdSource>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Hwmon(cfg) if cfg.enabled => {
+            spawn_source::<HwmonSource>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Prs10(cfg) if cfg.enabled => {
+            spawn_source::<Prs10Monitor>(name, cfg, chan, cancel)
+        }
+        SourceConfig::Uccm(cfg) if cfg.enabled => {
+            spawn_source::<UCCMMonitor>(name, cfg, chan, cancel)
+        }
+        _ => {
+            debug!("Disabled source {name} skipped");
+            None
+        }
+    }
+}
 
+fn run_target(
+    name: &str,
+    target: TargetConfig,
+    chan: ChimemonTargetChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    match target {
+        TargetConfig::ChronySock(cfg) if cfg.enabled => {
+            spawn_target::<ChronySockServer>(name, cfg, chan, cancel)
+        }
+        TargetConfig::Influxdb(cfg) if cfg.enabled => {
+            spawn_target::<InfluxTarget>(name, cfg, chan, cancel)
+        }
+        _ => {
+            debug!("Disabled target {name} skipped");
+            None
+        }
+    }
+}
+
+fn spawn_source<T: ChimemonSource + Send + Sync + 'static>(
+    name: &str,
+    config: T::Config,
+    chan: ChimemonSourceChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    let span = info_span!("source", task = name);
+    let s = T::new(name, config);
+    Some(tokio::spawn(s.run(chan, cancel).instrument(span)))
+}
+
+fn spawn_target<T: ChimemonTarget + Send + Sync + 'static>(
+    name: &str,
+    config: T::Config,
+    chan: ChimemonTargetChannel,
+    cancel: CancellationToken,
+) -> Option<JoinHandle<()>> {
+    let span = info_span!("target", task = name);
+    let t = T::new(name, config);
+    Some(tokio::spawn(async move {
+        t.run(chan, cancel).instrument(span).await
+    }))
+}
+
+struct EchoTarget {}
+struct EchoTargetConfig {}
+#[async_trait]
+impl ChimemonTarget for EchoTarget {
+    type Config = EchoTargetConfig;
+    const TASK_NAME: &'static str = "echo-task";
+
+    fn new(_name: &str, _config: Self::Config) -> Self {
+        EchoTarget {}
+    }
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Dummy receiver task started");
+        loop {
+            let msg = select! {
+                _ = cancel.cancelled() => {
+                    return
+                },
+                msg = chan.recv() => msg
+            };
+            match msg {
+                Ok(ChimemonMessage::SourceReport(report)) => {
+                    let metrics = report.details.to_metrics();
+                    info!("instance: {} metrics: {metrics:?}", report.name);
+                }
+                Ok(msg) => {
+                    info!("message: {msg:?}");
+                }
+                Err(e) => {
+                    warn!(error = ?e, "Error receiving message");
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args = Args::parse();
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::try_from_default_env()
+                .unwrap_or(EnvFilter::default().add_directive(args.log_level.level().into())),
+        )
+        // .event_format(format::Format::default().pretty())
+        .with_span_events(FmtSpan::CLOSE)
+        .init();
 
     info!("{PROGRAM_NAME} v{VERSION} starting...");
-    let fig = load_config(Path::new(&args.config_file));
+    let fig = Figment::new()
+        .merge(Config::default())
+        .merge(Toml::file(&args.config_file));
     debug!("{fig:?}");
     let config: Config = fig.extract()?;
 
     let mut tasks = Vec::new();
-    let (tx, _) = broadcast::channel(16);
+    let (sourcechan, _) = broadcast::channel(16);
-    let sourcechan: ChimemonSourceChannel = tx;
 
-    if config.influxdb.enabled {
+    let shutdown_token = CancellationToken::new();
-        info!(
-            "Connecting to influxdb {} org: {} using token",
-            &config.influxdb.url, &config.influxdb.org
-        );
-        let config = config.clone();
-        let influx = influxdb2::Client::new(
-            &config.influxdb.url,
-            &config.influxdb.org,
-            &config.influxdb.token,
-        );
 
-        let mut influxrx = sourcechan.subscribe();
+    for (name, target) in config.targets {
-
+        if let Some(task) = run_target(
-        tasks.push(tokio::spawn(
+            &name,
-            async move {
+            target,
-                let stream = async_stream::stream! {
+            sourcechan.subscribe(),
-                    while let Ok(msg) = influxrx.recv().await {
+            shutdown_token.clone(),
-                        match msg { ChimemonMessage::DataPoint(dp) => {
+        ) {
-                            yield dp
+            tasks.push(task)
-                        }, ChimemonMessage::DataPoints(dps) => {
+        }
-                            for p in dps {
-                                yield p
-                            }
-                        }, _ => {}
-                    }            }
-                };
-                influx.write(&config.influxdb.bucket, stream).await.unwrap();
-            }
-            .instrument(info_span!("influx-task")),
-        ));
     }
 
-    let chrony = if config.sources.chrony.enabled {
+    for (name, source) in config.sources {
-        Some(ChronyClient::new(config.to_owned()))
+        if let Some(task) = run_source(&name, source, sourcechan.clone(), shutdown_token.clone()) {
-    } else {
+            tasks.push(task)
-        None
+        }
-    };
-    if let Some(c) = chrony {
-        tasks.push(tokio::spawn(
-            c.run(sourcechan.clone())
-                .instrument(info_span!("chrony-task")),
-        ));
-    };
-
-    let hwmon = if config.sources.hwmon.enabled {
-        Some(HwmonSource::new(config.to_owned()))
-    } else {
-        None
-    };
-    if let Some(hwmon) = hwmon {
-        tasks.push(tokio::spawn(
-            hwmon
-                .run(sourcechan.clone())
-                .instrument(info_span!("hwmon-task")),
-        ));
-    };
-
-    let uccm = if config.sources.uccm.enabled {
-        Some(UCCMMonitor::new(config.to_owned()))
-    } else {
-        None
-    };
-    if let Some(uccm) = uccm {
-        tasks.push(tokio::spawn(
-            uccm.run(sourcechan.clone())
-                .instrument(info_span!("uccm-task")),
-        ));
-    };
-
-    let chrony_refclock = if config.targets.chrony.enabled {
-        Some(ChronySockServer::new(config.targets.chrony.to_owned()))
-    } else {
-        None
-    };
-    if let Some(chrony_refclock) = chrony_refclock {
-        tasks.push(tokio::spawn(
-            chrony_refclock
-                .run(sourcechan.subscribe())
-                .instrument(info_span!("chrony-refclock-task")),
-        ));
-    };
-
-    let gpsd = if config.sources.gpsd.enabled {
-        Some(GpsdSource::new(config.to_owned()).await.unwrap())
-    } else {
-        None
-    };
-    if let Some(gpsd) = gpsd {
-        tasks.push(tokio::spawn(
-            gpsd.run(sourcechan.clone())
-                .instrument(info_span!("gpsd-task")),
-        ))
     }
 
     if tasks.len() == 0 {
         error!("No tasks configured, exiting.");
         return Ok(()); // not an error, but exit before starting a dummy task
     }
-
+    if sourcechan.strong_count() == 0 {
+        warn!("No sources configured, no events will be generated");
+    }
     if sourcechan.receiver_count() == 0 {
-        warn!("No consumers configured, events will be discarded");
+        warn!("No targets configured, events will be discarded");
-        let mut chan = sourcechan.subscribe();
+    }
-        // spawn a dummy task to reap the channel and keep the process alive
+    if args.echo_task || sourcechan.receiver_count() == 0 {
-        tasks.push(tokio::spawn(
+        let c = EchoTargetConfig {};
-            async move {
+        tasks.push(
-                loop {
+            spawn_target::<EchoTarget>("echo", c, sourcechan.subscribe(), shutdown_token.clone())
-                    while let Ok(m) = chan.recv().await {
+                .unwrap(),
-                        info!("received {m:?}");
+        )
-                    }
-                }
-            }
-            .instrument(info_span!("dummy-receiver-task")),
-        ))
     }
 
     debug!("Task setup complete, tasks: {}", tasks.len());
+    ctrlc::set_handler(move || {
+        if shutdown_token.is_cancelled() {
+            info!("Forced shutdown");
+            std::process::exit(1);
+        }
+        info!("Shutting down");
+        shutdown_token.cancel()
+    })
+    .unwrap();
 
     join_all(tasks).await;
 

src/sources/chrony.rs

@@ -1,27 +1,30 @@
-use crate::{
-    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
-    SourceStatus,
-};
-use async_trait::async_trait;
-use chrony_candm::reply::{self, ReplyBody, SourceMode, SourceState};
-use chrony_candm::request::{self, RequestBody};
-use chrony_candm::{ClientOptions, blocking_query};
-use influxdb2::models::DataPoint;
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::Arc;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
-use tokio::join;
+use async_trait::async_trait;
+use chrony_candm::reply::{self, ReplyBody, SourceMode};
+use chrony_candm::request::{self, RequestBody};
+use chrony_candm::{ClientOptions, blocking_query};
+use tokio::select;
+use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 
+use crate::SourceMetricSet;
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceReport,
+    SourceReportDetails, SourceStatus, config::ChronyConfig,
+};
+
 pub struct ChronyClient {
     pub server: SocketAddr,
+    pub name: String,
     client_options: ClientOptions,
-    config: Config,
+    config: ChronyConfig,
 }
 
 #[derive(Debug)]
 pub struct ChronyTrackingReport {
-    tags: Arc<Vec<(String, String)>>,
+    tags: Arc<MetricTags>,
     pub ref_id: i64,
     pub ref_ip_addr: String,
     pub stratum: i64,
@@ -41,26 +44,24 @@ impl SourceReportDetails for ChronyTrackingReport {
     fn is_healthy(&self) -> bool {
         true
     }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-        let tags = &self.tags;
+        vec![SourceMetricSet {
-        vec![
+            tags: self.tags.clone(),
-            SourceMetric::new_int("ref_id", self.ref_id, tags.clone()),
+            metrics: vec![
-            SourceMetric::new_int("stratum", self.stratum, tags.clone()),
+                SourceMetric::new_int("ref_id", self.ref_id),
-            SourceMetric::new_int("leap_status", self.leap_status, tags.clone()),
+                SourceMetric::new_int("stratum", self.stratum),
-            SourceMetric::new_float("current_correction", self.current_correction, tags.clone()),
+                SourceMetric::new_int("leap_status", self.leap_status),
-            SourceMetric::new_float("last_offset", self.last_offset, tags.clone()),
+                SourceMetric::new_float("current_correction", self.current_correction),
-            SourceMetric::new_float("rms_offset", self.rms_offset, tags.clone()),
+                SourceMetric::new_float("last_offset", self.last_offset),
-            SourceMetric::new_float("freq_ppm", self.freq_ppm, tags.clone()),
+                SourceMetric::new_float("rms_offset", self.rms_offset),
-            SourceMetric::new_float("resid_freq_ppm", self.resid_freq_ppm, tags.clone()),
+                SourceMetric::new_float("freq_ppm", self.freq_ppm),
-            SourceMetric::new_float("skew_ppm", self.skew_ppm, tags.clone()),
+                SourceMetric::new_float("resid_freq_ppm", self.resid_freq_ppm),
-            SourceMetric::new_float("root_delay", self.root_delay, tags.clone()),
+                SourceMetric::new_float("skew_ppm", self.skew_ppm),
-            SourceMetric::new_float("root_dispersion", self.root_dispersion, tags.clone()),
+                SourceMetric::new_float("root_delay", self.root_delay),
-            SourceMetric::new_float(
+                SourceMetric::new_float("root_dispersion", self.root_dispersion),
-                "last_update_interval",
+                SourceMetric::new_float("last_update_interval", self.last_update_interval),
-                self.last_update_interval,
+            ],
-                tags.clone(),
+        }]
-            ),
-        ]
     }
 }
 
@@ -74,14 +75,14 @@ impl SourceReportDetails for ChronySourcesReport {
         //TODO: think about whether there is an idea of unhealthy sources
         true
     }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-        let mut metrics = Vec::with_capacity(8 * self.sources.len());
+        let mut metrics = Vec::with_capacity(self.sources.len());
 
         for source in &self.sources {
             let tags = Arc::new(vec![
-                ("ref_id".to_owned(), source.ip_addr.to_string()),
+                ("ref_id", source.ip_addr.to_string()),
                 (
-                    "mode".to_owned(),
+                    "mode",
                     match source.mode {
                         SourceMode::Client => String::from("server"),
                         SourceMode::Peer => String::from("peer"),
@@ -89,7 +90,7 @@ impl SourceReportDetails for ChronySourcesReport {
                     },
                 ),
                 (
-                    "state".to_owned(),
+                    "state",
                     match source.state {
                         reply::SourceState::Selected => String::from("best"),
                         reply::SourceState::NonSelectable => String::from("unusable"),
@@ -100,28 +101,19 @@ impl SourceReportDetails for ChronySourcesReport {
                     },
                 ),
             ]);
-            metrics.extend([
+            metrics.push(SourceMetricSet {
-                SourceMetric::new_int("poll", source.poll as i64, tags.clone()),
+                tags: tags,
-                SourceMetric::new_int("stratum", source.stratum as i64, tags.clone()),
+                metrics: vec![
-                SourceMetric::new_int("flags", source.flags.bits() as i64, tags.clone()),
+                    SourceMetric::new_int("poll", source.poll as i64),
-                SourceMetric::new_int(
+                    SourceMetric::new_int("stratum", source.stratum as i64),
-                    "reachability",
+                    SourceMetric::new_int("flags", source.flags.bits() as i64),
-                    source.reachability.count_ones() as i64,
+                    SourceMetric::new_int("reachability", source.reachability.count_ones() as i64),
-                    tags.clone(),
+                    SourceMetric::new_int("since_sample", source.since_sample as i64),
-                ),
+                    SourceMetric::new_float("orig_latest_meas", source.orig_latest_meas.into()),
-                SourceMetric::new_int("since_sample", source.since_sample as i64, tags.clone()),
+                    SourceMetric::new_float("latest_meas", source.latest_meas.into()),
-                SourceMetric::new_float(
+                    SourceMetric::new_float("latest_meas_err", source.latest_meas_err.into()),
-                    "orig_latest_meas",
+                ],
-                    source.orig_latest_meas.into(),
+            });
-                    tags.clone(),
-                ),
-                SourceMetric::new_float("latest_meas", source.latest_meas.into(), tags.clone()),
-                SourceMetric::new_float(
-                    "latest_meas_err",
-                    source.latest_meas_err.into(),
-                    tags.clone(),
-                ),
-            ]);
         }
 
         metrics
@@ -130,7 +122,7 @@ impl SourceReportDetails for ChronySourcesReport {
 
 fn report_from_tracking(
     t: &reply::Tracking,
-    config: &Config,
+    config: &ChronyConfig,
 ) -> Result<ChronyTrackingReport, Box<dyn std::error::Error>> {
     let report = ChronyTrackingReport {
         tags: Arc::new(vec![]), //TODO: allow configuring tags in the source
@@ -152,25 +144,6 @@ fn report_from_tracking(
 }
 
 impl ChronyClient {
-    pub fn new(config: Config) -> Self {
-        let server = config
-            .sources
-            .chrony
-            .host
-            .to_socket_addrs()
-            .unwrap()
-            .next()
-            .expect("Unable to parse host:port:");
-        let client_options = ClientOptions {
-            n_tries: 3,
-            timeout: Duration::from_secs(config.sources.chrony.timeout),
-        };
-        ChronyClient {
-            server,
-            client_options,
-            config,
-        }
-    }
     async fn query(&self, request: RequestBody) -> Result<reply::Reply, std::io::Error> {
         let server = self.server;
         let client_options = self.client_options;
@@ -266,7 +239,7 @@ impl ChronyClient {
 
         let tracking_data = report_from_tracking(&tracking, &self.config)?;
         let report = SourceReport {
-            name: "chrony-tracking".to_owned(),
+            name: self.name.clone(),
             status: SourceStatus::Unknown,
             details: Arc::new(tracking_data),
         };
@@ -284,7 +257,7 @@ impl ChronyClient {
         let sources = self.get_sources().await?;
         let details = ChronySourcesReport { sources };
         let report = SourceReport {
-            name: "chrony-sources".to_owned(),
+            name: self.name.clone(),
             status: SourceStatus::Unknown,
             details: Arc::new(details),
         };
@@ -296,42 +269,53 @@ impl ChronyClient {
 
 #[async_trait]
 impl ChimemonSource for ChronyClient {
-    async fn run(self, chan: ChimemonSourceChannel) {
+    type Config = ChronyConfig;
+    const TASK_NAME: &'static str = "chrony-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let server = config
+            .host
+            .to_socket_addrs()
+            .unwrap()
+            .next()
+            .expect("Unable to parse host:port:");
+        let client_options = ClientOptions {
+            n_tries: 3,
+            timeout: config.timeout,
+        };
+        ChronyClient {
+            name: name.to_owned(),
+            server,
+            client_options,
+            config,
+        }
+    }
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
         info!("Chrony task started");
 
-        let mut t_interval = tokio::time::interval(Duration::from_secs(
+        let mut t_interval = tokio::time::interval(self.config.tracking_interval);
-            self.config.sources.chrony.tracking_interval,
+        let mut s_interval = tokio::time::interval(self.config.sources_interval);
-        ));
+        loop {
-        let mut s_interval = tokio::time::interval(Duration::from_secs(
+            select! {
-            self.config.sources.chrony.sources_interval,
+                _ = cancel.cancelled() => {
-        ));
+                    return
-
+                },
-        let t_future = async {
+                _ = t_interval.tick() => {
-            let lchan = chan.clone();
+                    match self.tracking_poll(&chan).await {
-            loop {
+                        Ok(_) => (),
-                t_interval.tick().await;
+                        Err(e) => {
-
+                            warn!(error = ?e, "Error in chrony tracking task");
-                match self.tracking_poll(&lchan).await {
+                        }
-                    Ok(_) => (),
+                    }
-                    Err(e) => {
+                },
-                        warn!("Error in chrony task: {}", e.to_string());
+                _ = s_interval.tick() => {
+                    match self.sources_poll(&chan).await {
+                        Ok(_) => (),
+                        Err(e) => {
+                            warn!(error = ?e, "Error in chrony sources task");
+                        }
                     }
                 }
             }
-        };
+        }
-        let s_future = async {
-            let lchan = chan.clone();
-            loop {
-                s_interval.tick().await;
-
-                match self.sources_poll(&lchan).await {
-                    Ok(_) => (),
-                    Err(e) => {
-                        warn!("Error in chrony task: {}", e.to_string());
-                    }
-                }
-            }
-        };
-        join!(t_future, s_future);
     }
 }

src/sources/gpsd.rs

@@ -1,8 +1,3 @@
-use crate::{
-    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport,
-    SourceReportDetails, SourceStatus,
-};
-
 use std::collections::HashMap;
 use std::f64;
 use std::fmt::Debug;
@@ -12,20 +7,25 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use backoff::ExponentialBackoff;
-use futures::{SinkExt, Stream};
+use futures::{SinkExt, Stream, StreamExt};
-use futures::{StreamExt, task::Context};
+use gpsd_proto::{Device, Gst, Mode, Pps, Sky, Tpv, UnifiedResponse, Version};
-use gpsd_proto::{
+use serde::Serialize;
-    Device, Gst, Mode, Pps, ResponseHandshake, Sky, Tpv, UnifiedResponse, Version, Watch,
-};
-use serde::{Deserialize, Deserializer, Serialize};
 use serde_json;
 use tokio::net::{TcpStream, ToSocketAddrs, lookup_host};
-use tokio::time::{Interval, interval, timeout};
+use tokio::time::{interval, timeout};
 use tokio_util::codec::{Framed, LinesCodec};
-use tracing::{debug, debug_span, info, warn};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, debug_span, error, info, instrument, warn};
+
+use crate::SourceMetricSet;
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus, config::GpsdConfig,
+};
 
 pub struct GpsdSource {
-    pub config: Config,
+    pub name: String,
+    pub config: GpsdConfig,
     conn: GpsdTransport,
     devices: HashMap<String, Device>,
     last_gst: Option<Gst>,
@@ -77,20 +77,24 @@ impl SourceReportDetails for GpsdSourceReport {
     fn is_healthy(&self) -> bool {
         self.fix_type != GpsdFixType::Unknown && self.fix_type != GpsdFixType::NoFix
     }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-        let no_tags = Arc::new(vec![]);
+        let tags = Arc::new(vec![]);
-        vec![
+        vec![SourceMetricSet {
-            SourceMetric::new_int("sats_visible", self.sats_visible as i64, no_tags.clone()),
+            tags,
-            SourceMetric::new_int("sats_tracked", self.sats_tracked as i64, no_tags.clone()),
+            metrics: vec![
-            SourceMetric::new_float("tdop", self.tdop, no_tags.clone()),
+                SourceMetric::new_int("sats_visible", self.sats_visible as i64),
-        ]
+                SourceMetric::new_int("sats_tracked", self.sats_tracked as i64),
+                SourceMetric::new_float("tdop", self.tdop),
+            ],
+        }]
     }
 }
 
 impl GpsdSource {
-    pub async fn new(config: Config) -> Result<Self, std::io::Error> {
+    async fn inner_new(name: &str, config: GpsdConfig) -> Result<Self, std::io::Error> {
-        let conn = GpsdTransport::new(&config.sources.gpsd.host).await?;
+        let conn = GpsdTransport::new(&config.host).await?;
         Ok(Self {
+            name: name.to_owned(),
             config,
             conn,
             devices: HashMap::new(),
@@ -100,9 +104,6 @@ impl GpsdSource {
             last_sky: None,
         })
     }
-}
-
-impl GpsdSource {
     async fn send_status(&self, chan: &mut ChimemonSourceChannel) {
         let sky = self.last_sky.as_ref();
         let tpv = self.last_tpv.as_ref();
@@ -119,17 +120,21 @@ impl GpsdSource {
             .and_then(|sky| sky.tdop)
             .map_or(f64::INFINITY, |tdop| tdop as f64);
 
-        chan.send(ChimemonMessage::SourceReport(SourceReport {
+        if let Err(e) = chan.send(
-            name: "gpsd".into(),
+            SourceReport {
-            status: SourceStatus::Unknown,
+                name: self.name.clone(),
-            details: Arc::new(GpsdSourceReport {
+                status: SourceStatus::Unknown,
-                fix_type: tpv.map_or(GpsdFixType::Unknown, |tpv| tpv.mode.into()),
+                details: Arc::new(GpsdSourceReport {
-                sats_tracked,
+                    fix_type: tpv.map_or(GpsdFixType::Unknown, |tpv| tpv.mode.into()),
-                sats_visible,
+                    sats_tracked,
-                tdop,
+                    sats_visible,
-            }),
+                    tdop,
-        }))
+                }),
-        .unwrap();
+            }
+            .into(),
+        ) {
+            error!(error = ?e, "Unable to send to channel")
+        }
     }
 
     fn handle_msg(&mut self, msg: String) -> Result<(), Box<dyn std::error::Error>> {
@@ -165,10 +170,20 @@ impl GpsdSource {
 
 #[async_trait]
 impl ChimemonSource for GpsdSource {
-    async fn run(mut self, mut chan: ChimemonSourceChannel) {
+    type Config = GpsdConfig;
+    const TASK_NAME: &'static str = "gpsd-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        // TODO: refactor so this mess isn't necessary
+        // Should do async setup at the start of run(), not here
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .build()
+            .unwrap();
+        rt.block_on(Self::inner_new(name, config)).unwrap()
+    }
+    async fn run(mut self, mut chan: ChimemonSourceChannel, cancel: CancellationToken) {
         info!("gpsd task started");
-        self.conn.ensure_connection().await.unwrap();
+        self.conn.conn().await.unwrap();
-        let mut ticker = interval(Duration::from_secs(self.config.sources.gpsd.interval));
+        let mut ticker = interval(self.config.interval);
 
         let mut params = WatchParams::default();
         params.json = Some(true);
@@ -179,6 +194,9 @@ impl ChimemonSource for GpsdSource {
         loop {
             let framed = self.conn.framed.as_mut().expect("must be connected");
             tokio::select! {
+                _ = cancel.cancelled() => {
+                    return
+                },
                 _ = ticker.tick() => {
                     self.send_status(&mut chan).await
                 },
@@ -192,6 +210,7 @@ impl ChimemonSource for GpsdSource {
     }
 }
 
+#[derive(Debug)]
 struct GpsdTransport {
     host: SocketAddr,
     framed: Option<Framed<TcpStream, LinesCodec>>,
@@ -337,65 +356,70 @@ impl GpsdTransport {
             conn_backoff: ExponentialBackoff::default(),
         })
     }
-    async fn connect(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+    async fn connect_inner(
+        &self,
+    ) -> Result<Framed<TcpStream, LinesCodec>, Box<dyn std::error::Error>> {
         info!("Connecting to gpsd @ {}", self.host);
-        let mut framed = backoff::future::retry_notify(
+
-            self.conn_backoff.clone(),
+        let mut framed = Framed::new(TcpStream::connect(self.host).await?, LinesCodec::new());
-            async || {
+
-                Ok(Framed::new(
-                    TcpStream::connect(self.host).await?,
-                    LinesCodec::new(),
-                ))
-            },
-            |e, d| warn!("Failed to connect to {} after {:?}: `{}`", self.host, d, e),
-        )
-        .await?;
         debug!("Waiting for initial VERSION");
-        if let Ok(Some(Ok(r))) = timeout(Duration::from_secs(5), framed.next()).await {
+
-            if let Ok(version) = serde_json::from_str::<Version>(&r) {
+        match timeout(Duration::from_secs(5), framed.next()).await {
-                info!(
+            Ok(Some(Ok(r))) => {
-                    "Connected to gpsd @ {}, release {}",
+                if let Ok(version) = serde_json::from_str::<Version>(&r) {
-                    self.host, version.release
+                    info!(
-                )
+                        "Connected to gpsd @ {}, release {}",
-            } else {
+                        self.host, version.release
-                warn!("Got unexpected non-VERSION response after connection (`{r}`)")
+                    );
+                } else {
+                    warn!("Got unexpected non-VERSION response after connection (`{r}`)");
+                }
+                Ok(framed)
             }
+            _ => Err("Unexpected failure to receive initial handshake response".into()),
+        }
+    }
+
+    async fn conn(
+        &mut self,
+    ) -> Result<&mut Framed<TcpStream, LinesCodec>, Box<dyn std::error::Error>> {
+        if self.framed.is_none() {
+            let framed = backoff::future::retry_notify(
+                self.conn_backoff.clone(),
+                || async {
+                    self.connect_inner()
+                        .await
+                        .map_err(backoff::Error::transient)
+                },
+                |e, d| warn!("Failed to connect to {} after {:?}: `{}`", self.host, d, e),
+            )
+            .await?;
+
             self.framed = Some(framed);
-            Ok(())
-        } else {
-            Err("Unexpected failure to receive initial handshake response".into())
-        }
-    }
-    async fn ensure_connection(&mut self) -> Result<(), Box<dyn std::error::Error>> {
-        if let Some(conn) = &self.framed {
-            Ok(())
-        } else {
-            self.connect().await
         }
+
+        Ok(self.framed.as_mut().unwrap())
     }
+
+    #[instrument(level = "debug", skip_all, fields(cmd = cmd.to_string()))]
     async fn cmd_response(
         &mut self,
         cmd: &GpsdCommand,
     ) -> Result<Vec<UnifiedResponse>, Box<dyn std::error::Error>> {
-        debug!("Command: `{cmd:?}`");
-        self.ensure_connection().await?;
-
         let mut responses = Vec::new();
-        if let Some(conn) = &mut self.framed {
+        let conn = self.conn().await?;
-            debug!("Raw command: `{}`", cmd.to_string());
+
-            conn.send(cmd.to_string()).await?;
+        conn.send(cmd.to_string()).await?;
-            for _ in 0..cmd.expected_responses() {
+        for _ in 0..cmd.expected_responses() {
-                match conn.next().await {
+            match conn.next().await {
-                    None => return Err("Connection lost".into()),
+                None => return Err("Connection lost".into()),
-                    Some(Err(e)) => return Err(format!("Unable to parse response {e}").into()),
+                Some(Err(e)) => return Err(format!("Unable to parse response {e}").into()),
-                    Some(Ok(r)) => {
+                Some(Ok(r)) => {
-                        debug!("Raw response: `{r}`");
+                    debug!("Raw response: `{r}`");
-                        responses.push(serde_json::from_str::<UnifiedResponse>(&r)?)
+                    responses.push(serde_json::from_str::<UnifiedResponse>(&r)?)
-                    }
                 }
             }
-        } else {
-            return Err("Missing connection despite ensure".into());
         }
 
         Ok(responses)
@@ -407,12 +431,10 @@ impl GpsdTransport {
         impl Stream<Item = Result<UnifiedResponse, Box<dyn std::error::Error>>>,
         Box<dyn std::error::Error>,
     > {
-        self.ensure_connection().await?;
+        Ok(self
-        if let Some(conn) = &mut self.framed {
+            .conn()
-            Ok(conn.map(|line| Ok(serde_json::from_str::<UnifiedResponse>(&line?)?)))
+            .await?
-        } else {
+            .map(|line| Ok(serde_json::from_str::<UnifiedResponse>(&line?)?)))
-            Err("No connection after connecting.".into())
-        }
     }
 }
 
@@ -420,7 +442,7 @@ mod tests {
     use gpsd_proto::{ResponseData, UnifiedResponse};
     use tokio_stream::StreamExt;
 
-    use crate::gpsd::{GpsdCommand, GpsdTransport, WatchParams};
+    use crate::sources::gpsd::{GpsdCommand, GpsdTransport, WatchParams};
     use std::sync::Once;
 
     static INIT: Once = Once::new();
@@ -433,7 +455,7 @@ mod tests {
     async fn test_gpsd() {
         init_logger();
         let mut gpsd = GpsdTransport::new(&"192.168.65.93:2947").await.unwrap();
-        gpsd.connect().await.unwrap();
+        gpsd.conn().await.unwrap();
         let mut params = WatchParams::default();
         params.enable = Some(true);
         params.json = Some(true);

src/sources/hwmon.rs

@@ -0,0 +1,169 @@
+use std::{fs::File, io::Read, path::PathBuf, sync::Arc};
+
+use async_trait::async_trait;
+use tokio::select;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceMetricSet, SourceReport,
+    SourceReportDetails, SourceStatus, config::HwmonConfig,
+};
+
+pub struct HwmonSource {
+    name: String,
+    config: HwmonConfig,
+    sensors: Vec<Arc<HwmonSensor>>,
+}
+
+#[derive(Debug, Clone)]
+struct HwmonSensor {
+    name: String,
+    value_path: PathBuf,
+    device: String,
+    sensor: String,
+    label: Option<String>,
+    tags: Arc<MetricTags>,
+}
+
+impl HwmonSensor {
+    fn new(name: &str, device: &str, sensor: &str) -> Self {
+        let value_path = PathBuf::from(HWMON_ROOT)
+            .join(device)
+            .join(sensor.to_owned() + "_input");
+        let label_path_raw = PathBuf::from(HWMON_ROOT)
+            .join(device)
+            .join(sensor.to_owned() + "_label");
+        let label = if label_path_raw.is_file() {
+            let mut f =
+                File::open(&label_path_raw).expect(&format!("Unable to open `{label_path_raw:?}`"));
+            let mut label = String::new();
+            f.read_to_string(&mut label)
+                .expect(&format!("Unable to read from `{label_path_raw:?}"));
+            Some(label.trim().to_owned())
+        } else {
+            None
+        };
+        let mut tags_vec = vec![
+            ("name", name.to_owned()),
+            ("device", device.to_owned()),
+            ("sensor", sensor.to_owned()),
+        ];
+        if let Some(label) = &label {
+            tags_vec.push(("label", label.clone()))
+        }
+        Self {
+            value_path,
+            label,
+            device: device.to_owned(),
+            sensor: sensor.to_owned(),
+            name: name.to_owned(),
+            tags: Arc::new(tags_vec),
+        }
+    }
+}
+
+#[derive(Debug)]
+struct HwmonReport {
+    values: Vec<(Arc<HwmonSensor>, f64)>,
+}
+
+impl SourceReportDetails for HwmonReport {
+    fn is_healthy(&self) -> bool {
+        //self.alarms.iter().any(|(_sensor, alarm)| *alarm)
+        true
+    }
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let mut metrics = Vec::new();
+        for (sensor, value) in &self.values {
+            metrics.push(SourceMetricSet {
+                tags: sensor.tags.clone(),
+                metrics: vec![SourceMetric::new_float("hwmon_value", *value)],
+            })
+        }
+        // for (sensor, alarm) in &self.alarms {
+        //     metrics.push(SourceMetric::new_bool(
+        //         "hwmon_alarm",
+        //         *alarm,
+        //         sensor.tags.clone(),
+        //     ))
+        // }
+
+        metrics
+    }
+}
+
+const HWMON_ROOT: &str = "/sys/class/hwmon";
+
+impl HwmonSource {
+    async fn get_raw_value(sensor: &HwmonSensor) -> Result<String, std::io::Error> {
+        tokio::fs::read_to_string(&sensor.value_path).await
+    }
+}
+
+#[async_trait]
+impl ChimemonSource for HwmonSource {
+    type Config = HwmonConfig;
+    const TASK_NAME: &'static str = "hwmon-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let sensors = config
+            .sensors
+            .iter()
+            .map(|(k, v)| Arc::new(HwmonSensor::new(k, &v.device, &v.sensor)))
+            .collect();
+        debug!("config: {config:?}");
+        HwmonSource {
+            name: name.to_owned(),
+            config,
+            sensors,
+        }
+    }
+    async fn run(self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
+        info!("hwmon task started");
+        let mut interval = tokio::time::interval(self.config.interval);
+        loop {
+            select! {
+                _ = cancel.cancelled() => { return; },
+                _ = interval.tick() => {
+                    let mut values = Vec::new();
+                    for s in &self.sensors {
+                        debug!("Sensor {s:?}");
+                        match HwmonSource::get_raw_value(s).await {
+                            Ok(sensor_val) => {
+                            debug!(
+                                "hwmon {} raw value {}",
+                                s.value_path.to_string_lossy(),
+                                sensor_val
+                            );
+                            if let Ok(parsed) = sensor_val.trim().parse::<f64>() {
+                                values.push((s.clone(), parsed));
+                            } else {
+                                error!(
+                                    "Unable to parse sensor value {sensor_val} at {}",
+                                    s.value_path.to_string_lossy()
+                                );
+                            }
+                        },
+                       Err(e) => {
+                            error!("Unable to get hwmon sensor value ({})", e.to_string());
+                            continue
+                        }
+                        }
+                    }
+                    let report = SourceReport {
+                        name: self.name.clone(),
+                        status: SourceStatus::Healthy,
+                        details: Arc::new(HwmonReport { values }),
+                    };
+                    info!("Writing hwmon data");
+                    match chan.send(report.into()) {
+                        Ok(_) => {}
+                        Err(e) => {
+                            warn!("Unable to send to message channel ({e})")
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

src/sources/mod.rs

@@ -0,0 +1,5 @@
+pub mod chrony;
+pub mod gpsd;
+pub mod hwmon;
+pub mod prs10;
+pub mod uccm;

src/sources/prs10.rs

@@ -0,0 +1,601 @@
+use std::any::type_name;
+use std::str::FromStr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use bitflags::bitflags;
+use itertools::Itertools;
+use tokio::io::{
+    AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter, ReadHalf, WriteHalf,
+};
+use tokio::select;
+use tokio::sync::OnceCell;
+use tokio::time::{interval, timeout};
+use tokio_serial::{SerialPort, SerialStream};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, debug_span, error, info, instrument, warn};
+
+use crate::SourceMetricSet;
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, MetricTags, SourceMetric, SourceReport,
+    SourceReportDetails, SourceStatus, config::Prs10Config, fatal,
+};
+
+#[derive(Debug)]
+pub struct Prs10Info {
+    pub model: String,
+    pub version: String,
+    pub serial: String,
+}
+
+impl TryFrom<&[u8]> for Prs10Info {
+    type Error = Box<dyn std::error::Error>;
+    fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
+        let parts = value.splitn(3, |c| *c == b'_');
+        let (model, version, serial) = parts
+            .collect_tuple()
+            .ok_or("Not enough parts in ID response")?;
+        Ok(Self {
+            model: str::from_utf8(model)?.to_string(),
+            version: str::from_utf8(version)?.to_string(),
+            serial: str::from_utf8(serial)?.to_string(),
+        })
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10PowerLampFlags: u8 {
+        const ELEC_VOLTAGE_LOW = (1<<0);
+        const ELEC_VOLTAGE_HIGH = (1<<1);
+        const HEAT_VOLTAGE_LOW = (1<<2);
+        const HEAT_VOLTAGE_HIGH = (1<<3);
+        const LAMP_LIGHT_LOW = (1<<4);
+        const LAMP_LIGHT_HIGH = (1<<5);
+        const GATE_VOLTAGE_LOW = (1<<6);
+        const GATE_VOLTAGE_HIGH = (1<<7);
+    }
+}
+
+impl Prs10PowerLampFlags {
+    pub fn get_metrics(&self, tags: Arc<MetricTags>) -> Vec<SourceMetric> {
+        // Define the mapping statically
+        const FLAG_LABELS: [(&Prs10PowerLampFlags, &str); 8] = [
+            (&Prs10PowerLampFlags::ELEC_VOLTAGE_LOW, "elec_voltage_low"),
+            (&Prs10PowerLampFlags::ELEC_VOLTAGE_HIGH, "elec_voltage_high"),
+            (&Prs10PowerLampFlags::HEAT_VOLTAGE_LOW, "heat_voltage_low"),
+            (&Prs10PowerLampFlags::HEAT_VOLTAGE_HIGH, "heat_voltage_high"),
+            (&Prs10PowerLampFlags::LAMP_LIGHT_LOW, "lamp_light_low"),
+            (&Prs10PowerLampFlags::LAMP_LIGHT_HIGH, "lamp_light_high"),
+            (&Prs10PowerLampFlags::GATE_VOLTAGE_LOW, "gate_voltage_low"),
+            (&Prs10PowerLampFlags::GATE_VOLTAGE_HIGH, "gate_voltage_high"),
+        ];
+
+        // Generate metrics based on flag availability
+        FLAG_LABELS
+            .iter()
+            .map(|(flag, label)| {
+                // We track whether each flag is set (true) or not (false)
+                SourceMetric::new_bool(*label, self.contains(**flag))
+            })
+            .collect()
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10RfFlags: u8 {
+        const PLL_UNLOCK = (1<<0);
+        const XTAL_VAR_LOW = (1<<1);
+        const XTAL_VAR_HIGH = (1<<2);
+        const VCO_CTRL_LOW = (1<<3);
+        const VCO_CTRL_HIGH = (1<<4);
+        const AGC_CTRL_LOW = (1<<5);
+        const AGC_CTRL_HIGH = (1<<6);
+        const PLL_BAD_PARAM = (1<<7);
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10TempFlags: u8 {
+        const LAMP_TEMP_LOW = (1<<0);
+        const LAMP_TEMP_HIGH = (1<<1);
+        const XTAL_TEMP_LOW = (1<<2);
+        const XTAL_TEMP_HIGH = (1<<3);
+        const CELL_TEMP_LOW = (1<<4);
+        const CELL_TEMP_HIGH = (1<<5);
+        const CASE_TEMP_LOW = (1<<6);
+        const CASE_TEMP_HIGH = (1<<7);
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10FllFlags: u8 {
+        const FLL_OFF = (1<<0);
+        const FLL_DISABLED = (1<<1);
+        const EFC_HIGH = (1<<2);
+        const EFC_LOW = (1<<3);
+        const CAL_VOLTAGE_HIGH = (1<<4);
+        const CAL_VOLTAGE_LOW = (1<<5);
+        const _ = !0;
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10PpsFlags: u8 {
+        const PLL_DISABLED = (1<<0);
+        const PPS_WARMUP = (1<<1);
+        const PLL_ACTIVE = (1<<2);
+        const PPS_BAD = (1<<3);
+        const PPS_INTERVAL_LONG = (1<<4);
+        const PLL_RESTART = (1<<5);
+        const PLL_SATURATED = (1<<6);
+        const PPS_MISSING = (1<<7);
+    }
+}
+
+bitflags! {
+    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
+    pub struct Prs10SystemFlags: u8 {
+        const LAMP_RESTART = (1<<0);
+        const WDT_RESET = (1<<1);
+        const BAD_INT_VECTOR = (1<<2);
+        const EEPROM_WRITE_FAIL = (1<<3);
+        const EEPROM_CORRUPT = (1<<4);
+        const BAD_COMMAND = (1<<5);
+        const BAD_COMMAND_PARAM = (1<<6);
+        const SYSTEM_RESET = (1<<7);
+    }
+}
+
+#[derive(Debug, Eq, PartialEq, Clone)]
+pub struct Prs10Status {
+    pub volt_lamp_flags: Prs10PowerLampFlags,
+    pub rf_flags: Prs10RfFlags,
+    pub temp_flags: Prs10TempFlags,
+    pub fll_flags: Prs10FllFlags,
+    pub pps_flags: Prs10PpsFlags,
+    pub system_flags: Prs10SystemFlags,
+}
+
+impl Default for Prs10Status {
+    fn default() -> Self {
+        Self {
+            volt_lamp_flags: Prs10PowerLampFlags::empty(),
+            rf_flags: Prs10RfFlags::empty(),
+            temp_flags: Prs10TempFlags::empty(),
+            fll_flags: Prs10FllFlags::empty(),
+            pps_flags: Prs10PpsFlags::empty(),
+            system_flags: Prs10SystemFlags::empty(),
+        }
+    }
+}
+
+impl SourceReportDetails for Prs10Status {
+    fn is_healthy(&self) -> bool {
+        const HEALTHY_PPS: Prs10PpsFlags = Prs10PpsFlags::from_bits(4).unwrap();
+        self.volt_lamp_flags.is_empty()
+            && self.rf_flags.is_empty()
+            && self.temp_flags.is_empty()
+            && self.fll_flags.is_empty()
+            && self.pps_flags == HEALTHY_PPS
+    }
+
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                SourceMetric::new_int("volt_lamp_flags", self.volt_lamp_flags.bits() as i64),
+                SourceMetric::new_int("rf_flags", self.rf_flags.bits() as i64),
+                SourceMetric::new_int("temp_flags", self.temp_flags.bits() as i64),
+                SourceMetric::new_int("fll_flags", self.fll_flags.bits() as i64),
+                SourceMetric::new_int("pps_flags", self.pps_flags.bits() as i64),
+                // system flags are kind of useless because we can't guarantee they get upstreamed and will only appear once since they are 'event flags'
+            ],
+        }]
+    }
+}
+
+impl TryFrom<&[u8]> for Prs10Status {
+    type Error = Box<dyn std::error::Error>;
+    fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
+        let (volt_lamp_flags, rf_flags, temp_flags, fll_flags, pps_flags, system_flags) = value
+            .splitn(6, |c| *c == b',')
+            .map(|s| str::from_utf8(s).unwrap().parse::<u8>())
+            .collect_tuple()
+            .ok_or("Not enough parts in ST reply")?;
+
+        let volt_lamp_flags = volt_lamp_flags?;
+        let rf_flags = rf_flags?;
+        let temp_flags = temp_flags?;
+        let fll_flags = fll_flags?;
+        let pps_flags = pps_flags?;
+        let system_flags = system_flags?;
+
+        Ok(Self {
+            volt_lamp_flags: Prs10PowerLampFlags::from_bits(volt_lamp_flags).ok_or_else(|| {
+                format!("Invalid bits set ({volt_lamp_flags}) for power/lamp flags")
+            })?,
+            rf_flags: Prs10RfFlags::from_bits(rf_flags)
+                .ok_or_else(|| format!("Invalid bits set ({rf_flags}) for RF flags"))?,
+            temp_flags: Prs10TempFlags::from_bits(temp_flags)
+                .ok_or_else(|| format!("Invalid bits set ({temp_flags}) for temp flags"))?,
+            fll_flags: Prs10FllFlags::from_bits(fll_flags)
+                .ok_or_else(|| format!("Invalid bits set ({fll_flags}) for FLL flags"))?,
+            pps_flags: Prs10PpsFlags::from_bits(pps_flags)
+                .ok_or_else(|| format!("Invalid bits set ({pps_flags}) for PPS flags"))?,
+            system_flags: Prs10SystemFlags::from_bits(system_flags)
+                .ok_or_else(|| format!("Invalid bits set ({system_flags}) for system flags"))?,
+        })
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Prs10Stats {
+    pub ocxo_efc: u32,
+    pub error_signal_volts: f64,
+    pub detect_signal_volts: f64,
+    pub freq_offset_ppt: i16,
+    pub mag_efc: u16,
+    pub heat_volts: f64,
+    pub elec_volts: f64,
+    pub lamp_fet_drain_volts: f64,
+    pub lamp_fet_gate_volts: f64,
+    pub ocxo_heat_volts: f64,
+    pub cell_heat_volts: f64,
+    pub lamp_heat_volts: f64,
+    pub rb_photo: f64,
+    pub rb_photo_iv: f64,
+    pub case_temp: f64,
+    pub ocxo_therm: f64,
+    pub cell_therm: f64,
+    pub lamp_therm: f64,
+    pub ext_cal_volts: f64,
+    pub analog_gnd_volts: f64,
+    pub if_vco_varactor_volts: f64,
+    pub op_vco_varactor_volts: f64,
+    pub mul_amp_gain_volts: f64,
+    pub rf_lock_volts: f64,
+}
+
+impl SourceReportDetails for Prs10Stats {
+    fn is_healthy(&self) -> bool {
+        true
+    }
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
+        let tags = Arc::new(vec![]);
+        vec![SourceMetricSet {
+            tags,
+            metrics: vec![
+                // Integer Metrics
+                SourceMetric::new_int("ocxo_efc", self.ocxo_efc as i64),
+                // Float Metrics
+                SourceMetric::new_float("error_signal_volts", self.error_signal_volts),
+                SourceMetric::new_float("detect_signal_volts", self.detect_signal_volts),
+                SourceMetric::new_float("heat_volts", self.heat_volts),
+                SourceMetric::new_float("elec_volts", self.elec_volts),
+                SourceMetric::new_float("lamp_fet_drain_volts", self.lamp_fet_drain_volts),
+                SourceMetric::new_float("lamp_fet_gate_volts", self.lamp_fet_gate_volts),
+                SourceMetric::new_float("ocxo_heat_volts", self.ocxo_heat_volts),
+                SourceMetric::new_float("cell_heat_volts", self.cell_heat_volts),
+                SourceMetric::new_float("lamp_heat_volts", self.lamp_heat_volts),
+                SourceMetric::new_float("rb_photo", self.rb_photo),
+                SourceMetric::new_float("rb_photo_iv", self.rb_photo_iv),
+                SourceMetric::new_float("case_temp", self.case_temp),
+                SourceMetric::new_float("ocxo_therm", self.ocxo_therm),
+                SourceMetric::new_float("cell_therm", self.cell_therm),
+                SourceMetric::new_float("lamp_therm", self.lamp_therm),
+                SourceMetric::new_float("ext_cal_volts", self.ext_cal_volts),
+                SourceMetric::new_float("analog_gnd_volts", self.analog_gnd_volts),
+                SourceMetric::new_float("if_vco_varactor_volts", self.if_vco_varactor_volts),
+                SourceMetric::new_float("op_vco_varactor_volts", self.op_vco_varactor_volts),
+                SourceMetric::new_float("mul_amp_gain_volts", self.mul_amp_gain_volts),
+                SourceMetric::new_float("rf_lock_volts", self.rf_lock_volts),
+                // U16 Metrics (optional, but can be treated as integers)
+                SourceMetric::new_int("freq_offset_ppt", self.freq_offset_ppt as i64),
+                SourceMetric::new_int("mag_efc", self.mag_efc as i64),
+            ],
+        }]
+    }
+}
+
+#[derive(Debug)]
+pub struct Prs10Monitor {
+    name: String,
+    config: Prs10Config,
+    rx: ReadHalf<SerialStream>,
+    tx: BufWriter<WriteHalf<SerialStream>>,
+    info: OnceCell<Prs10Info>,
+}
+
+impl Prs10Monitor {
+    pub fn info(&self) -> &Prs10Info {
+        self.info.get().expect("info() used before run()")
+    }
+
+    #[instrument(level = "debug", skip_all, fields(cmd = String::from_utf8_lossy(cmd).to_string()))]
+    pub async fn cmd_response(&mut self, cmd: &[u8]) -> Result<Vec<u8>, std::io::Error> {
+        self.tx.write_all(cmd).await?;
+        self.tx.write_u8(b'\r').await?;
+        self.tx.flush().await?;
+        let mut reader = BufReader::new(&mut self.rx);
+        let mut buf = Vec::new();
+        let read = timeout(self.config.timeout, reader.read_until(b'\r', &mut buf)).await??;
+        buf.truncate(buf.len() - 1); // strip "\r"
+        debug!(
+            "raw response: ({read}) `{}`",
+            str::from_utf8(&buf).unwrap_or("<garbage>")
+        );
+        Ok(buf)
+    }
+
+    async fn set_info(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+        let id = self.get_id().await?;
+        self.info.set(id)?;
+        debug!("Set info to {:?}", self.info);
+        Ok(())
+    }
+
+    pub async fn get_status(&mut self) -> Result<Prs10Status, Box<dyn std::error::Error>> {
+        debug!("Getting status");
+        let resp = self.cmd_response(b"ST?").await?;
+        let status = resp.as_slice().try_into();
+        debug!("Got: {status:?}");
+        status
+    }
+
+    pub async fn get_id(&mut self) -> Result<Prs10Info, Box<dyn std::error::Error>> {
+        debug!("Getting identity");
+        let resp = self.cmd_response(b"ID?").await?;
+        let id = resp.as_slice().try_into();
+        debug!("Got: {id:?}");
+        id
+    }
+
+    pub async fn get_analog(&mut self, id: u16) -> Result<f64, Box<dyn std::error::Error>> {
+        let mut cmd = b"AD".to_vec();
+        cmd.extend_from_slice(id.to_string().as_bytes());
+        cmd.push(b'?');
+        let value = self.get_parsed(&cmd).await?;
+        debug!("Got: {value}");
+        Ok(value)
+    }
+
+    pub async fn get_parsed<T: FromStr>(
+        &mut self,
+        cmd: &[u8],
+    ) -> Result<T, Box<dyn std::error::Error>>
+    where
+        T::Err: std::error::Error + 'static,
+    {
+        debug!(
+            "Getting parsed <{}> value for command {}",
+            type_name::<T>(),
+            str::from_utf8(cmd).unwrap_or("<garbage>"),
+        );
+        let resp = self.cmd_response(cmd).await?;
+        let val = str::from_utf8(&resp)?.parse::<T>()?;
+        Ok(val)
+    }
+
+    pub async fn get_ocxo_efc(&mut self) -> Result<u32, Box<dyn std::error::Error>> {
+        debug!("Getting u16,u16 -> u32 for OCXO EFC value");
+        let resp = self.cmd_response(b"FC?").await?;
+        let values = resp
+            .splitn(2, |c| *c == b',')
+            .map(|s| str::from_utf8(s).unwrap().parse::<u16>())
+            .collect_tuple()
+            .ok_or("Not enough values in response to FC?")?;
+        if let (Ok(high), Ok(low)) = values {
+            Ok((high as u32) << 8 | low as u32)
+        } else {
+            Err("Unparseable numbers in response to FC?".into())
+        }
+    }
+
+    pub async fn get_detected_signals(&mut self) -> Result<(f64, f64), Box<dyn std::error::Error>> {
+        debug!("Getting i16,i16 -> f64,f64 detected signals pair");
+        let resp = self.cmd_response(b"DS?").await?;
+        let (error, signal) = resp
+            .splitn(2, |c| *c == b',')
+            .map(|s| str::from_utf8(s).unwrap().parse::<i16>())
+            .collect_tuple()
+            .ok_or("Not enough values in response to DS?".to_string())?;
+        Ok((error? as f64 * 0.15e-6, signal? as f64 * 0.001))
+    }
+
+    #[instrument(skip_all)]
+    async fn status_poll(&mut self) -> Result<ChimemonMessage, Box<dyn std::error::Error>> {
+        let status = self.get_status().await?;
+        Ok(SourceReport {
+            name: self.name.clone(),
+            status: if status.is_healthy() {
+                SourceStatus::Healthy
+            } else {
+                SourceStatus::Unknown
+            },
+            details: Arc::new(status),
+        }
+        .into())
+    }
+
+    #[instrument(skip_all)]
+    async fn stats_poll(&mut self) -> Result<ChimemonMessage, Box<dyn std::error::Error>> {
+        const ANALOG_SCALING: [f64; 20] = [
+            0.0, 10.0, 10.0, 10.0, 10.0, 1.0, 1.0, 1.0, 1.0, 4.0, 100.0, 1.0, 1.0, 1.0, 1.0, 1.0,
+            4.0, 4.0, 4.0, 1.0,
+        ];
+
+        let stats_span = debug_span!("get_stats_serial");
+        let stats_guard = stats_span.enter();
+        let ocxo_efc = self.get_ocxo_efc().await?;
+        let (error_signal_volts, detect_signal_volts) = self.get_detected_signals().await?;
+        let freq_offset_ppt = self.get_parsed(b"SF?").await?;
+        let mag_efc = self.get_parsed(b"MR?").await?;
+        let mut analog_values = [0.0; 20];
+        for i in 1u16..=19 {
+            analog_values[i as usize] = self.get_analog(i).await? * ANALOG_SCALING[i as usize]
+        }
+        drop(stats_guard);
+
+        Ok(SourceReport {
+            name: self.name.clone(),
+            status: SourceStatus::Unknown,
+            details: Arc::new(Prs10Stats {
+                ocxo_efc,
+                error_signal_volts,
+                detect_signal_volts,
+                freq_offset_ppt,
+                mag_efc,
+                heat_volts: analog_values[1],
+                elec_volts: analog_values[2],
+                lamp_fet_drain_volts: analog_values[3],
+                lamp_fet_gate_volts: analog_values[4],
+                ocxo_heat_volts: analog_values[5],
+                cell_heat_volts: analog_values[6],
+                lamp_heat_volts: analog_values[7],
+                rb_photo: analog_values[8],
+                rb_photo_iv: analog_values[9],
+                case_temp: analog_values[10],
+                ocxo_therm: analog_values[11],
+                cell_therm: analog_values[12],
+                lamp_therm: analog_values[13],
+                ext_cal_volts: analog_values[14],
+                analog_gnd_volts: analog_values[15],
+                if_vco_varactor_volts: analog_values[16],
+                op_vco_varactor_volts: analog_values[17],
+                mul_amp_gain_volts: analog_values[18],
+                rf_lock_volts: analog_values[19],
+            }),
+        }
+        .into())
+    }
+
+    async fn reset_rx_state(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+        // flush any pending input and potential responses from the receiver side
+        self.tx.write_u8(b'\r').await?;
+        self.tx.flush().await?;
+        let mut discard = vec![];
+        loop {
+            match timeout(Duration::from_millis(100), self.rx.read_buf(&mut discard)).await {
+                Ok(_) => discard.clear(),
+                Err(_) => break,
+            }
+        }
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl ChimemonSource for Prs10Monitor {
+    type Config = Prs10Config;
+    const TASK_NAME: &'static str = "prs10-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let builder = tokio_serial::new(&config.port, config.baud)
+            .timeout(config.timeout)
+            .data_bits(tokio_serial::DataBits::Eight)
+            .parity(tokio_serial::Parity::None)
+            .stop_bits(tokio_serial::StopBits::One)
+            .flow_control(tokio_serial::FlowControl::None);
+        let mut port = SerialStream::open(&builder).unwrap_or_else(|e| {
+            fatal!(
+                "Failed to open serial port `{}` ({})",
+                config.port,
+                e.to_string()
+            )
+        });
+        port.set_exclusive(true).unwrap_or_else(|e| {
+            fatal!(
+                "Can't lock serial port `{}` ({})",
+                config.port,
+                e.to_string()
+            )
+        });
+        info!(
+            "Opened serial port {}@{}",
+            port.name().unwrap(),
+            port.baud_rate().unwrap()
+        );
+
+        let (rx, tx) = tokio::io::split(port);
+        let tx = BufWriter::new(tx);
+        Self {
+            name: name.to_owned(),
+            config,
+            rx,
+            tx,
+            info: OnceCell::new(),
+        }
+    }
+
+    async fn run(mut self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
+        info!("PRS10 task started");
+        if let Err(e) = self.reset_rx_state().await {
+            error!(error = ?e, "Error clearing PRS10 RX state");
+            return;
+        }
+        if let Err(e) = self.set_info().await {
+            error!(error = ?e, "Error starting PRS10");
+            return;
+        }
+        info!(
+            "Connected to PRS10 model: {} version: {} serial: {}",
+            self.info().model,
+            self.info().version,
+            self.info().serial
+        );
+
+        let mut status_timer = interval(self.config.status_interval);
+        let mut stats_timer = interval(self.config.stats_interval);
+
+        loop {
+            let msg = select! {
+                _ = cancel.cancelled() => {
+                    return;
+                },
+                _ = status_timer.tick() => {
+                    self.status_poll().await
+                },
+                _ = stats_timer.tick() => {
+                    self.stats_poll().await
+                }
+            };
+            match msg {
+                Ok(msg) => {
+                    if let Err(e) = chan.send(msg) {
+                        error!("Unable to send to channel {e}")
+                    }
+                }
+                Err(e) => error!("Error in poll task: {e}"),
+            }
+        }
+    }
+}
+
+mod tests {
+    use crate::sources::prs10::{Prs10Info, Prs10PowerLampFlags, Prs10PpsFlags, Prs10Status};
+    #[test]
+    fn test_info_parse() -> Result<(), Box<dyn std::error::Error>> {
+        const INFO_VECTOR: &[u8] = b"PRS10_3.15_SN_12345";
+        let info: Prs10Info = INFO_VECTOR.try_into()?;
+        assert_eq!(info.model, "PRS10");
+        assert_eq!(info.version, "3.15");
+        assert_eq!(info.serial, "SN_12345");
+        Ok(())
+    }
+    #[test]
+    fn test_status_parse() -> Result<(), Box<dyn std::error::Error>> {
+        //TODO: Add vectors for some more complicated state
+        const STATUS_VECTOR1: &[u8] = b"0,0,0,0,4,0";
+        let status: Prs10Status = STATUS_VECTOR1.try_into()?;
+        let mut expect = Prs10Status::default();
+        expect.pps_flags.set(Prs10PpsFlags::PLL_ACTIVE, true);
+        assert_eq!(status, expect);
+        Ok(())
+    }
+}

src/sources/uccm.rs

@@ -1,25 +1,26 @@
-use crate::{
+use std::io::{BufRead, Cursor};
-    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric,
+use std::str;
-    SourceReportDetails, TimeReport,
+use std::sync::Arc;
-};
+
 use async_trait::async_trait;
 use bitflags::bitflags;
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, BytesMut};
 use chrono::{DateTime, Duration, NaiveDateTime, Utc};
-use figment::value::Map;
-use influxdb2::models::DataPoint;
-use influxdb2::models::data_point::DataPointBuilder;
 use itertools::Itertools;
-use std::io::{BufRead, Cursor};
-use std::str;
-use std::sync::Arc;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
-use tokio::join;
+use tokio::select;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use tokio_serial::{SerialPort, SerialStream};
-use tracing::{debug, info, warn};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus, TimeReport, config::UCCMConfig,
+};
+use crate::{SourceMetricSet, fatal};
 
 pub const GPS_EPOCH: i64 = 315964800; // Doesn't seem possible to have a const DateTime object
 pub type UccmEndian = BigEndian;
@@ -32,11 +33,11 @@ pub enum UCCMMonitorParseState {
 }
 
 pub struct UCCMMonitor {
-    // pub port: SerialStream,
+    pub name: String,
+    config: UCCMConfig,
     rx: ReadHalf<SerialStream>,
     tx: WriteHalf<SerialStream>,
     pub info: Option<UCCMInfo>,
-    config: Config,
 }
 
 #[derive(Debug)]
@@ -59,88 +60,33 @@ impl SourceReportDetails for UCCMTODReport {
             && !self.flags.contains(UCCMFlags::GPS_LOS)
     }
 
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-        let no_tags = Arc::new(vec![]);
+        let tags = Arc::new(vec![]);
-        vec![
+        vec![SourceMetricSet {
-            SourceMetric::new_int("leaps", self.leaps as i64, no_tags.clone()),
+            tags,
-            SourceMetric::new_bool(
+            metrics: vec![
-                "osc_lock",
+                SourceMetric::new_int("leaps", self.leaps as i64),
-                self.flags.contains(UCCMFlags::OSC_LOCK),
+                SourceMetric::new_bool("osc_lock", self.flags.contains(UCCMFlags::OSC_LOCK)),
-                no_tags.clone(),
+                SourceMetric::new_bool("leap_flag", self.flags.contains(UCCMFlags::LEAP_FLAG)),
-            ),
+                SourceMetric::new_bool("init_unlock", self.flags.contains(UCCMFlags::INIT_UNLOCK)),
-            SourceMetric::new_bool(
+                SourceMetric::new_bool(
-                "leap_flag",
+                    "init_no_sats",
-                self.flags.contains(UCCMFlags::LEAP_FLAG),
+                    self.flags.contains(UCCMFlags::INIT_NO_SATS),
-                no_tags.clone(),
+                ),
-            ),
+                SourceMetric::new_bool(
-            SourceMetric::new_bool(
+                    "have_gps_time",
-                "init_unlock",
+                    self.flags.contains(UCCMFlags::HAVE_GPS_TIME),
-                self.flags.contains(UCCMFlags::INIT_UNLOCK),
+                ),
-                no_tags.clone(),
+                SourceMetric::new_bool("power_fail", self.flags.contains(UCCMFlags::POWER_FAIL)),
-            ),
+                SourceMetric::new_bool("no_gps_sync", self.flags.contains(UCCMFlags::NO_GPS_SYNC)),
-            SourceMetric::new_bool(
+                SourceMetric::new_bool(
-                "init_no_sats",
+                    "no_gps_sync2",
-                self.flags.contains(UCCMFlags::INIT_NO_SATS),
+                    self.flags.contains(UCCMFlags::NO_GPS_SYNC2),
-                no_tags.clone(),
+                ),
-            ),
+                SourceMetric::new_bool("ant_fault", self.flags.contains(UCCMFlags::NO_ANT)),
-            SourceMetric::new_bool(
+                SourceMetric::new_bool("gps_los", self.flags.contains(UCCMFlags::GPS_LOS)),
-                "have_gps_time",
+            ],
-                self.flags.contains(UCCMFlags::HAVE_GPS_TIME),
+        }]
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "power_fail",
-                self.flags.contains(UCCMFlags::POWER_FAIL),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "no_gps_sync",
-                self.flags.contains(UCCMFlags::NO_GPS_SYNC),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "no_gps_sync2",
-                self.flags.contains(UCCMFlags::NO_GPS_SYNC2),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "ant_fault",
-                self.flags.contains(UCCMFlags::NO_ANT),
-                no_tags.clone(),
-            ),
-            SourceMetric::new_bool(
-                "gps_los",
-                self.flags.contains(UCCMFlags::GPS_LOS),
-                no_tags.clone(),
-            ),
-        ]
-    }
-}
-
-impl UCCMTODReport {
-    pub fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
-        let mut builder =
-            DataPoint::builder(measurement).timestamp(self.time.timestamp_nanos_opt().unwrap());
-        builder = builder.field("leaps", self.leaps as i64);
-        builder = builder.field("osc_lock", self.flags.contains(UCCMFlags::OSC_LOCK));
-        builder = builder.field("leap_flag", self.flags.contains(UCCMFlags::LEAP_FLAG));
-        builder = builder.field("init_unlock", self.flags.contains(UCCMFlags::INIT_UNLOCK));
-        builder = builder.field("init_no_sats", self.flags.contains(UCCMFlags::INIT_NO_SATS));
-        builder = builder.field(
-            "have_gps_time",
-            self.flags.contains(UCCMFlags::HAVE_GPS_TIME),
-        );
-        builder = builder.field("power_fail", self.flags.contains(UCCMFlags::POWER_FAIL));
-        builder = builder.field("no_gps_sync", self.flags.contains(UCCMFlags::NO_GPS_SYNC));
-        builder = builder.field("no_gps_sync2", self.flags.contains(UCCMFlags::NO_GPS_SYNC2));
-        builder = builder.field("ant_fault", self.flags.contains(UCCMFlags::NO_ANT));
-        builder = builder.field("gps_los", self.flags.contains(UCCMFlags::GPS_LOS));
-
-        builder = tags
-            .iter()
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
-
-        builder
     }
 }
 
@@ -149,14 +95,16 @@ pub struct UCCMLoopDiagReport {
     pub ocxo: f32,
 }
 
-impl UCCMLoopDiagReport {
+impl SourceReportDetails for UCCMLoopDiagReport {
-    pub fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
+    fn is_healthy(&self) -> bool {
-        let mut builder = DataPoint::builder(measurement);
+        true
-        builder = builder.field("ocxo_offset", self.ocxo as f64);
+    }
-        builder = tags
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-            .iter()
+        let tags = Arc::new(vec![]);
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
+        vec![SourceMetricSet {
-        builder
+            tags,
+            metrics: vec![SourceMetric::new_float("ocxo_offset", self.ocxo as f64)],
+        }]
     }
 }
 
@@ -166,16 +114,9 @@ pub struct UCCMGpsSvTracking {
     pub cno: u8,
 }
 
-impl UCCMGpsSvTracking {
+impl From<&UCCMGpsSvTracking> for SourceMetric {
-    fn as_builder(&self, measurement: &String, tags: &Map<String, String>) -> DataPointBuilder {
+    fn from(value: &UCCMGpsSvTracking) -> Self {
-        let mut builder = DataPoint::builder(measurement)
+        SourceMetric::new_int("sv_cno", value.cno as i64)
-            .field("sv_cno", self.cno as i64)
-            .tag("sv_id", self.sv.to_string());
-        builder = tags
-            .iter()
-            .fold(builder, |builder, (k, v)| builder.tag(k, v));
-
-        builder
     }
 }
 
@@ -184,17 +125,20 @@ pub struct UCCMGPSSatsReport {
     pub tracked_svs: Vec<UCCMGpsSvTracking>,
 }
 
-impl UCCMGPSSatsReport {
+impl SourceReportDetails for UCCMGPSSatsReport {
-    pub fn build(&self, measurement: &String, tags: &Map<String, String>) -> Vec<DataPoint> {
+    fn is_healthy(&self) -> bool {
-        self.tracked_svs
+        self.tracked_svs.len() >= 4
-            .iter()
+    }
-            .map(|sv| sv.as_builder(measurement, tags))
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-            .map(|b| b.build().unwrap())
+        vec![SourceMetricSet {
-            .collect()
+            tags: Arc::new(vec![]),
+            metrics: self.tracked_svs.iter().map(|sv| sv.into()).collect(),
+        }]
     }
 }
 
 bitflags! {
+    #[derive(Debug)]
     pub struct UCCMFlags: u32 {
         const OSC_LOCK = (1<<29);
         const LEAP_FLAG = (1<<25);
@@ -272,20 +216,23 @@ impl SourceReportDetails for UCCMStatusReport {
     fn is_healthy(&self) -> bool {
         self.gps_pps_valid
     }
-    fn to_metrics(&self) -> Vec<SourceMetric> {
+    fn to_metrics(&self) -> Vec<SourceMetricSet> {
-        let no_tags = Arc::new(vec![]);
+        let tags = Arc::new(vec![]);
-        vec![
+        vec![SourceMetricSet {
-            SourceMetric::new_int("tfom", self.tfom as i64, no_tags.clone()),
+            tags,
-            SourceMetric::new_int("ffom", self.ffom as i64, no_tags.clone()),
+            metrics: vec![
-            SourceMetric::new_float("gps_phase", self.gps_phase as f64, no_tags.clone()),
+                SourceMetric::new_int("tfom", self.tfom as i64),
-            // TODO: sv info
+                SourceMetric::new_int("ffom", self.ffom as i64),
-            // TOOD: timestamp
+                SourceMetric::new_float("gps_phase", self.gps_phase as f64),
-            SourceMetric::new_float("ant_voltage", self.ant_voltage as f64, no_tags.clone()),
+                // TODO: sv info
-            SourceMetric::new_float("ant_current", self.ant_current as f64, no_tags.clone()),
+                // TOOD: timestamp
-            SourceMetric::new_float("temp", self.temp as f64, no_tags.clone()),
+                SourceMetric::new_float("ant_voltage", self.ant_voltage as f64),
-            SourceMetric::new_int("efc_dac", self.efc_dac as i64, no_tags.clone()),
+                SourceMetric::new_float("ant_current", self.ant_current as f64),
-            SourceMetric::new_float("freq_error", self.freq_error as f64, no_tags.clone()),
+                SourceMetric::new_float("temp", self.temp as f64),
-        ]
+                SourceMetric::new_int("efc_dac", self.efc_dac as i64),
+                SourceMetric::new_float("freq_error", self.freq_error as f64),
+            ],
+        }]
     }
 }
 
@@ -398,30 +345,6 @@ impl TryFrom<&str> for UCCMGPSSatsReport {
 }
 
 impl UCCMMonitor {
-    pub fn new(config: Config) -> Self {
-        let builder = tokio_serial::new(&config.sources.uccm.port, config.sources.uccm.baud)
-            .timeout(config.sources.uccm.timeout)
-            .data_bits(tokio_serial::DataBits::Eight)
-            .parity(tokio_serial::Parity::None)
-            .stop_bits(tokio_serial::StopBits::One)
-            .flow_control(tokio_serial::FlowControl::None);
-        let mut port = SerialStream::open(&builder).expect("Must be able to open serial port");
-        port.set_exclusive(true).expect("Can't lock serial port");
-        info!(
-            "Opened serial port {}@{}",
-            port.name().unwrap(),
-            port.baud_rate().unwrap()
-        );
-        let (rx, tx) = tokio::io::split(port);
-        UCCMMonitor {
-            // port,
-            rx,
-            tx,
-            info: None,
-            config,
-        }
-    }
-
     pub async fn send_cmd(&mut self, cmd: &[u8]) -> Result<String, std::io::Error> {
         debug!("cmd: `{:?}`", String::from_utf8_lossy(cmd));
         self.tx.write_all(cmd).await.unwrap();
@@ -477,10 +400,10 @@ impl UCCMMonitor {
         state: Arc<Mutex<UCCMMonitorParseState>>,
     ) {
         let mut rdbuf = BytesMut::with_capacity(1024);
-        let mut last_loop_diag: Option<UCCMLoopDiagReport> = None;
+        let mut last_loop_diag: Option<Arc<UCCMLoopDiagReport>> = None;
-        let mut last_gps_sats: Option<UCCMGPSSatsReport> = None;
+        let mut last_gps_sats: Option<Arc<UCCMGPSSatsReport>> = None;
 
-        let mut last_sent_report = Utc::now() - self.config.sources.uccm.status_interval;
+        let mut last_sent_report = Utc::now() - self.config.status_interval;
 
         loop {
             match tokio::io::AsyncReadExt::read_buf(&mut self.rx, &mut rdbuf).await {
@@ -515,46 +438,45 @@ impl UCCMMonitor {
                                 && tod
                                     .flags
                                     .contains(UCCMFlags::OSC_LOCK | UCCMFlags::HAVE_GPS_TIME);
-                            chan.send(ChimemonMessage::TimeReport(TimeReport {
+                            chan.send(
-                                system_time: sysnow,
+                                TimeReport {
-                                offset,
+                                    system_time: sysnow,
-                                leaps: tod.leaps as isize,
+                                    offset,
-                                leap_flag: tod.flags.contains(UCCMFlags::LEAP_FLAG),
+                                    leaps: tod.leaps as isize,
-                                valid,
+                                    leap_flag: tod.flags.contains(UCCMFlags::LEAP_FLAG),
-                            }))
+                                    valid,
+                                }
+                                .into(),
+                            )
                             .expect("Unable to send to channel");
                             if sysnow - last_sent_report
-                                >= Duration::from_std(self.config.sources.uccm.status_interval)
+                                >= Duration::from_std(self.config.status_interval).unwrap()
-                                    .unwrap()
                             {
-                                let mut points = vec![
-                                    tod.as_builder(
-                                        &self.config.sources.uccm.measurement,
-                                        &self.config.influxdb.tags,
-                                    )
-                                    .build()
-                                    .unwrap(),
-                                ];
                                 if let Some(loop_diag) = &last_loop_diag {
-                                    points.push(
+                                    if let Err(e) = chan.send(
-                                        loop_diag
+                                        SourceReport {
-                                            .as_builder(
+                                            name: "uccm".to_owned(),
-                                                &self.config.sources.uccm.measurement,
+                                            status: SourceStatus::Unknown,
-                                                &self.config.influxdb.tags,
+                                            details: loop_diag.clone(),
-                                            )
+                                        }
-                                            .build()
+                                        .into(),
-                                            .unwrap(),
+                                    ) {
-                                    )
+                                        error!(error = ?e, "Unable to send message to channel");
+                                    }
                                 }
                                 if let Some(gps_sats) = &last_gps_sats {
-                                    points.extend(gps_sats.build(
+                                    if let Err(e) = chan.send(
-                                        &self.config.sources.uccm.measurement,
+                                        SourceReport {
-                                        &self.config.influxdb.tags,
+                                            name: "uccm".to_owned(),
-                                    ));
+                                            status: SourceStatus::Unknown,
+                                            details: gps_sats.clone(),
+                                        }
+                                        .into(),
+                                    ) {
+                                        error!(error = ?e, "Unable to send message to channel");
+                                    }
                                 }
 
-                                chan.send(ChimemonMessage::DataPoints(points))
-                                    .expect("Unable to send to channel");
                                 last_sent_report = sysnow;
                             }
                         }
@@ -568,7 +490,7 @@ impl UCCMMonitor {
                     let loop_report = UCCMLoopDiagReport::try_from(loop_diag_resp.as_str());
                     let gps_report = UCCMGPSSatsReport::try_from(gps_sats_resp.as_str());
                     if let Ok(loop_report) = loop_report {
-                        last_loop_diag = Some(loop_report)
+                        last_loop_diag = Some(Arc::new(loop_report))
                     } else {
                         warn!(
                             "Unable to parse loop diag report `{}`: {}",
@@ -577,7 +499,7 @@ impl UCCMMonitor {
                         );
                     }
                     if let Ok(gps_report) = gps_report {
-                        last_gps_sats = Some(gps_report)
+                        last_gps_sats = Some(Arc::new(gps_report))
                     } else {
                         warn!(
                             "Unable to parse GPS sats report `{}`: {}",
@@ -596,7 +518,38 @@ impl UCCMMonitor {
 
 #[async_trait]
 impl ChimemonSource for UCCMMonitor {
-    async fn run(mut self, chan: ChimemonSourceChannel) {
+    type Config = UCCMConfig;
+    const TASK_NAME: &'static str = "uccm-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let builder = tokio_serial::new(&config.port, config.baud)
+            .timeout(config.timeout)
+            .data_bits(tokio_serial::DataBits::Eight)
+            .parity(tokio_serial::Parity::None)
+            .stop_bits(tokio_serial::StopBits::One)
+            .flow_control(tokio_serial::FlowControl::None);
+        let mut port = match SerialStream::open(&builder) {
+            Ok(port) => port,
+            Err(e) => fatal!(error = ?e, "Error opening port {}", &config.port),
+        };
+        if let Err(e) = port.set_exclusive(true) {
+            fatal!(error= ?e, "Can't lock serial port");
+        };
+        info!(
+            "Opened serial port {}@{}",
+            port.name().unwrap(),
+            port.baud_rate().unwrap()
+        );
+        let (rx, tx) = tokio::io::split(port);
+        UCCMMonitor {
+            name: name.to_owned(),
+            config,
+            rx,
+            tx,
+            info: None,
+        }
+    }
+
+    async fn run(mut self, chan: ChimemonSourceChannel, cancel: CancellationToken) {
         info!("UCCM task starting");
         if self.get_info().await.is_err() {
             warn!("Error starting UCCM");
@@ -622,7 +575,10 @@ impl ChimemonSource for UCCMMonitor {
         //         wfut.await;
         //     }
         // });
-
+        //
-        join!(rx_handle).0.unwrap();
+        select! {
+            _ = cancel.cancelled() => { return },
+            _ = rx_handle => { return }
+        };
     }
 }

src/targets/chrony_refclock.rs

@@ -0,0 +1,88 @@
+use std::mem;
+use std::os::unix::net::UnixDatagram;
+use std::path::PathBuf;
+
+use async_trait::async_trait;
+use libc::{c_double, c_int, timeval};
+use tokio::select;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, info, warn};
+
+use crate::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, config::ChronySockConfig};
+
+const CHRONY_MAGIC: c_int = 0x534f434b;
+
+pub struct ChronySockServer {
+    name: String,
+    sock_path: PathBuf,
+}
+
+#[repr(C)]
+#[derive(Debug)]
+pub struct ChronyTimeReport {
+    tv: timeval,
+    offset: c_double,
+    pulse: c_int,
+    leap: c_int,
+    _pad: c_int,
+    magic: c_int,
+}
+
+impl ChronySockServer {}
+
+#[async_trait]
+impl ChimemonTarget for ChronySockServer {
+    type Config = ChronySockConfig;
+    const TASK_NAME: &'static str = "chrony-refclock-task";
+    fn new(name: &str, config: ChronySockConfig) -> Self {
+        ChronySockServer {
+            name: name.to_owned(),
+            sock_path: config.sock.into(),
+        }
+    }
+    async fn run(mut self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Chrony refclock task started");
+        loop {
+            select! {
+                _ = cancel.cancelled() => { return }
+                msg = chan.recv() => {
+                    match msg {
+                        Ok(ChimemonMessage::TimeReport(tr)) => {
+                            debug!(tr = ?tr, "Got timereport");
+                            if tr.valid {
+                                {
+                                    let frame = ChronyTimeReport {
+                                        tv: timeval {
+                                            tv_sec: TryInto::<libc::time_t>::try_into(
+                                                tr.system_time.timestamp(),
+                                            )
+                                            .unwrap(),
+                                            tv_usec: tr.system_time.timestamp_subsec_micros()
+                                                as libc::suseconds_t,
+                                        },
+                                        offset: tr.offset.num_nanoseconds().unwrap() as f64 / 1e9,
+                                        leap: if tr.leap_flag { 1 } else { 0 },
+                                        pulse: 0,
+                                        _pad: 0,
+                                        magic: CHRONY_MAGIC,
+                                    };
+                                    let bs = unsafe {
+                                        std::slice::from_raw_parts(
+                                            (&frame as *const ChronyTimeReport) as *const u8,
+                                            mem::size_of::<ChronyTimeReport>(),
+                                        )
+                                    };
+                                    debug!("Sending to chrony sock {:#?}", frame);
+                                    let sock = UnixDatagram::unbound().unwrap();
+                                    sock.send_to(bs, &self.sock_path).unwrap();
+                                }
+                            }
+                        },
+                        Err(e) => warn!("Error receiving from channel: {}", e.to_string()),
+                        _ => continue,
+                    }
+                }
+            }
+        }
+    }
+}

src/targets/influx.rs

@@ -0,0 +1,110 @@
+use async_trait::async_trait;
+use futures::stream;
+use influxdb2::{
+    Client,
+    models::{DataPoint, FieldValue},
+};
+use tokio::{select, sync::broadcast, time::timeout};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, instrument};
+
+use crate::{
+    ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, MetricValue, SourceReport,
+    config::InfluxConfig, fatal,
+};
+
+pub struct InfluxTarget {
+    name: String,
+    config: InfluxConfig,
+    influx: Client,
+}
+
+impl From<MetricValue> for FieldValue {
+    fn from(value: MetricValue) -> Self {
+        match value {
+            MetricValue::Bool(b) => FieldValue::Bool(b),
+            MetricValue::Float(f) => FieldValue::F64(f),
+            MetricValue::Int(i) => FieldValue::I64(i),
+        }
+    }
+}
+
+#[async_trait]
+impl ChimemonTarget for InfluxTarget {
+    type Config = InfluxConfig;
+    const TASK_NAME: &'static str = "influx-task";
+    fn new(name: &str, config: Self::Config) -> Self {
+        let influx = Client::new(&config.url, &config.org, &config.token);
+        Self {
+            name: name.to_owned(),
+            config: config,
+            influx,
+        }
+    }
+    async fn run(self, mut chan: ChimemonTargetChannel, cancel: CancellationToken) {
+        info!("Influx task started");
+
+        loop {
+            let msg = select! {
+                _ = cancel.cancelled() => { return },
+                msg = chan.recv() => msg
+            };
+            debug!(msg = ?msg, "Got msg");
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(broadcast::error::RecvError::Closed) => {
+                    fatal!("Permanent channel closed, terminating")
+                }
+                Err(broadcast::error::RecvError::Lagged(_)) => {
+                    error!("Channel lagged");
+                    continue;
+                }
+            };
+            if let Err(e) = self.handle_msg(&msg).await {
+                error!(error = ?e, msg=?&msg, "Error handling message");
+            }
+        }
+    }
+}
+
+impl InfluxTarget {
+    #[instrument(skip_all)]
+    async fn handle_source_report(
+        &self,
+        sr: &SourceReport,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        debug!("Handling source report {}", sr.name);
+        let mut dps = Vec::new();
+        for metric_set in &sr.details.to_metrics() {
+            let mut builder = DataPoint::builder(&sr.name);
+            builder = self
+                .config
+                .tags
+                .iter()
+                .fold(builder, |builder, (k, v)| builder.tag(k, v));
+            builder = metric_set
+                .tags
+                .iter()
+                .fold(builder, |builder, (k, v)| builder.tag(*k, v));
+            builder = metric_set.metrics.iter().fold(builder, |builder, metric| {
+                builder.field(metric.name, metric.value)
+            });
+            dps.push(builder.build()?);
+        }
+        debug!("Sending {} datapoints to influx", dps.len());
+        timeout(
+            self.config.timeout,
+            self.influx.write(&self.config.bucket, stream::iter(dps)),
+        )
+        .await??;
+        debug!("All datapoints sent");
+        Ok(())
+    }
+    async fn handle_msg(&self, msg: &ChimemonMessage) -> Result<(), Box<dyn std::error::Error>> {
+        debug!(msg = ?msg, "Handling msg");
+        match msg {
+            ChimemonMessage::TimeReport(_tr) => Ok(()),
+            ChimemonMessage::SourceReport(sr) => self.handle_source_report(sr).await,
+        }
+    }
+}

src/targets/mod.rs

@@ -0,0 +1,2 @@
+pub mod chrony_refclock;
+pub mod influx;