Add influxdb

.gitignore

@@ -1,9 +1,2 @@
-venv/
+/target/
-__pycache__/
+Cargo.lock
-*.py[cod]
-*.egg-info/
-dist/
-build/
-.pytest_cache/
-.mypy_cache/
-.ruff_cache/

Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "towerd"
+version = "0.1.0"
+edition = "2021"
+description = "Tower daemon for KW1FOX-1"
+
+[dependencies]
+anyhow = "1"
+clap = { version = "4", features = ["derive"] }
+futures = "0.3"
+influxdb2 = { version = "0.5", default-features = false, features = ["rustls"] }
+rppal = "0.22"
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

README.md

@@ -3,16 +3,24 @@
 I wrote this to control the "hardware" on KW1FOX-1 tower.
 
 its.. not really intended for use anywhere else but, if anything inspires you go ahead and grab it!
-Just some simple python.
 
-# Install
+## Install
 
 ```shell
 sudo git clone https://git.kitsunehosting.net/Kenwood/towerd.git /opt/towerd
 cd /opt/towerd
-sudo python3 -m venv venv
+cargo build --release
-sudo venv/bin/pip install -r requirements.txt
 sudo cp systemd/towerd.service /etc/systemd/system/
 sudo systemctl daemon-reload
 sudo systemctl enable --now towerd
 ```
+
+## Usage
+
+```shell
+./target/release/towerd --verbose
+
+# systemd
+sudo systemctl status towerd
+sudo journalctl -u towerd -f
+```

requirements.txt

@@ -1 +0,0 @@
-gpiozero>=2.0

src/alarm.rs

@@ -0,0 +1,59 @@
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use tokio::sync::{watch, Mutex};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Status {
+    Ok,
+    Alarm,
+}
+
+// Shared reference to alarm board
+pub type SharedAlarms = Arc<AlarmBoard>;
+
+pub struct AlarmBoard {
+    faults: Mutex<HashSet<&'static str>>,
+    status_tx: watch::Sender<Status>,
+}
+
+impl AlarmBoard {
+    // Create a new alarm board
+    pub fn new() -> (SharedAlarms, watch::Receiver<Status>) {
+        let (status_tx, status_rx) = watch::channel(Status::Ok);
+        let board = Arc::new(Self {
+            faults: Mutex::new(HashSet::new()),
+            status_tx,
+        });
+        (board, status_rx)
+    }
+
+    // Set a fault (source, string, active)
+    pub async fn set_fault(&self, source: &'static str, active: bool) {
+        let mut faults = self.faults.lock().await;
+        if active {
+            faults.insert(source);
+        } else {
+            faults.remove(source);
+        }
+
+        let status = if faults.is_empty() {
+            Status::Ok
+        } else {
+            Status::Alarm
+        };
+
+        let _ = self.status_tx.send_if_modified(|current| {
+            if *current != status {
+                *current = status;
+                true
+            } else {
+                false
+            }
+        });
+    }
+
+    pub fn status(&self) -> Status {
+        *self.status_tx.borrow()
+    }
+}

src/config.rs

@@ -0,0 +1,56 @@
+#[derive(Debug, Clone)]
+pub struct Config {
+    pub status_pin: u8,
+    pub fan_pin: u8,
+    pub fan_on_temp_c: f64,
+    pub fan_off_temp_c: f64,
+    pub poll_interval_s: f64,
+    pub thermal_alarm_temp_c: f64,
+    pub influx: Option<InfluxConfig>,
+}
+
+#[derive(Debug, Clone)]
+pub struct InfluxConfig {
+    pub url: String,
+    pub org: String,
+    pub bucket: String,
+    pub token: String,
+    pub host_tag: String,
+    pub metrics_interval_s: f64,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            status_pin: 4,
+            fan_pin: 17,
+            fan_on_temp_c: 40.0,
+            fan_off_temp_c: 35.0,
+            poll_interval_s: 2.0,
+            thermal_alarm_temp_c: 80.0,
+            influx: InfluxConfig::from_env(),
+        }
+    }
+}
+
+impl InfluxConfig {
+    pub fn from_env() -> Option<Self> {
+        let token = std::env::var("TOWERD_INFLUX_TOKEN").ok()?;
+        let org = std::env::var("TOWERD_INFLUX_ORG").ok()?;
+        let bucket = std::env::var("TOWERD_INFLUX_BUCKET").ok()?;
+
+        Some(Self {
+            url: std::env::var("TOWERD_INFLUX_URL")
+                .unwrap_or_else(|_| "http://influx.kitsunehosting.net:8086".into()),
+            org,
+            bucket,
+            token,
+            host_tag: std::env::var("TOWERD_INFLUX_HOST")
+                .unwrap_or_else(|_| "kw1fox-1".into()),
+            metrics_interval_s: std::env::var("TOWERD_INFLUX_INTERVAL_S")
+                .ok()
+                .and_then(|s| s.parse().ok())
+                .unwrap_or(30.0),
+        })
+    }
+}

src/daemon.rs

@@ -0,0 +1,47 @@
+use std::sync::Arc;
+
+use tracing::info;
+
+use crate::alarm::AlarmBoard;
+use crate::config::Config;
+use crate::gpio::Fan;
+use crate::tasks;
+
+pub async fn run(config: Config) -> anyhow::Result<()> {
+    let (alarms, status_rx) = AlarmBoard::new();
+    let fan = Arc::new(tokio::sync::Mutex::new(Fan::new(config.fan_pin)?));
+
+    info!(
+        status_pin = config.status_pin,
+        fan_pin = config.fan_pin,
+        "Tower daemon started"
+    );
+
+    if config.influx.is_none() {
+        info!("InfluxDB metrics disabled (set TOWERD_INFLUX_TOKEN, TOWERD_INFLUX_ORG, and TOWERD_INFLUX_BUCKET to enable)");
+    }
+
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            info!("Received shutdown signal");
+        }
+        result = tasks::status::run(config.status_pin, status_rx) => {
+            result?;
+        }
+        result = tasks::thermal::run(config.clone(), fan.clone(), alarms.clone()) => {
+            result?;
+        }
+        result = async {
+            match config.influx {
+                Some(influx) => tasks::metrics::run(influx, fan.clone(), alarms.clone()).await,
+                None => std::future::pending().await,
+            }
+        } => {
+            result?;
+        }
+    }
+
+    fan.lock().await.stop();
+    info!("Tower daemon stopped");
+    Ok(())
+}

src/gpio.rs

@@ -0,0 +1,42 @@
+use std::sync::Arc;
+
+use anyhow::Context;
+use rppal::gpio::{Gpio, OutputPin};
+use tokio::sync::Mutex;
+
+pub type SharedFan = Arc<Mutex<Fan>>;
+
+pub struct Fan {
+    _gpio: Gpio,
+    pin: OutputPin,
+}
+
+impl Fan {
+    pub fn new(fan_pin: u8) -> anyhow::Result<Self> {
+        let gpio = Gpio::new().context("failed to initialize GPIO")?;
+
+        let mut pin = gpio
+            .get(fan_pin)
+            .context("failed to open fan pin")?
+            .into_output();
+        pin.set_low();
+
+        Ok(Self { _gpio: gpio, pin })
+    }
+
+    pub fn on(&self) -> bool {
+        self.pin.is_set_high()
+    }
+
+    pub fn set_on(&mut self, on: bool) {
+        if on {
+            self.pin.set_high();
+        } else {
+            self.pin.set_low();
+        }
+    }
+
+    pub fn stop(&mut self) {
+        self.pin.set_low();
+    }
+}

src/main.rs

@@ -0,0 +1,40 @@
+mod alarm;
+mod config;
+mod daemon;
+mod gpio;
+mod tasks;
+mod thermal;
+
+use clap::Parser;
+use tracing::Level;
+use tracing_subscriber::EnvFilter;
+
+use crate::config::Config;
+
+#[derive(Parser)]
+#[command(name = "towerd", about = "Tower Daemon")]
+struct Args {
+    #[arg(short, long, help = "Enable debug logging")]
+    verbose: bool,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Args::parse();
+
+    let default_level = if args.verbose {
+        Level::DEBUG
+    } else {
+        Level::INFO
+    };
+    let filter = EnvFilter::builder()
+        .with_default_directive(default_level.into())
+        .from_env_lossy();
+
+    tracing_subscriber::fmt()
+        .with_env_filter(filter)
+        .with_target(false)
+        .init();
+
+    daemon::run(Config::default()).await
+}

src/tasks/metrics.rs

@@ -0,0 +1,74 @@
+use std::time::Duration;
+
+use futures::stream;
+use influxdb2::models::DataPoint;
+use influxdb2::Client;
+use tokio::time::MissedTickBehavior;
+use tracing::{info, warn};
+
+use crate::alarm::{SharedAlarms, Status};
+use crate::config::InfluxConfig;
+use crate::gpio::SharedFan;
+use crate::thermal;
+
+pub async fn run(config: InfluxConfig, fan: SharedFan, alarms: SharedAlarms) -> anyhow::Result<()> {
+    // Get client
+    let client = Client::new(&config.url, &config.org, &config.token);
+
+    // Get interval
+    let mut interval =
+        tokio::time::interval(Duration::from_secs_f64(config.metrics_interval_s));
+    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+    // Debug logging
+    info!(
+        url = %config.url,
+        org = %config.org,
+        bucket = %config.bucket,
+        host = %config.host_tag,
+        interval_s = config.metrics_interval_s,
+        "Metrics task started"
+    );
+
+    loop { // Main loop
+        interval.tick().await;
+
+        // Publish the metrics
+        match publish(&client, &config, &fan, &alarms).await {
+            Ok(()) => alarms.set_fault("metrics", false).await,
+            Err(e) => {
+                // Set the fault if the publish fails
+                warn!(error = %e, "Failed to publish metrics");
+                alarms.set_fault("metrics", true).await;
+            }
+        }
+    }
+}
+
+// Publish the metrics
+async fn publish(
+    client: &Client,
+    config: &InfluxConfig,
+    fan: &SharedFan,
+    alarms: &SharedAlarms,
+) -> anyhow::Result<()> {
+    let temp_c = tokio::task::spawn_blocking(thermal::read_cpu_temp_c)
+        .await??;
+
+    let fan_on = fan.lock().await.on();
+    let alarm = matches!(alarms.status(), Status::Alarm);
+
+    let point = DataPoint::builder("tower")
+        .tag("host", &config.host_tag)
+        .field("cpu_temp_c", temp_c)
+        .field("fan_on", fan_on)
+        .field("alarm", alarm)
+        .build()?;
+
+    client
+        .write(&config.bucket, stream::iter(vec![point]))
+        .await
+        .map_err(|e| anyhow::anyhow!("influxdb write failed: {e}"))?;
+
+    Ok(())
+}

src/tasks/mod.rs

@@ -0,0 +1,3 @@
+pub mod metrics;
+pub mod status;
+pub mod thermal;

src/tasks/status.rs

@@ -0,0 +1,45 @@
+use std::time::Duration;
+
+use anyhow::Context;
+use rppal::gpio::{Gpio, OutputPin};
+use tokio::sync::watch;
+
+use crate::alarm::Status;
+
+const SLOW_ON: Duration = Duration::from_secs(1);
+const SLOW_OFF: Duration = Duration::from_secs(1);
+const FAST_ON: Duration = Duration::from_millis(150);
+const FAST_OFF: Duration = Duration::from_millis(150);
+
+struct StatusLed {
+    pin: OutputPin,
+}
+
+impl Drop for StatusLed {
+    fn drop(&mut self) {
+        self.pin.set_low();
+    }
+}
+
+pub async fn run(status_pin: u8, status_rx: watch::Receiver<Status>) -> anyhow::Result<()> {
+    let gpio = Gpio::new().context("failed to initialize GPIO")?;
+    let led = gpio
+        .get(status_pin)
+        .context("failed to open status pin")?
+        .into_output();
+    let mut led = StatusLed { pin: led };
+
+    loop {
+        // Get the on and off times based on the status
+        let (on_time, off_time) = match *status_rx.borrow() {
+            Status::Ok => (SLOW_ON, SLOW_OFF),
+            Status::Alarm => (FAST_ON, FAST_OFF),
+        };
+
+        led.pin.set_high();
+        tokio::time::sleep(on_time).await;
+
+        led.pin.set_low();
+        tokio::time::sleep(off_time).await;
+    }
+}

src/tasks/thermal.rs

@@ -0,0 +1,57 @@
+use std::time::Duration;
+
+use tokio::time::MissedTickBehavior;
+use tracing::{debug, info};
+
+use crate::alarm::SharedAlarms;
+use crate::config::Config;
+use crate::gpio::{Fan, SharedFan};
+use crate::thermal;
+
+pub async fn run(config: Config, fan: SharedFan, alarms: SharedAlarms) -> anyhow::Result<()> {
+    // Setup the interval for the thermal task
+    let mut interval =
+        tokio::time::interval(Duration::from_secs_f64(config.poll_interval_s));
+    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+    loop { // Main loop
+        interval.tick().await;
+
+        // Read the temperature
+        let temp_c = tokio::task::spawn_blocking(thermal::read_cpu_temp_c)
+            .await??;
+
+        // Check if the temperature is over the alarm threshold
+        let over_temp = temp_c >= config.thermal_alarm_temp_c;
+        alarms.set_fault("thermal", over_temp).await;
+
+        // Update the fan
+        let mut fan = fan.lock().await;
+        update_fan(&mut fan, &config, temp_c);
+
+        // Debug logging
+        debug!(
+            temp_c,
+            fan = if fan.on() { "on" } else { "off" },
+            alarm = ?alarms.status(),
+        );
+    }
+}
+
+fn update_fan(fan: &mut Fan, config: &Config, temp_c: f64) {
+    if !fan.on() && temp_c >= config.fan_on_temp_c {
+        fan.set_on(true);
+        info!(
+            temp_c,
+            threshold = config.fan_on_temp_c,
+            "Fan on"
+        );
+    } else if fan.on() && temp_c <= config.fan_off_temp_c {
+        fan.set_on(false);
+        info!(
+            temp_c,
+            threshold = config.fan_off_temp_c,
+            "Fan off"
+        );
+    }
+}

src/thermal.rs

@@ -0,0 +1,21 @@
+use std::fs;
+use std::path::Path;
+
+const THERMAL_ZONE: &str = "/sys/class/thermal/thermal_zone0/temp";
+
+pub fn read_cpu_temp_c() -> anyhow::Result<f64> {
+    let path = Path::new(THERMAL_ZONE);
+    if !path.exists() {
+        // Bail throws an error and exits for us
+        anyhow::bail!(
+            "Thermal sensor not found at {THERMAL_ZONE}. Maybe running on bad hardware?"
+        );
+    }
+
+    let millidegrees: i32 = fs::read_to_string(path)?
+        .trim()
+        .parse()
+        .map_err(|e| anyhow::anyhow!("failed to parse thermal reading: {e}"))?;
+
+    Ok(millidegrees as f64 / 1000.0)
+}

systemd/towerd.service

@@ -6,7 +6,7 @@ After=multi-user.target
 Type=simple
 User=root
 WorkingDirectory=/opt/towerd
-ExecStart=/opt/towerd/venv/bin/python -m towerd
+ExecStart=/opt/towerd/target/release/towerd
 Restart=on-failure
 RestartSec=5
 

towerd/__init__.py

@@ -1,3 +0,0 @@
-"""Tower Daemon"""
-
-__version__ = "0.1.0"

towerd/__main__.py

@@ -1,3 +0,0 @@
-from towerd.main import main
-
-main()

towerd/gpio_ctrl.py

@@ -1,59 +0,0 @@
-from enum import Enum
-
-from gpiozero import LED, OutputDevice
-
-
-class Status(Enum):
-    OK = "ok"
-    ERROR = "error"
-
-
-# Blink timings (seconds)
-SLOW_ON = 1.0
-SLOW_OFF = 1.0
-FAST_ON = 0.15
-FAST_OFF = 0.15
-
-
-class TowerGPIO:
-    """GPIO controls"""
-
-    def __init__(self, status_pin: int = 4, fan_pin: int = 17) -> None:
-        self._status_led = LED(status_pin)
-        self._fan = OutputDevice(fan_pin, active_high=True, initial_value=False)
-        self._status = Status.OK
-
-    @property
-    def status(self) -> Status:
-        return self._status
-
-    @property
-    def fan_on(self) -> bool:
-        return self._fan.value
-
-    def set_status(self, status: Status) -> None:
-        if status == self._status:
-            return
-        self._status = status
-        self._apply_blink()
-
-    def set_fan(self, on: bool) -> None:
-        if on:
-            self._fan.on()
-        else:
-            self._fan.off()
-
-    def start(self) -> None:
-        self._apply_blink()
-
-    def stop(self) -> None:
-        self._status_led.off()
-        self._fan.off()
-        self._status_led.close()
-        self._fan.close()
-
-    def _apply_blink(self) -> None:
-        if self._status is Status.OK:
-            self._status_led.blink(on_time=SLOW_ON, off_time=SLOW_OFF, background=True)
-        else:
-            self._status_led.blink(on_time=FAST_ON, off_time=FAST_OFF, background=True)

towerd/main.py

@@ -1,96 +0,0 @@
-import argparse
-import logging
-import signal
-import time
-from dataclasses import dataclass
-
-from towerd.gpio_ctrl import Status, TowerGPIO
-from towerd.thermal import read_cpu_temp_c
-
-log = logging.getLogger(__name__)
-
-
-@dataclass(frozen=True)
-class Config:
-    status_pin: int = 4
-    fan_pin: int = 17
-    fan_on_temp_c: float = 50.0
-    fan_off_temp_c: float = 45.0
-    poll_interval_s: float = 2.0
-    error_temp_c: float = 80.0
-
-
-class TowerDaemon:
-    def __init__(self, config: Config) -> None:
-        self.config = config
-        self._running = True
-        self._gpio = TowerGPIO(
-            status_pin=config.status_pin,
-            fan_pin=config.fan_pin,
-        )
-
-    def run(self) -> None:
-        signal.signal(signal.SIGTERM, self._handle_signal)
-        signal.signal(signal.SIGINT, self._handle_signal)
-
-        self._gpio.start()
-        log.info(
-            "Tower daemon started (status GPIO %d, fan GPIO %d)",
-            self.config.status_pin,
-            self.config.fan_pin,
-        )
-
-        try:
-            while self._running:
-                self._tick()
-                time.sleep(self.config.poll_interval_s)
-        finally:
-            self._gpio.stop()
-            log.info("Tower daemon stopped")
-
-    def _tick(self) -> None:
-        temp_c = read_cpu_temp_c()
-        self._update_fan(temp_c)
-        self._update_status(temp_c)
-
-        log.debug(
-            "temp=%.1fC fan=%s status=%s",
-            temp_c,
-            "on" if self._gpio.fan_on else "off",
-            self._gpio.status.value,
-        )
-
-    def _update_fan(self, temp_c: float) -> None:
-        if not self._gpio.fan_on and temp_c >= self.config.fan_on_temp_c:
-            self._gpio.set_fan(True)
-            log.info("Fan on (%.1fC >= %.1fC)", temp_c, self.config.fan_on_temp_c)
-        elif self._gpio.fan_on and temp_c <= self.config.fan_off_temp_c:
-            self._gpio.set_fan(False)
-            log.info("Fan off (%.1fC <= %.1fC)", temp_c, self.config.fan_off_temp_c)
-
-    def _update_status(self, temp_c: float) -> None:
-        if temp_c >= self.config.error_temp_c:
-            self._gpio.set_status(Status.ERROR)
-        else:
-            self._gpio.set_status(Status.OK)
-
-    def _handle_signal(self, signum: int, _frame) -> None:
-        log.info("Received signal %s, shutting down", signum)
-        self._running = False
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="Tower Daemon")
-    parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
-    args = parser.parse_args()
-
-    logging.basicConfig(
-        level=logging.DEBUG if args.verbose else logging.INFO,
-        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
-    )
-
-    TowerDaemon(Config()).run()
-
-
-if __name__ == "__main__":
-    main()

towerd/thermal.py

@@ -1,14 +0,0 @@
-from pathlib import Path
-
-THERMAL_ZONE = Path("/sys/class/thermal/thermal_zone0/temp")
-
-
-def read_cpu_temp_c() -> float:
-    """Get the CPU temp"""
-    if not THERMAL_ZONE.exists():
-        raise FileNotFoundError(
-            f"Thermal sensor not found at {THERMAL_ZONE}. "
-            "Maybe running on bad hardware?"
-        )
-    millidegrees = int(THERMAL_ZONE.read_text().strip())
-    return millidegrees / 1000.0