incremental improvements

Cargo.toml

@@ -1,31 +1,39 @@
 [package]
 name = "chimemon"
 version = "0.2.0"
-edition = "2021"
+edition = "2024"
+
+[features]
+default = []
+release-logs = ["tracing/max_level_info"]
 
 [dependencies]
 serde = "1.0"
 serde_derive = "1.0"
-influxdb2 = { version = "0.3.3", features = [
-    "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"
-env_logger = "0.9.1"
 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"
+gpsd_proto = { version = "1.0.0" }
+tokio-util = { version = "0.7.17", features = ["codec"] }
+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", "ansi", "time", "env-filter"] }
+serialport = "4.8.1"
+gethostname = "1.1.0"
+bitflags = "2.10.0"
+influxdb2 = "0.3.9"
+chrono = "0.4.43"
 
 [dependencies.chrony-candm]
 git = "https://github.com/aws/chrony-candm"

config.toml.dist

@@ -23,6 +23,13 @@
     port = "/dev/ttyUSB0"
     status_interval = 10
     measurement = "uccm_gpsdo"
+    
+    [sources.prs10]
+    enabled = true
+    port = "/dev/ttyUSB0"
+    status_interval = 10
+    stats_interval = 10
+    measurement = "prs10_gpsdo"
 
 
 [targets]
@@ -38,4 +45,4 @@ token = ""
 
 [influxdb.tags]
 # host = "qubit" # default is the local hostname
-# arbitrary = "tags" # are allowed
+# arbitrary = "tags" # are allowed

src/chrony.rs

@@ -1,276 +0,0 @@
-use async_trait::async_trait;
-use chimemon::{ChimemonSource, ChimemonSourceChannel, Config};
-use chrony_candm::reply::{self, ReplyBody, SourceMode};
-use chrony_candm::request::{self, RequestBody};
-use chrony_candm::{blocking_query, ClientOptions};
-use influxdb2::models::DataPoint;
-use log::{info, warn};
-use std::net::{SocketAddr, ToSocketAddrs};
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use tokio::join;
-
-pub struct ChronyClient {
-    pub server: SocketAddr,
-    client_options: ClientOptions,
-    config: Config,
-}
-
-fn datapoint_from_tracking(
-    t: &reply::Tracking,
-    config: &Config,
-) -> Result<DataPoint, Box<dyn std::error::Error>> {
-    let now = SystemTime::now().duration_since(UNIX_EPOCH)?;
-    let measurement = config.sources.chrony.measurement_prefix.to_owned()
-        + &config.sources.chrony.tracking_measurement;
-    let mut builder =
-        DataPoint::builder(&measurement).timestamp(now.as_nanos().try_into().unwrap());
-    for (key, value) in &config.influxdb.tags {
-        builder = builder.tag(key, value);
-    }
-
-    let point = builder
-        .field("ref_id", t.ref_id as i64)
-        .field("ref_ip_addr", t.ip_addr.to_string())
-        .field("stratum", t.stratum as i64)
-        .field("leap_status", t.leap_status as i64)
-        .field("current_correction", f64::from(t.current_correction))
-        .field("last_offset", f64::from(t.last_offset))
-        .field("rms_offset", f64::from(t.rms_offset))
-        .field("freq_ppm", f64::from(t.freq_ppm))
-        .field("resid_freq_ppm", f64::from(t.resid_freq_ppm))
-        .field("skew_ppm", f64::from(t.skew_ppm))
-        .field("root_delay", f64::from(t.root_delay))
-        .field("root_dispersion", f64::from(t.root_dispersion))
-        .field("last_update_interval", f64::from(t.last_update_interval))
-        .build()?;
-
-    Ok(point)
-}
-
-pub fn datapoint_from_sourcedata(
-    d: &reply::SourceData,
-    config: &Config,
-) -> Result<DataPoint, Box<dyn std::error::Error>> {
-    let now = SystemTime::now().duration_since(UNIX_EPOCH)?;
-    let measurement = config.sources.chrony.measurement_prefix.to_owned()
-        + &config.sources.chrony.sources_measurement;
-    let mut builder =
-        DataPoint::builder(&measurement).timestamp(now.as_nanos().try_into().unwrap());
-    for (key, value) in &config.influxdb.tags {
-        builder = builder.tag(key, value)
-    }
-    builder = builder
-        .tag("ref_id", d.ip_addr.to_string())
-        .tag(
-            "mode",
-            match d.mode {
-                SourceMode::Client => String::from("server"),
-                SourceMode::Peer => String::from("peer"),
-                SourceMode::Ref => String::from("refclock"),
-            },
-        )
-        .tag(
-            "state",
-            match d.state {
-                reply::SourceState::Selected => String::from("best"),
-                reply::SourceState::NonSelectable => String::from("unusable"),
-                reply::SourceState::Falseticker => String::from("falseticker"),
-                reply::SourceState::Jittery => String::from("jittery"),
-                reply::SourceState::Unselected => String::from("combined"),
-                reply::SourceState::Selectable => String::from("unused"),
-            },
-        )
-        .field("poll", d.poll as i64)
-        .field("stratum", d.stratum as i64)
-        .field("flags", d.flags.bits() as i64)
-        .field("reachability", d.reachability.count_ones() as i64)
-        .field("since_sample", d.since_sample as i64)
-        .field("orig_latest_meas", f64::from(d.orig_latest_meas))
-        .field("latest_meas", f64::from(d.latest_meas))
-        .field("latest_meas_err", f64::from(d.latest_meas_err));
-    let point = builder.build()?;
-
-    Ok(point)
-}
-
-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;
-        tokio::task::spawn_blocking(move || blocking_query(request, client_options, &server))
-            .await
-            .map_err(|e| {
-                std::io::Error::new(
-                    std::io::ErrorKind::Other,
-                    format!("Error joining thread: {}", e),
-                )
-            })?
-    }
-
-    pub async fn get_tracking(&self) -> Result<reply::Tracking, std::io::Error> {
-        let reply = self.query(RequestBody::Tracking).await?;
-
-        match reply.body {
-            ReplyBody::Tracking(tracking) => Ok(tracking),
-            _ => Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidData,
-                "Unexpected response type",
-            )),
-        }
-    }
-
-    pub async fn get_sources(&self) -> Result<Vec<reply::SourceData>, std::io::Error> {
-        let reply = self.query(RequestBody::NSources).await?;
-
-        let nsources = match reply.body {
-            ReplyBody::NSources(ns) => Ok(i32::try_from(ns.n_sources).unwrap()),
-            _ => Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidData,
-                "Unexpected response type",
-            )),
-        }?;
-
-        let mut res = Vec::with_capacity(
-            nsources
-                .try_into()
-                .expect("Ridiculously unconvertible number of sources"),
-        );
-
-        for x in 0..nsources {
-            res.push(self.get_source(x).await?);
-        }
-
-        Ok(res)
-    }
-
-    async fn get_source(&self, index: i32) -> Result<reply::SourceData, std::io::Error> {
-        let reply = self
-            .query(RequestBody::SourceData(request::SourceData { index }))
-            .await?;
-
-        let sourcedata = match reply.body {
-            ReplyBody::SourceData(sourcedata) => Ok(sourcedata),
-            _ => Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidData,
-                "Invalid response",
-            )),
-        }?;
-
-        // if sourcedata.mode == SourceMode::Ref {
-        //     // Get the name if it's a refclock
-        //     let reply = timeout(
-        //         self.timeout,
-        //         self.client.query(
-        //             RequestBody::NtpSourceName(request::NtpSourceName { ip_addr: sourcedata.ip_addr }),
-        //             self.server,
-        //         ),
-        //     )
-        //     .await??;
-
-        //     let sourcename = match reply.body {
-        //         ReplyBody::NtpSourceName(sourcename) => Ok(sourcename),
-        //         _ => Err(std::io::Error::new(
-        //             std::io::ErrorKind::InvalidData,
-        //             "Invalid response",
-        //         )),
-        //     }?;
-
-        //     sourcedata.ip_addr = sourcename;
-        // }
-
-        Ok(sourcedata)
-    }
-
-    async fn tracking_poll(
-        &self,
-        chan: &ChimemonSourceChannel,
-    ) -> Result<(), Box<dyn std::error::Error>> {
-        let tracking = self.get_tracking().await?;
-
-        let tracking_data = datapoint_from_tracking(&tracking, &self.config)?;
-
-        info!("Sending tracking data");
-
-        chan.send(tracking_data.into())
-            .expect("Unable to send tracking data to targets");
-
-        Ok(())
-    }
-
-    async fn sources_poll(
-        &self,
-        chan: &ChimemonSourceChannel,
-    ) -> Result<(), Box<dyn std::error::Error>> {
-        let sources = self.get_sources().await?;
-        let mut dps = Vec::with_capacity(sources.len());
-        for ds in sources {
-            let source_data = datapoint_from_sourcedata(&ds, &self.config)?;
-            dps.push(source_data);
-        }
-        info!("Sending source data");
-        chan.send(dps.into())
-            .expect("Unable to send source data to targets");
-        Ok(())
-    }
-}
-
-#[async_trait]
-impl ChimemonSource for ChronyClient {
-    async fn run(self, chan: ChimemonSourceChannel) {
-        info!("Chrony task started");
-
-        let mut t_interval = tokio::time::interval(Duration::from_secs(
-            self.config.sources.chrony.tracking_interval,
-        ));
-        let mut s_interval = tokio::time::interval(Duration::from_secs(
-            self.config.sources.chrony.sources_interval,
-        ));
-
-        let t_future = async {
-            let lchan = chan.clone();
-            loop {
-                t_interval.tick().await;
-
-                match self.tracking_poll(&lchan).await {
-                    Ok(_) => (),
-                    Err(e) => {
-                        warn!("Error in chrony task: {}", e.to_string());
-                    }
-                }
-            }
-        };
-        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/hwmon.rs

@@ -1,83 +0,0 @@
-use async_trait::async_trait;
-use chimemon::{ChimemonSource, ChimemonSourceChannel, Config};
-use futures::{stream, StreamExt};
-use influxdb2::models::DataPoint;
-use log::{debug, info};
-use std::{
-    path::PathBuf,
-    time::{Duration, SystemTime, UNIX_EPOCH},
-};
-
-pub struct HwmonSource {
-    config: Config,
-    sensors: Vec<HwmonSensor>,
-}
-
-struct HwmonSensor {
-    path: PathBuf,
-    device: String,
-    sensor: String,
-    name: String,
-}
-
-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)| HwmonSensor {
-                name: k.into(),
-                device: v.name.clone(),
-                sensor: v.sensor.clone(),
-                path: PathBuf::from(HWMON_ROOT).join(&v.name).join(&v.sensor),
-            })
-            .collect();
-
-        HwmonSource { config, sensors }
-    }
-
-    async fn get_raw_value(sensor: &HwmonSensor) -> Result<String, std::io::Error> {
-        tokio::fs::read_to_string(&sensor.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 s = stream::iter(&self.sensors).then(|s| async {
-                let sensor_val = HwmonSource::get_raw_value(s)
-                    .await
-                    .expect("Unable to read sensor");
-                debug!(
-                    "hwmon {} raw value {}",
-                    s.path.to_string_lossy(),
-                    sensor_val
-                );
-                let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
-                let mut builder = DataPoint::builder(&self.config.sources.hwmon.measurement)
-                    .timestamp(now.as_nanos().try_into().unwrap());
-                for (key, value) in &self.config.influxdb.tags {
-                    builder = builder.tag(key, value)
-                }
-                builder = builder
-                    .tag("name", &s.name)
-                    .tag("sensor", &s.sensor)
-                    .tag("device", &s.device)
-                    .field("value", sensor_val.trim().parse::<i64>().unwrap());
-                builder.build().unwrap()
-            });
-            info!("Writing hwmon data");
-            chan.send(s.collect::<Vec<DataPoint>>().await.into())
-                .unwrap();
-        }
-    }
-}

src/lib.rs

@@ -1,19 +1,24 @@
+pub mod sources;
+pub mod targets;
+
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use figment::{
+    Figment,
     providers::{Format, Serialized, Toml},
     util::map,
     value::Map,
-    Figment,
 };
 use gethostname::gethostname;
 use influxdb2::models::DataPoint;
 use serde_derive::{Deserialize, Serialize};
-use std::path::Path;
 use tokio::sync::broadcast::*;
 
+use std::{fmt::Debug, path::Path, sync::Arc};
+
 #[derive(Serialize, Deserialize, Clone)]
 pub struct InfluxConfig {
+    pub enabled: bool,
     pub url: String,
     pub org: String,
     pub bucket: String,
@@ -25,6 +30,7 @@ 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(),
@@ -100,6 +106,47 @@ impl Default for HwmonConfig {
         }
     }
 }
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+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(Serialize, Deserialize, Clone, Debug)]
+pub struct Prs10Config {
+    pub enabled: bool,
+    pub port: String,
+    pub baud: u32,
+    pub timeout: std::time::Duration,
+    pub status_interval: std::time::Duration,
+    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),
+        }
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct TimeReport {
     pub system_time: DateTime<Utc>,
@@ -109,6 +156,67 @@ pub struct TimeReport {
     pub valid: bool,
 }
 
+#[derive(Clone, Debug)]
+pub enum SourceStatus {
+    Healthy,
+    LossOfSignal(Option<String>),
+    LossOfSync(Option<String>),
+    Other(Option<String>),
+    Unknown,
+}
+
+#[derive(Clone, Debug)]
+pub enum MetricValue {
+    Int(i64),
+    Float(f64),
+    Bool(bool),
+}
+
+#[derive(Clone, Debug)]
+pub struct SourceMetric {
+    name: String,
+    value: MetricValue,
+    tags: Arc<Vec<(String, String)>>,
+}
+
+impl SourceMetric {
+    pub fn new_int(name: &str, value: i64, tags: Arc<Vec<(String, String)>>) -> Self {
+        Self {
+            name: name.to_owned(),
+            value: MetricValue::Int(value),
+            tags,
+        }
+    }
+
+    pub fn new_float(name: &str, value: f64, tags: Arc<Vec<(String, String)>>) -> Self {
+        Self {
+            name: name.to_owned(),
+            value: MetricValue::Float(value),
+            tags,
+        }
+    }
+
+    pub fn new_bool(name: &str, value: bool, tags: Arc<Vec<(String, String)>>) -> Self {
+        Self {
+            name: name.to_owned(),
+            value: MetricValue::Bool(value),
+            tags,
+        }
+    }
+}
+
+pub trait SourceReportDetails: Debug + Send + Sync {
+    fn to_metrics(&self) -> Vec<SourceMetric>;
+    fn is_healthy(&self) -> bool;
+}
+
+#[derive(Clone, Debug)]
+pub struct SourceReport {
+    pub name: String,
+    pub status: SourceStatus,
+    pub details: Arc<dyn SourceReportDetails>,
+}
+
 #[derive(Serialize, Deserialize, Clone)]
 pub struct UCCMConfig {
     pub enabled: bool,
@@ -137,6 +245,8 @@ pub struct SourcesConfig {
     pub chrony: ChronyConfig,
     pub hwmon: HwmonConfig,
     pub uccm: UCCMConfig,
+    pub gpsd: GpsdConfig,
+    pub prs10: Prs10Config,
 }
 
 #[derive(Serialize, Deserialize, Clone, Default)]
@@ -155,11 +265,12 @@ pub fn load_config(filename: &Path) -> Figment {
     Figment::from(Serialized::defaults(Config::default())).merge(Toml::file(filename))
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug, Clone)]
 pub enum ChimemonMessage {
     DataPoint(DataPoint),
     DataPoints(Vec<DataPoint>),
     TimeReport(TimeReport),
+    SourceReport(SourceReport),
 }
 
 impl From<DataPoint> for ChimemonMessage {
@@ -179,6 +290,12 @@ impl From<TimeReport> for ChimemonMessage {
     }
 }
 
+impl From<SourceReport> for ChimemonMessage {
+    fn from(sr: SourceReport) -> Self {
+        ChimemonMessage::SourceReport(sr)
+    }
+}
+
 pub type ChimemonSourceChannel = Sender<ChimemonMessage>;
 pub type ChimemonTargetChannel = Receiver<ChimemonMessage>;
 

src/main.rs

@@ -1,20 +1,23 @@
-mod chrony;
-mod chrony_refclock;
-mod hwmon;
-mod uccm;
-
 use clap::{Parser, ValueEnum};
-use env_logger::{self, Env};
 use futures::future::join_all;
-use log::{debug, info};
 use std::path::Path;
 use tokio::sync::broadcast;
+use tracing::{Instrument, debug, error, info, info_span, warn};
+use tracing_subscriber::{
+    self, EnvFilter,
+    fmt::format::{self, FmtSpan},
+    prelude::*,
+};
 
-use crate::{chrony::*, chrony_refclock::ChronySockServer, hwmon::HwmonSource, uccm::UCCMMonitor};
 use chimemon::*;
+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 {
@@ -35,33 +38,55 @@ struct Args {
 
 #[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let args = Args::parse();
+    tracing_subscriber::fmt()
-    let loglevel = args
+        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::default()))
-        .log_level
+        // .event_format(format::Format::default().pretty())
-        .to_possible_value()
+        .with_span_events(FmtSpan::CLOSE)
-        .unwrap()
+        .init();
-        .get_name()
-        .to_owned();
-    env_logger::Builder::from_env(Env::default().default_filter_or(loglevel)).init();
 
-    info!("{} v{} starting...", PROGRAM_NAME, VERSION);
+    let args = Args::parse();
+
+    info!("{PROGRAM_NAME} v{VERSION} starting...");
     let fig = load_config(Path::new(&args.config_file));
-    debug!("{:?}", fig);
+    debug!("{fig:?}");
     let config: Config = fig.extract()?;
 
     let mut tasks = Vec::new();
     let (tx, _) = broadcast::channel(16);
     let sourcechan: ChimemonSourceChannel = tx;
 
-    info!(
+    if config.influxdb.enabled {
-        "Connecting to influxdb {} org: {} using token",
+        info!(
-        &config.influxdb.url, &config.influxdb.org
+            "Connecting to influxdb {} org: {} using token",
-    );
+            &config.influxdb.url, &config.influxdb.org
-    let influx = influxdb2::Client::new(
+        );
-        &config.influxdb.url,
+        let config = config.clone();
-        &config.influxdb.org,
+        let influx = influxdb2::Client::new(
-        &config.influxdb.token,
+            &config.influxdb.url,
-    );
+            &config.influxdb.org,
+            &config.influxdb.token,
+        );
+
+        let mut influxrx = sourcechan.subscribe();
+
+        tasks.push(tokio::spawn(
+            async move {
+                let stream = async_stream::stream! {
+                    while let Ok(msg) = influxrx.recv().await {
+                        match msg { ChimemonMessage::DataPoint(dp) => {
+                            yield dp
+                        }, 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 {
         Some(ChronyClient::new(config.to_owned()))
@@ -69,7 +94,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         None
     };
     if let Some(c) = chrony {
-        tasks.push(tokio::spawn(c.run(sourcechan.clone())));
+        tasks.push(tokio::spawn(
+            c.run(sourcechan.clone())
+                .instrument(info_span!("chrony-task")),
+        ));
     };
 
     let hwmon = if config.sources.hwmon.enabled {
@@ -78,53 +106,92 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         None
     };
     if let Some(hwmon) = hwmon {
-        tasks.push(tokio::spawn(hwmon.run(sourcechan.clone())));
+        tasks.push(tokio::spawn(
+            hwmon
+                .run(sourcechan.clone())
+                .instrument(info_span!("hwmon-task")),
+        ));
     };
 
     let uccm = if config.sources.uccm.enabled {
-        info!("Spawning UCCMMonitor");
         Some(UCCMMonitor::new(config.to_owned()))
     } else {
-        info!("UCCMMonitor not configured");
         None
     };
     if let Some(uccm) = uccm {
-        tasks.push(tokio::spawn(uccm.run(sourcechan.clone())));
+        tasks.push(tokio::spawn(
+            uccm.run(sourcechan.clone())
+                .instrument(info_span!("uccm-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")),
+        ))
+    }
+
+    let prs10 = if config.sources.prs10.enabled {
+        Some(Prs10Monitor::new(config.sources.prs10))
+    } else {
+        None
+    };
+    if let Some(prs10) = prs10 {
+        tasks.push(tokio::spawn(
+            prs10
+                .run(sourcechan.clone())
+                .instrument(info_span!("prs10-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())));
+        tasks.push(tokio::spawn(
+            chrony_refclock
+                .run(sourcechan.subscribe())
+                .instrument(info_span!("chrony-refclock-task")),
+        ));
     };
 
-    let mut influxrx = sourcechan.subscribe();
+    if tasks.len() == 0 {
+        error!("No tasks configured, exiting.");
+        return Ok(()); // not an error, but exit before starting a dummy task
+    }
 
-    tasks.push(tokio::spawn(async move {
+    if sourcechan.receiver_count() == 0 {
-        let stream = async_stream::stream! {
+        warn!("No consumers configured, events will be discarded");
-            while let Ok(msg) = influxrx.recv().await {
+        let mut chan = sourcechan.subscribe();
-                match msg { ChimemonMessage::DataPoint(dp) => {
+        // spawn a dummy task to reap the channel and keep the process alive
-                    yield dp
+        tasks.push(tokio::spawn(
-                }, ChimemonMessage::DataPoints(dps) => {
+            async move {
-                    for p in dps {
+                loop {
-                        yield p
+                    while let Ok(m) = chan.recv().await {
+                        info!("received {m:?}");
+                        match m {
+                            ChimemonMessage::SourceReport(report) => {
+                                let metrics = report.details.to_metrics();
+                                info!("metrics: {metrics:?}");
+                            }
+                            _ => {}
+                        }
                     }
-                }, ChimemonMessage::TimeReport(_tr) => {}
+                }
-            }            }
+            }
-        };
+            .instrument(info_span!("dummy-receiver-task")),
-        influx.write(&config.influxdb.bucket, stream).await.unwrap();
+        ))
-    }));
+    }
 
-    // let mut debugrx = sourcechan.subscribe();
+    debug!("Task setup complete, tasks: {}", tasks.len());
-    // tasks.push(tokio::spawn(async move {
-    //     loop {
-    //         let v = debugrx.recv().await;
-    //         warn!("streamed: {:?}", v.unwrap());
-    //     }
-    // }));
 
     join_all(tasks).await;
 

src/sources/chrony.rs

@@ -0,0 +1,336 @@
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus,
+};
+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 std::net::{SocketAddr, ToSocketAddrs};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::join;
+use tracing::{info, warn};
+
+pub struct ChronyClient {
+    pub server: SocketAddr,
+    client_options: ClientOptions,
+    config: Config,
+}
+
+#[derive(Debug)]
+pub struct ChronyTrackingReport {
+    tags: Arc<Vec<(String, String)>>,
+    pub ref_id: i64,
+    pub ref_ip_addr: String,
+    pub stratum: i64,
+    pub leap_status: i64,
+    pub current_correction: f64,
+    pub last_offset: f64,
+    pub rms_offset: f64,
+    pub freq_ppm: f64,
+    pub resid_freq_ppm: f64,
+    pub skew_ppm: f64,
+    pub root_delay: f64,
+    pub root_dispersion: f64,
+    pub last_update_interval: f64,
+}
+
+impl SourceReportDetails for ChronyTrackingReport {
+    fn is_healthy(&self) -> bool {
+        true
+    }
+    fn to_metrics(&self) -> Vec<SourceMetric> {
+        let tags = &self.tags;
+        vec![
+            SourceMetric::new_int("ref_id", self.ref_id, tags.clone()),
+            SourceMetric::new_int("stratum", self.stratum, tags.clone()),
+            SourceMetric::new_int("leap_status", self.leap_status, tags.clone()),
+            SourceMetric::new_float("current_correction", self.current_correction, tags.clone()),
+            SourceMetric::new_float("last_offset", self.last_offset, tags.clone()),
+            SourceMetric::new_float("rms_offset", self.rms_offset, tags.clone()),
+            SourceMetric::new_float("freq_ppm", self.freq_ppm, tags.clone()),
+            SourceMetric::new_float("resid_freq_ppm", self.resid_freq_ppm, tags.clone()),
+            SourceMetric::new_float("skew_ppm", self.skew_ppm, tags.clone()),
+            SourceMetric::new_float("root_delay", self.root_delay, tags.clone()),
+            SourceMetric::new_float("root_dispersion", self.root_dispersion, tags.clone()),
+            SourceMetric::new_float(
+                "last_update_interval",
+                self.last_update_interval,
+                tags.clone(),
+            ),
+        ]
+    }
+}
+
+#[derive(Debug)]
+pub struct ChronySourcesReport {
+    pub sources: Vec<reply::SourceData>,
+}
+
+impl SourceReportDetails for ChronySourcesReport {
+    fn is_healthy(&self) -> bool {
+        //TODO: think about whether there is an idea of unhealthy sources
+        true
+    }
+    fn to_metrics(&self) -> Vec<SourceMetric> {
+        let mut metrics = Vec::with_capacity(8 * self.sources.len());
+
+        for source in &self.sources {
+            let tags = Arc::new(vec![
+                ("ref_id".to_owned(), source.ip_addr.to_string()),
+                (
+                    "mode".to_owned(),
+                    match source.mode {
+                        SourceMode::Client => String::from("server"),
+                        SourceMode::Peer => String::from("peer"),
+                        SourceMode::Ref => String::from("refclock"),
+                    },
+                ),
+                (
+                    "state".to_owned(),
+                    match source.state {
+                        reply::SourceState::Selected => String::from("best"),
+                        reply::SourceState::NonSelectable => String::from("unusable"),
+                        reply::SourceState::Falseticker => String::from("falseticker"),
+                        reply::SourceState::Jittery => String::from("jittery"),
+                        reply::SourceState::Unselected => String::from("combined"),
+                        reply::SourceState::Selectable => String::from("unused"),
+                    },
+                ),
+            ]);
+            metrics.extend([
+                SourceMetric::new_int("poll", source.poll as i64, tags.clone()),
+                SourceMetric::new_int("stratum", source.stratum as i64, tags.clone()),
+                SourceMetric::new_int("flags", source.flags.bits() as i64, tags.clone()),
+                SourceMetric::new_int(
+                    "reachability",
+                    source.reachability.count_ones() as i64,
+                    tags.clone(),
+                ),
+                SourceMetric::new_int("since_sample", source.since_sample as i64, tags.clone()),
+                SourceMetric::new_float(
+                    "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
+    }
+}
+
+fn report_from_tracking(
+    t: &reply::Tracking,
+    config: &Config,
+) -> Result<ChronyTrackingReport, Box<dyn std::error::Error>> {
+    let report = ChronyTrackingReport {
+        tags: Arc::new(vec![]), //TODO: allow configuring tags in the source
+        ref_id: t.ref_id as i64,
+        ref_ip_addr: t.ip_addr.to_string(),
+        stratum: t.stratum as i64,
+        leap_status: t.leap_status as i64,
+        current_correction: t.current_correction.into(),
+        last_offset: t.last_offset.into(),
+        rms_offset: t.rms_offset.into(),
+        freq_ppm: t.freq_ppm.into(),
+        resid_freq_ppm: t.resid_freq_ppm.into(),
+        skew_ppm: t.skew_ppm.into(),
+        root_delay: t.root_delay.into(),
+        root_dispersion: t.root_dispersion.into(),
+        last_update_interval: t.last_update_interval.into(),
+    };
+    Ok(report)
+}
+
+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;
+        tokio::task::spawn_blocking(move || blocking_query(request, client_options, &server))
+            .await
+            .map_err(|e| {
+                std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    format!("Error joining thread: {}", e),
+                )
+            })?
+    }
+
+    pub async fn get_tracking(&self) -> Result<reply::Tracking, std::io::Error> {
+        let reply = self.query(RequestBody::Tracking).await?;
+
+        match reply.body {
+            ReplyBody::Tracking(tracking) => Ok(tracking),
+            _ => Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "Unexpected response type",
+            )),
+        }
+    }
+
+    pub async fn get_sources(&self) -> Result<Vec<reply::SourceData>, std::io::Error> {
+        let reply = self.query(RequestBody::NSources).await?;
+
+        let nsources = match reply.body {
+            ReplyBody::NSources(ns) => Ok(i32::try_from(ns.n_sources).unwrap()),
+            _ => Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "Unexpected response type",
+            )),
+        }?;
+
+        let mut res = Vec::with_capacity(
+            nsources
+                .try_into()
+                .expect("Ridiculously unconvertible number of sources"),
+        );
+
+        for x in 0..nsources {
+            res.push(self.get_source(x).await?);
+        }
+
+        Ok(res)
+    }
+
+    async fn get_source(&self, index: i32) -> Result<reply::SourceData, std::io::Error> {
+        let reply = self
+            .query(RequestBody::SourceData(request::SourceData { index }))
+            .await?;
+
+        let sourcedata = match reply.body {
+            ReplyBody::SourceData(sourcedata) => Ok(sourcedata),
+            _ => Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "Invalid response",
+            )),
+        }?;
+
+        // if sourcedata.mode == SourceMode::Ref {
+        //     // Get the name if it's a refclock
+        //     let reply = timeout(
+        //         self.timeout,
+        //         self.client.query(
+        //             RequestBody::NtpSourceName(request::NtpSourceName { ip_addr: sourcedata.ip_addr }),
+        //             self.server,
+        //         ),
+        //     )
+        //     .await??;
+
+        //     let sourcename = match reply.body {
+        //         ReplyBody::NtpSourceName(sourcename) => Ok(sourcename),
+        //         _ => Err(std::io::Error::new(
+        //             std::io::ErrorKind::InvalidData,
+        //             "Invalid response",
+        //         )),
+        //     }?;
+
+        //     sourcedata.ip_addr = sourcename;
+        // }
+
+        Ok(sourcedata)
+    }
+
+    async fn tracking_poll(
+        &self,
+        chan: &ChimemonSourceChannel,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let tracking = self.get_tracking().await?;
+
+        let tracking_data = report_from_tracking(&tracking, &self.config)?;
+        let report = SourceReport {
+            name: "chrony-tracking".to_owned(),
+            status: SourceStatus::Unknown,
+            details: Arc::new(tracking_data),
+        };
+
+        info!("Sending tracking data");
+
+        chan.send(report.into())?;
+        Ok(())
+    }
+
+    async fn sources_poll(
+        &self,
+        chan: &ChimemonSourceChannel,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let sources = self.get_sources().await?;
+        let details = ChronySourcesReport { sources };
+        let report = SourceReport {
+            name: "chrony-sources".to_owned(),
+            status: SourceStatus::Unknown,
+            details: Arc::new(details),
+        };
+        info!("Sending source data");
+        chan.send(report.into())?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl ChimemonSource for ChronyClient {
+    async fn run(self, chan: ChimemonSourceChannel) {
+        info!("Chrony task started");
+
+        let mut t_interval = tokio::time::interval(Duration::from_secs(
+            self.config.sources.chrony.tracking_interval,
+        ));
+        let mut s_interval = tokio::time::interval(Duration::from_secs(
+            self.config.sources.chrony.sources_interval,
+        ));
+
+        let t_future = async {
+            let lchan = chan.clone();
+            loop {
+                t_interval.tick().await;
+
+                match self.tracking_poll(&lchan).await {
+                    Ok(_) => (),
+                    Err(e) => {
+                        warn!("Error in chrony task: {}", e.to_string());
+                    }
+                }
+            }
+        };
+        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

@@ -0,0 +1,453 @@
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport,
+    SourceReportDetails, SourceStatus,
+};
+
+use std::collections::HashMap;
+use std::f64;
+use std::fmt::Debug;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use backoff::ExponentialBackoff;
+use futures::StreamExt;
+use futures::{SinkExt, Stream};
+use gpsd_proto::{Device, Gst, Mode, Pps, Sky, Tpv, UnifiedResponse, Version};
+use serde::Serialize;
+use serde_json;
+use tokio::net::{TcpStream, ToSocketAddrs, lookup_host};
+use tokio::time::{interval, timeout};
+use tokio_util::codec::{Framed, LinesCodec};
+use tracing::{debug, debug_span, info, instrument, warn};
+
+pub struct GpsdSource {
+    pub config: Config,
+    conn: GpsdTransport,
+    devices: HashMap<String, Device>,
+    last_gst: Option<Gst>,
+    last_pps: Option<Pps>,
+    last_tpv: Option<Tpv>,
+    last_sky: Option<Sky>,
+}
+
+#[derive(Eq, PartialEq, Clone, Copy, Debug)]
+pub enum GpsdFixType {
+    Unknown,
+    NoFix,
+    Fix2D,
+    Fix3D,
+    Surveyed,
+}
+
+impl From<u32> for GpsdFixType {
+    fn from(value: u32) -> Self {
+        match value {
+            0 => Self::Unknown,
+            1 => Self::NoFix,
+            2 => Self::Fix2D,
+            3 => Self::Fix3D,
+            _ => panic!("Invalid fix mode {value}"),
+        }
+    }
+}
+
+impl From<Mode> for GpsdFixType {
+    fn from(value: Mode) -> Self {
+        match value {
+            Mode::NoFix => Self::NoFix,
+            Mode::Fix2d => Self::Fix2D,
+            Mode::Fix3d => Self::Fix3D,
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct GpsdSourceReport {
+    fix_type: GpsdFixType,
+    sats_visible: u8,
+    sats_tracked: u8,
+    tdop: f64,
+}
+
+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> {
+        let no_tags = Arc::new(vec![]);
+        vec![
+            SourceMetric::new_int("sats_visible", self.sats_visible as i64, no_tags.clone()),
+            SourceMetric::new_int("sats_tracked", self.sats_tracked as i64, no_tags.clone()),
+            SourceMetric::new_float("tdop", self.tdop, no_tags.clone()),
+        ]
+    }
+}
+
+impl GpsdSource {
+    pub async fn new(config: Config) -> Result<Self, std::io::Error> {
+        let conn = GpsdTransport::new(&config.sources.gpsd.host).await?;
+        Ok(Self {
+            config,
+            conn,
+            devices: HashMap::new(),
+            last_gst: None,
+            last_pps: None,
+            last_tpv: None,
+            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();
+
+        let (sats_tracked, sats_visible) = sky.map_or((0, 0), |sky| {
+            let sats = sky.satellites.as_deref().unwrap_or_default();
+            (
+                sats.iter().filter(|s| s.used).count() as u8,
+                sats.len() as u8,
+            )
+        });
+
+        let tdop = sky
+            .and_then(|sky| sky.tdop)
+            .map_or(f64::INFINITY, |tdop| tdop as f64);
+
+        chan.send(ChimemonMessage::SourceReport(SourceReport {
+            name: "gpsd".into(),
+            status: SourceStatus::Unknown,
+            details: Arc::new(GpsdSourceReport {
+                fix_type: tpv.map_or(GpsdFixType::Unknown, |tpv| tpv.mode.into()),
+                sats_tracked,
+                sats_visible,
+                tdop,
+            }),
+        }))
+        .unwrap();
+    }
+
+    fn handle_msg(&mut self, msg: String) -> Result<(), Box<dyn std::error::Error>> {
+        let _span = debug_span!("handle_msg").entered();
+        let parsed = serde_json::from_str::<UnifiedResponse>(&msg)?;
+        debug!("Received {parsed:?}");
+        match parsed {
+            UnifiedResponse::Device(d) => {
+                if let Some(path) = &d.path {
+                    self.devices.insert(path.to_owned(), d);
+                } else {
+                    warn!("No path on DEVICE response, ignoring.");
+                }
+            }
+            UnifiedResponse::Gst(g) => self.last_gst = Some(g),
+            UnifiedResponse::Pps(p) => self.last_pps = Some(p),
+            UnifiedResponse::Sky(s) => {
+                self.last_sky = Some({
+                    let mut s = s;
+                    if s.satellites.is_none() {
+                        s.satellites = self.last_sky.as_mut().and_then(|old| old.satellites.take());
+                    }
+                    s
+                })
+            }
+            UnifiedResponse::Tpv(t) => self.last_tpv = Some(t),
+            _ => warn!("Unhandled response `{parsed:?}`"),
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl ChimemonSource for GpsdSource {
+    async fn run(mut self, mut chan: ChimemonSourceChannel) {
+        info!("gpsd task started");
+        self.conn.conn().await.unwrap();
+        let mut ticker = interval(Duration::from_secs(self.config.sources.gpsd.interval));
+
+        let mut params = WatchParams::default();
+        params.json = Some(true);
+        self.conn
+            .cmd_response(&GpsdCommand::Watch(Some(params)))
+            .await
+            .unwrap();
+        loop {
+            let framed = self.conn.framed.as_mut().expect("must be connected");
+            tokio::select! {
+                _ = ticker.tick() => {
+                    self.send_status(&mut chan).await
+                },
+                maybe_msg = framed.next() => {
+                    if let Some(Ok(msg)) = maybe_msg {
+                        self.handle_msg(msg).unwrap()
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+struct GpsdTransport {
+    host: SocketAddr,
+    framed: Option<Framed<TcpStream, LinesCodec>>,
+    conn_backoff: ExponentialBackoff,
+}
+
+#[derive(Clone, Copy, Debug, Serialize)]
+#[repr(u8)]
+pub enum RawMode {
+    Off = 0,
+    RawHex = 1,
+    RawBin = 2,
+}
+
+#[derive(Clone, Copy, Debug, Serialize)]
+#[repr(u8)]
+pub enum NativeMode {
+    Nmea = 0,
+    Alt = 1,
+}
+
+#[derive(Clone, Copy, Debug, Serialize)]
+#[repr(u8)]
+pub enum ParityMode {
+    None = b'N',
+    Odd = b'O',
+    Even = b'E',
+}
+
+#[derive(Clone, Debug, Serialize)]
+struct WatchParams {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    enable: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    json: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    nmea: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    raw: Option<RawMode>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    scaled: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    split24: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pps: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    device: Option<String>,
+}
+
+impl Default for WatchParams {
+    fn default() -> Self {
+        WatchParams {
+            enable: Some(true),
+            json: Some(false),
+            nmea: None,
+            raw: None,
+            scaled: None,
+            split24: None,
+            pps: None,
+            device: None,
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize)]
+struct DeviceParams {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    bps: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    cycle: Option<f64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    flags: Option<u8>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    hexdata: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    native: Option<NativeMode>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    parity: Option<ParityMode>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    path: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    readonly: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    sernum: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    stopbits: Option<u8>,
+}
+
+#[derive(Clone, Debug)]
+enum GpsdCommand {
+    Version, // no params
+    Devices, // no params
+    Watch(Option<WatchParams>),
+    Poll, // I don't understand the protocol for this one
+    Device(Option<DeviceParams>),
+}
+
+impl GpsdCommand {
+    fn command_string(&self) -> &str {
+        match self {
+            Self::Version => "?VERSION",
+            Self::Devices => "?DEVICES",
+            Self::Watch(_) => "?WATCH",
+            Self::Poll => "?POLL",
+            Self::Device(_) => "?DEVICE",
+        }
+    }
+
+    fn expected_responses(&self) -> usize {
+        match self {
+            Self::Version => 1,
+            Self::Devices => 1,
+            Self::Watch(_) => 2,
+            Self::Poll => 1,
+            Self::Device(_) => 1,
+        }
+    }
+}
+
+impl ToString for GpsdCommand {
+    fn to_string(&self) -> String {
+        let s = self.command_string().to_owned();
+        match self {
+            Self::Version | Self::Devices | Self::Poll | Self::Watch(None) | Self::Device(None) => {
+                s + ";"
+            }
+            Self::Watch(Some(w)) => s + "=" + &serde_json::to_string(w).unwrap() + ";",
+            Self::Device(Some(d)) => s + "=" + &serde_json::to_string(d).unwrap() + ";",
+        }
+    }
+}
+
+impl GpsdTransport {
+    async fn new<T: ToSocketAddrs + Debug>(host: &T) -> Result<Self, std::io::Error> {
+        // TODO: implement proper handling of multiple responses
+        let host_addr = lookup_host(host).await?.next().ok_or(std::io::Error::new(
+            std::io::ErrorKind::NotFound,
+            format!("No response looking up `{:?}`", host),
+        ))?;
+        Ok(Self {
+            host: host_addr,
+            framed: None,
+            conn_backoff: ExponentialBackoff::default(),
+        })
+    }
+    async fn connect_inner(
+        &self,
+    ) -> Result<Framed<TcpStream, LinesCodec>, Box<dyn std::error::Error>> {
+        info!("Connecting to gpsd @ {}", self.host);
+
+        let mut framed = Framed::new(TcpStream::connect(self.host).await?, LinesCodec::new());
+
+        debug!("Waiting for initial VERSION");
+
+        match timeout(Duration::from_secs(5), framed.next()).await {
+            Ok(Some(Ok(r))) => {
+                if let Ok(version) = serde_json::from_str::<Version>(&r) {
+                    info!(
+                        "Connected to gpsd @ {}, release {}",
+                        self.host, version.release
+                    );
+                } 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(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>> {
+        let mut responses = Vec::new();
+        let conn = self.conn().await?;
+
+        conn.send(cmd.to_string()).await?;
+        for _ in 0..cmd.expected_responses() {
+            match conn.next().await {
+                None => return Err("Connection lost".into()),
+                Some(Err(e)) => return Err(format!("Unable to parse response {e}").into()),
+                Some(Ok(r)) => {
+                    debug!("Raw response: `{r}`");
+                    responses.push(serde_json::from_str::<UnifiedResponse>(&r)?)
+                }
+            }
+        }
+
+        Ok(responses)
+    }
+
+    async fn stream(
+        &mut self,
+    ) -> Result<
+        impl Stream<Item = Result<UnifiedResponse, Box<dyn std::error::Error>>>,
+        Box<dyn std::error::Error>,
+    > {
+        Ok(self
+            .conn()
+            .await?
+            .map(|line| Ok(serde_json::from_str::<UnifiedResponse>(&line?)?)))
+    }
+}
+
+mod tests {
+    use gpsd_proto::{ResponseData, UnifiedResponse};
+    use tokio_stream::StreamExt;
+
+    use crate::sources::gpsd::{GpsdCommand, GpsdTransport, WatchParams};
+    use std::sync::Once;
+
+    static INIT: Once = Once::new();
+
+    fn init_logger() {
+        INIT.call_once(|| tracing_subscriber::fmt::init());
+    }
+
+    #[tokio::test]
+    async fn test_gpsd() {
+        init_logger();
+        let mut gpsd = GpsdTransport::new(&"192.168.65.93:2947").await.unwrap();
+        gpsd.conn().await.unwrap();
+        let mut params = WatchParams::default();
+        params.enable = Some(true);
+        params.json = Some(true);
+        let res = {
+            gpsd.cmd_response(&GpsdCommand::Watch(Some(params)))
+                .await
+                .unwrap()
+        };
+        println!("{res:?}");
+        while let Some(Ok(s)) = gpsd.framed.as_mut().unwrap().next().await {
+            let res2 = serde_json::from_str::<UnifiedResponse>(&s);
+            println!("{res2:?}");
+        }
+    }
+}

src/sources/hwmon.rs

@@ -0,0 +1,154 @@
+use crate::{
+    ChimemonSource, ChimemonSourceChannel, Config, SourceMetric, SourceReport, SourceReportDetails,
+    SourceStatus,
+};
+use async_trait::async_trait;
+use std::{fs::File, io::Read, path::PathBuf, sync::Arc, time::Duration};
+use tracing::{debug, error, info, 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/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,570 @@
+use std::str::FromStr;
+use std::sync::Arc;
+
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Prs10Config, SourceMetric,
+    SourceReport, SourceReportDetails, SourceStatus,
+};
+use async_trait::async_trait;
+use bitflags::bitflags;
+use itertools::Itertools;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio::select;
+use tokio::sync::OnceCell;
+use tokio::time::{interval, timeout};
+use tokio_serial;
+use tokio_serial::{SerialPort, SerialStream};
+use tracing::{debug, debug_span, error, info, instrument, warn};
+
+#[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, no_tags: Arc<Vec<(String, String)>>) -> 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), no_tags.clone())
+            })
+            .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);
+    }
+}
+
+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<SourceMetric> {
+        let no_tags = Arc::new(vec![]);
+        vec![
+            SourceMetric::new_int(
+                "volt_lamp_flags",
+                self.volt_lamp_flags.bits() as i64,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_int("rf_flags", self.rf_flags.bits() as i64, no_tags.clone()),
+            SourceMetric::new_int("temp_flags", self.temp_flags.bits() as i64, no_tags.clone()),
+            SourceMetric::new_int("fll_flags", self.fll_flags.bits() as i64, no_tags.clone()),
+            SourceMetric::new_int("pps_flags", self.pps_flags.bits() as i64, no_tags.clone()),
+            // 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")?;
+        Ok(Self {
+            volt_lamp_flags: Prs10PowerLampFlags::from_bits(volt_lamp_flags?)
+                .ok_or("Invalid bits set ({volt_lamp_flags}) for power/lamp flags")?,
+            rf_flags: Prs10RfFlags::from_bits(rf_flags?)
+                .ok_or("Invalid bits set ({rf_flags}) for RF flags")?,
+            temp_flags: Prs10TempFlags::from_bits(temp_flags?)
+                .ok_or("Invalid bits set ({temp_flags}) for temp flags")?,
+            fll_flags: Prs10FllFlags::from_bits(fll_flags?)
+                .ok_or("Invalid bits set ({fll_flags}) for FLL flags")?,
+            pps_flags: Prs10PpsFlags::from_bits(pps_flags?)
+                .ok_or("Invalid bits set ({pps_flags}) for PPS flags")?,
+            system_flags: Prs10SystemFlags::from_bits(system_flags?)
+                .ok_or("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: u16,
+    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<SourceMetric> {
+        let no_tags = Arc::new(vec![]);
+        vec![
+            // Integer Metrics
+            SourceMetric::new_int("ocxo_efc", self.ocxo_efc as i64, no_tags.clone()),
+            // Float Metrics
+            SourceMetric::new_float(
+                "error_signal_volts",
+                self.error_signal_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float(
+                "detect_signal_volts",
+                self.detect_signal_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float("heat_volts", self.heat_volts, no_tags.clone()),
+            SourceMetric::new_float("elec_volts", self.elec_volts, no_tags.clone()),
+            SourceMetric::new_float(
+                "lamp_fet_drain_volts",
+                self.lamp_fet_drain_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float(
+                "lamp_fet_gate_volts",
+                self.lamp_fet_gate_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float("ocxo_heat_volts", self.ocxo_heat_volts, no_tags.clone()),
+            SourceMetric::new_float("cell_heat_volts", self.cell_heat_volts, no_tags.clone()),
+            SourceMetric::new_float("lamp_heat_volts", self.lamp_heat_volts, no_tags.clone()),
+            SourceMetric::new_float("rb_photo", self.rb_photo, no_tags.clone()),
+            SourceMetric::new_float("rb_photo_iv", self.rb_photo_iv, no_tags.clone()),
+            SourceMetric::new_float("case_temp", self.case_temp, no_tags.clone()),
+            SourceMetric::new_float("ocxo_therm", self.ocxo_therm, no_tags.clone()),
+            SourceMetric::new_float("cell_therm", self.cell_therm, no_tags.clone()),
+            SourceMetric::new_float("lamp_therm", self.lamp_therm, no_tags.clone()),
+            SourceMetric::new_float("ext_cal_volts", self.ext_cal_volts, no_tags.clone()),
+            SourceMetric::new_float("analog_gnd_volts", self.analog_gnd_volts, no_tags.clone()),
+            SourceMetric::new_float(
+                "if_vco_varactor_volts",
+                self.if_vco_varactor_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float(
+                "op_vco_varactor_volts",
+                self.op_vco_varactor_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float(
+                "mul_amp_gain_volts",
+                self.mul_amp_gain_volts,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_float("rf_lock_volts", self.rf_lock_volts, no_tags.clone()),
+            // U16 Metrics (optional, but can be treated as integers)
+            SourceMetric::new_int(
+                "freq_offset_ppt",
+                self.freq_offset_ppt as i64,
+                no_tags.clone(),
+            ),
+            SourceMetric::new_int("mag_efc", self.mag_efc as i64, no_tags.clone()),
+        ]
+    }
+}
+
+#[derive(Debug)]
+pub struct Prs10Monitor {
+    rx: ReadHalf<SerialStream>,
+    tx: WriteHalf<SerialStream>,
+    info: OnceCell<Prs10Info>,
+    config: Prs10Config,
+}
+
+impl Prs10Monitor {
+    pub fn new(config: Prs10Config) -> 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).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);
+
+        Self {
+            rx,
+            tx,
+            config,
+            info: OnceCell::new(),
+        }
+    }
+
+    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.unwrap();
+        self.tx.write_u8(b'\r').await.unwrap();
+        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!("response: ({read}) `{buf:?}`");
+        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>> {
+        debug!("Getting analog value {id}");
+        let mut cmd = b"AD".to_vec();
+        cmd.extend_from_slice(id.to_string().as_bytes());
+        cmd.push(b'?');
+        let resp = self.cmd_response(&cmd).await?;
+        let value = str::from_utf8(&resp)?.parse::<f64>()?;
+
+        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 int value for command {cmd:?}");
+        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 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::<u16>())
+            .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))
+    }
+
+    async fn status_poll(&mut self) -> Result<ChimemonMessage, Box<dyn std::error::Error>> {
+        debug!("polling status");
+        let status = self.get_status().await?;
+        Ok(ChimemonMessage::SourceReport(SourceReport {
+            name: "prs10".into(),
+            status: if status.is_healthy() {
+                SourceStatus::Healthy
+            } else {
+                SourceStatus::Unknown
+            },
+            details: Arc::new(status),
+        }))
+    }
+
+    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,
+        ];
+
+        debug!("polling stats");
+
+        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(ChimemonMessage::SourceReport(SourceReport {
+            name: "prs10".into(),
+            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],
+            }),
+        }))
+    }
+}
+
+#[async_trait]
+impl ChimemonSource for Prs10Monitor {
+    async fn run(mut self, chan: ChimemonSourceChannel) {
+        info!("PRS10 task starting");
+        if let Err(e) = self.set_info().await {
+            warn!("Error starting PRS10: {e:?}");
+            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! {
+                _ = 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,14 +1,16 @@
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, SourceMetric,
+    SourceReportDetails, TimeReport,
+};
 use async_trait::async_trait;
 use bitflags::bitflags;
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, BytesMut};
-use chimemon::{ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Config, TimeReport};
 use chrono::{DateTime, Duration, NaiveDateTime, Utc};
 use figment::value::Map;
-use influxdb2::models::data_point::DataPointBuilder;
 use influxdb2::models::DataPoint;
+use influxdb2::models::data_point::DataPointBuilder;
 use itertools::Itertools;
-use log::{debug, info, warn};
 use std::io::{BufRead, Cursor};
 use std::str;
 use std::sync::Arc;
@@ -17,6 +19,7 @@ use tokio::join;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use tokio_serial::{SerialPort, SerialStream};
+use tracing::{debug, info, warn};
 
 pub const GPS_EPOCH: i64 = 315964800; // Doesn't seem possible to have a const DateTime object
 pub type UccmEndian = BigEndian;
@@ -43,6 +46,77 @@ pub struct UCCMTODReport {
     pub flags: UCCMFlags,
 }
 
+impl SourceReportDetails for UCCMTODReport {
+    fn is_healthy(&self) -> bool {
+        self.flags.contains(UCCMFlags::OSC_LOCK)
+            && self.flags.contains(UCCMFlags::HAVE_GPS_TIME)
+            && !self.flags.contains(UCCMFlags::INIT_UNLOCK)
+            && !self.flags.contains(UCCMFlags::INIT_NO_SATS)
+            && !self.flags.contains(UCCMFlags::POWER_FAIL)
+            && !self.flags.contains(UCCMFlags::NO_GPS_SYNC)
+            && !self.flags.contains(UCCMFlags::NO_GPS_SYNC2)
+            && !self.flags.contains(UCCMFlags::NO_ANT)
+            && !self.flags.contains(UCCMFlags::GPS_LOS)
+    }
+
+    fn to_metrics(&self) -> Vec<SourceMetric> {
+        let no_tags = Arc::new(vec![]);
+        vec![
+            SourceMetric::new_int("leaps", self.leaps as i64, no_tags.clone()),
+            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),
+                no_tags.clone(),
+            ),
+            SourceMetric::new_bool(
+                "init_unlock",
+                self.flags.contains(UCCMFlags::INIT_UNLOCK),
+                no_tags.clone(),
+            ),
+            SourceMetric::new_bool(
+                "init_no_sats",
+                self.flags.contains(UCCMFlags::INIT_NO_SATS),
+                no_tags.clone(),
+            ),
+            SourceMetric::new_bool(
+                "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 =
@@ -121,6 +195,7 @@ impl UCCMGPSSatsReport {
 }
 
 bitflags! {
+    #[derive(Debug)]
     pub struct UCCMFlags: u32 {
         const OSC_LOCK = (1<<29);
         const LEAP_FLAG = (1<<25);
@@ -194,6 +269,27 @@ pub struct UCCMStatusReport {
     pub freq_error: f32,
 }
 
+impl SourceReportDetails for UCCMStatusReport {
+    fn is_healthy(&self) -> bool {
+        self.gps_pps_valid
+    }
+    fn to_metrics(&self) -> Vec<SourceMetric> {
+        let no_tags = Arc::new(vec![]);
+        vec![
+            SourceMetric::new_int("tfom", self.tfom as i64, no_tags.clone()),
+            SourceMetric::new_int("ffom", self.ffom as i64, no_tags.clone()),
+            SourceMetric::new_float("gps_phase", self.gps_phase as f64, no_tags.clone()),
+            // TODO: sv info
+            // TOOD: timestamp
+            SourceMetric::new_float("ant_voltage", self.ant_voltage as f64, no_tags.clone()),
+            SourceMetric::new_float("ant_current", self.ant_current as f64, no_tags.clone()),
+            SourceMetric::new_float("temp", self.temp as f64, no_tags.clone()),
+            SourceMetric::new_int("efc_dac", self.efc_dac as i64, no_tags.clone()),
+            SourceMetric::new_float("freq_error", self.freq_error as f64, no_tags.clone()),
+        ]
+    }
+}
+
 pub struct UCCMInfo {
     pub vendor: String,
     pub model: String,
@@ -257,7 +353,8 @@ impl TryFrom<&str> for UCCMLoopDiagReport {
                 "No lines!",
             ))??;
             let ocxo_val = ocxo_line
-                .split(':').nth(1)
+                .split(':')
+                .nth(1)
                 .ok_or(std::io::Error::new(
                     std::io::ErrorKind::InvalidData,
                     "no colon!",
@@ -280,7 +377,10 @@ impl TryFrom<&str> for UCCMGPSSatsReport {
             "Invalid response (expected `NSATS CNOS`)",
         ))?;
         let nsats = nsats.parse::<u8>().map_err(|e| {
-            std::io::Error::new(std::io::ErrorKind::InvalidData, format!("Invalid number of sats ({e})"))
+            std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!("Invalid number of sats ({e})"),
+            )
         })?;
         let tracked_svs = cnos
             .split(',')
@@ -428,13 +528,14 @@ impl UCCMMonitor {
                                 >= Duration::from_std(self.config.sources.uccm.status_interval)
                                     .unwrap()
                             {
-                                let mut points = vec![tod
+                                let mut points = vec![
-                                    .as_builder(
+                                    tod.as_builder(
                                         &self.config.sources.uccm.measurement,
                                         &self.config.influxdb.tags,
                                     )
                                     .build()
-                                    .unwrap()];
+                                    .unwrap(),
+                                ];
                                 if let Some(loop_diag) = &last_loop_diag {
                                     points.push(
                                         loop_diag

src/targets/chrony_refclock.rs

@@ -1,10 +1,10 @@
+use crate::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, ChronySockConfig};
 use async_trait::async_trait;
-use chimemon::{ChimemonMessage, ChimemonTarget, ChimemonTargetChannel, ChronySockConfig};
 use libc::{c_double, c_int, timeval};
-use log::debug;
 use std::mem;
 use std::os::unix::net::UnixDatagram;
 use std::path::PathBuf;
+use tracing::debug;
 
 const CHRONY_MAGIC: c_int = 0x534f434b;
 

src/targets/mod.rs

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