ktims/chimemon
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: ktims:main
Branches Tags
ktims:main ktims:gpsd
..
compare: ktims:430c1acdef3f10c21d6197cd67f6e936589a126e
Branches Tags
ktims:gpsd ktims:main

6 Commits
main ... 430c1acdef

Author SHA1 Message Date
Keenan Tims
430c1acdef incremental improvements 2026-02-02 01:37:47 -08:00
Keenan Tims
7717aa9177 refactor, prs10 stats, etc 2026-02-02 00:41:19 -08:00
Keenan Tims
7f24bf5a91 prs10: intermediate work 2026-02-01 20:06:25 -08:00
Keenan Tims
adbe09b9d2 prs10: intermediate work 2026-02-01 20:06:17 -08:00
Keenan Tims
ea412f4a66 uccm: source report 2026-02-01 15:40:27 -08:00
Keenan Tims
4dabcbe985 refactoring, log->tracing, gpsd source 2026-01-03 15:38:21 -08:00
15 changed files with 2629 additions and 1004 deletions
Expand all files Collapse all files

1299
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

26
Cargo.toml
View File

@@ -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"

9
config.toml.dist
View File

@@ -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

276
src/chrony.rs
View File

@@ -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);
}
}

83
src/hwmon.rs
View File

@@ -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();
}
}
}

123
src/lib.rs
View File

@@ -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>;

177
src/main.rs
View File

@@ -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();
let loglevel = args
.log_level
.to_possible_value()
.unwrap()
.get_name()
.to_owned();
env_logger::Builder::from_env(Env::default().default_filter_or(loglevel)).init();
tracing_subscriber::fmt()
.with_env_filter(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::default()))
// .event_format(format::Format::default().pretty())
.with_span_events(FmtSpan::CLOSE)
.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!(
"Connecting to influxdb {} org: {} using token",
&config.influxdb.url, &config.influxdb.org
);
let influx = influxdb2::Client::new(
&config.influxdb.url,
&config.influxdb.org,
&config.influxdb.token,
);
if config.influxdb.enabled {
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();
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 {
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
if sourcechan.receiver_count() == 0 {
warn!("No consumers configured, events will be discarded");
let mut chan = sourcechan.subscribe();
// spawn a dummy task to reap the channel and keep the process alive
tasks.push(tokio::spawn(
async move {
loop {
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) => {}
} }
};
influx.write(&config.influxdb.bucket, stream).await.unwrap();
}));
}
}
.instrument(info_span!("dummy-receiver-task")),
))
}
// let mut debugrx = sourcechan.subscribe();
// tasks.push(tokio::spawn(async move {
// loop {
// let v = debugrx.recv().await;
// warn!("streamed: {:?}", v.unwrap());
// }
// }));
debug!("Task setup complete, tasks: {}", tasks.len());
join_all(tasks).await;

336
src/sources/chrony.rs Normal file
View File

@@ -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);
}
}

453
src/sources/gpsd.rs Normal file
View File

@@ -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:?}");
}
}
}

154
src/sources/hwmon.rs Normal file
View File

@@ -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})")
}
}
}
}
}

5
src/sources/mod.rs Normal file
View File

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

570
src/sources/prs10.rs Normal file
View File

@@ -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(())
}
}

117
src/uccm.rs → src/sources/uccm.rs
View File

@@ -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
.as_builder(
let mut points = vec![
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

4
src/chrony_refclock.rs → src/targets/chrony_refclock.rs
View File

@@ -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;

1
src/targets/mod.rs Normal file
View File

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