prs10: intermediate work

Cargo.lock

@@ -195,6 +195,15 @@ version = "0.21.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"
 
+[[package]]
+name = "bit-struct"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "270fbbb014407467f7a2c9b1fa0b74057d5cbc452f18bac3bb5aad601e590521"
+dependencies = [
+ "num-traits",
+]
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -259,6 +268,7 @@ dependencies = [
  "async-stream",
  "async-trait",
  "backoff",
+ "bit-struct",
  "bitflags 1.3.2",
  "byteorder",
  "bytes",
@@ -272,7 +282,6 @@ dependencies = [
  "influxdb2",
  "itertools 0.14.0",
  "libc",
- "log",
  "serde",
  "serde_derive",
  "serde_json",

Cargo.toml

@@ -15,7 +15,6 @@ influxdb2 = { version = "0.3.3", features = [
 ], default-features = false }
 tokio = { version = "1", features = ["rt", "io-util"] }
 clap = { version = "4.0", features = ["derive"] }
-log = "0.4"
 figment = { version = "0.10", features = ["toml"] }
 gethostname = "0.3"
 futures = "0.3.24"
@@ -36,6 +35,7 @@ backoff = { version = "0.4.0", features = ["tokio"] }
 serde_repr = "0.1.20"
 tracing = "0.1.44"
 tracing-subscriber = { version = "0.3.22", features = ["fmt"] }
+bit-struct = { version = "0.3.2", default-features = false }
 
 [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/lib.rs

@@ -2,6 +2,7 @@ pub mod chrony;
 pub mod chrony_refclock;
 pub mod gpsd;
 pub mod hwmon;
+pub mod prs10;
 pub mod uccm;
 
 use async_trait::async_trait;
@@ -127,6 +128,29 @@ impl Default for GpsdConfig {
     }
 }
 
+#[derive(Serialize, Deserialize, Clone)]
+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(10),
+        }
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct TimeReport {
     pub system_time: DateTime<Utc>,
@@ -226,6 +250,7 @@ pub struct SourcesConfig {
     pub hwmon: HwmonConfig,
     pub uccm: UCCMConfig,
     pub gpsd: GpsdConfig,
+    pub prs10: Prs10Config,
 }
 
 #[derive(Serialize, Deserialize, Clone, Default)]

src/main.rs

@@ -5,7 +5,10 @@ use tokio::sync::broadcast;
 use tracing::{Instrument, debug, error, info, info_span, warn};
 use tracing_subscriber;
 
-use crate::{chrony::*, chrony_refclock::ChronySockServer, hwmon::HwmonSource, uccm::UCCMMonitor};
+use crate::{
+    chrony::*, chrony_refclock::ChronySockServer, hwmon::HwmonSource, prs10::Prs10Monitor,
+    uccm::UCCMMonitor,
+};
 use chimemon::{gpsd::GpsdSource, *};
 
 const PROGRAM_NAME: &str = "chimemon";
@@ -138,6 +141,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         ))
     }
 
+    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")),
+        ))
+    }
+
     if tasks.len() == 0 {
         error!("No tasks configured, exiting.");
         return Ok(()); // not an error, but exit before starting a dummy task
@@ -152,6 +168,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 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:?}");
+                            }
+                            _ => {}
+                        }
                     }
                 }
             }

src/prs10.rs

@@ -0,0 +1,487 @@
+use std::sync::Arc;
+
+use crate::{
+    ChimemonMessage, ChimemonSource, ChimemonSourceChannel, Prs10Config, SourceMetric,
+    SourceReport, SourceReportDetails, SourceStatus,
+};
+use async_trait::async_trait;
+use bit_struct::u4;
+use bitflags::bitflags;
+use itertools::Itertools;
+use serde::Deserialize;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf};
+use tokio::select;
+use tokio::sync::OnceCell;
+use tokio::time::{Interval, interval, timeout};
+use tokio_serial;
+use tokio_serial::{SerialPort, SerialStream};
+use tracing::{debug, error, info, 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! {
+    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: Vec<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"),
+        ];
+
+        let no_tags = Arc::new(vec![]);
+
+        // 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! {
+    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! {
+    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! {
+    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! {
+    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! {
+    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)]
+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 { bits: 4 };
+        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")?,
+        })
+    }
+}
+
+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,
+}
+
+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()),
+            // 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()),
+        ]
+    }
+}
+
+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()")
+    }
+
+    pub async fn cmd_response(&mut self, cmd: &[u8]) -> Result<Vec<u8>, std::io::Error> {
+        debug!("cmd: `{:?}`", String::from_utf8_lossy(cmd));
+        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!("cmd 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: u4) -> 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_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_float(&mut self, cmd: &[u8]) -> Result<f64, Box<dyn std::error::Error>> {
+        debug!("Getting float value for command {cmd:?}");
+        let resp = self.cmd_response(cmd).await?;
+        let val = str::from_utf8(&resp)?.parse::<f64>()?;
+        Ok(val)
+    }
+
+    async fn status_poll(&mut self) -> Option<ChimemonMessage> {
+        debug!("polling status");
+        let status = self.get_status().await;
+        if let Ok(status) = status {
+            Some(ChimemonMessage::SourceReport(SourceReport {
+                name: "prs10".into(),
+                status: if status.is_healthy() {
+                    SourceStatus::Healthy
+                } else {
+                    SourceStatus::Unknown
+                },
+                details: Arc::new(status),
+            }))
+        } else {
+            None
+        }
+    }
+
+    async fn stats_poll(&mut self) -> Option<ChimemonMessage> {
+        debug!("polling stats");
+
+        let ocxo_efc = self.get_ocxo_efc().await;
+
+        Some(ChimemonMessage::SourceReport(SourceReport {
+            name: "prs10".into(),
+            status: SourceStatus::Unknown,
+            details: Arc::new(Prs10Stats {}),
+        }))
+    }
+}
+
+#[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 pps_timer = interval(self.config.stats_interval);
+
+        loop {
+            let msg = select! {
+                _ = status_timer.tick() => {
+                    self.status_poll().await
+                },
+                _ = pps_timer.tick() => {
+                    self.stats_poll().await
+                }
+            };
+            if let Some(msg) = msg {
+                chan.send(msg).expect("Unable to send to channel");
+            }
+        }
+    }
+}
+
+mod tests {
+    use crate::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(())
+    }
+}