This commit is contained in:
Generated
+1
-1
@@ -572,7 +572,7 @@ checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
|
||||
|
||||
[[package]]
|
||||
name = "fluxo-rs"
|
||||
version = "0.5.4"
|
||||
version = "0.6.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"bluer",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "fluxo-rs"
|
||||
version = "0.5.4"
|
||||
version = "0.6.0"
|
||||
edition = "2024"
|
||||
|
||||
[[bin]]
|
||||
|
||||
@@ -1,12 +1,18 @@
|
||||
# fluxo
|
||||
|
||||
`fluxo` is a high-performance system metrics daemon and client designed specifically for Waybar. It entirely replaces standard shell scripts with a compiled Rust binary that collects data via a background polling loop and serves it over a Unix socket.
|
||||
`fluxo` is a high-performance system metrics daemon and client for status bars. It entirely replaces standard shell scripts with a compiled Rust binary that collects data via a background polling loop and serves it over a Unix socket.
|
||||
|
||||
With its **100% Native, Content-Based Event-Driven Architecture**, it consumes effectively 0% CPU while idle and signals Waybar to redraw *only* when the rendered UI text or icons physically change.
|
||||
With its **100% Native, Content-Based Event-Driven Architecture**, it consumes effectively 0% CPU while idle and pushes updates *only* when the rendered UI text or icons physically change.
|
||||
|
||||
It speaks two dialects, both driven by that same change detection:
|
||||
|
||||
- **Waybar** — the daemon sends `SIGRTMIN+N` and the bar re-runs `fluxo <module>`. See [Waybar Configuration](#waybar-configuration).
|
||||
- **Any other bar** (Quickshell, ags, eww, …) — hold one `fluxo stream <modules...>` child open and read newline-delimited JSON from its stdout. See [Streaming](#streaming).
|
||||
|
||||
## Key Features
|
||||
|
||||
- **100% Native Architecture**: Zero shell-outs or subprocesses. Uses `bluer` for Bluetooth, `libpulse-binding` for audio, `zbus` for MPRIS/DND, and `notify` for backlight.
|
||||
- **Push-Based for Any Bar**: `fluxo stream` turns the daemon into a push source, so bars that keep long-lived child processes never have to poll or re-exec a client.
|
||||
- **Content-Based Event Signaling**: `fluxo` evaluates your custom configuration formats internally. It only sends a `SIGRTMIN+X` signal to Waybar if the resulting string or CSS class has actually changed, eliminating pointless re-renders from raw polling fluctuations.
|
||||
- **Zero-Latency Interactions**: Direct library bindings mean that when you change your volume or connect a Bluetooth device via the CLI, the daemon updates instantly.
|
||||
- **Circuit Breaker (Failsafe)**: Automatically detects failing modules and enters a "Cool down" state, preventing resource waste and log spam. Fallback caching keeps your bar looking clean even during brief failures.
|
||||
@@ -115,6 +121,58 @@ To achieve zero-latency updates and zero-polling CPU usage, set `interval: 0` on
|
||||
}
|
||||
```
|
||||
|
||||
## Streaming
|
||||
|
||||
Waybar's model is *signal the bar, the bar re-runs a command*, which is why the
|
||||
one-shot client exists. Other bars have no equivalent of a signal, so they end up
|
||||
emulating one by polling — spawning a `fluxo <module>` client per module per
|
||||
interval, which means an exec and a dynamic link per reading, and values that are
|
||||
still up to one interval stale.
|
||||
|
||||
`fluxo stream` is the push-based path for those bars. Subscribe once, and the
|
||||
daemon writes a JSON line whenever a module's rendered output actually changes:
|
||||
|
||||
```console
|
||||
$ fluxo stream cpu mem net
|
||||
{"module":"cpu","text":"7.4|41.0","tooltip":"AMD Ryzen 7 PRO 3700U","class":"normal","percentage":7}
|
||||
{"module":"mem","text":"9.41|15.49","class":"normal","percentage":60}
|
||||
{"module":"cpu","text":"6.9|40.5","tooltip":"AMD Ryzen 7 PRO 3700U","class":"normal","percentage":6}
|
||||
```
|
||||
|
||||
Each line is one complete JSON object: a `module` key naming the source, plus the
|
||||
module's usual `text` and — when set — `tooltip`, `class` and `percentage`. A full
|
||||
snapshot of every subscribed module is sent immediately on connect, so a bar
|
||||
renders correctly the moment it attaches rather than after the first change.
|
||||
|
||||
Two deliberate differences from the one-shot client:
|
||||
|
||||
- `text` is **not** padded with figure-spaces and zero-width spaces. That padding
|
||||
exists to stop Waybar's proportional font from reflowing, and is noise to a
|
||||
consumer doing its own layout.
|
||||
- Modules without a watch channel (`power`, `game`, `pool`) are re-evaluated on a
|
||||
5 second sweep rather than event-driven. Unchanged output is never written, so a
|
||||
quiet sweep costs nothing on the wire.
|
||||
|
||||
Quickshell, for example, needs one `Process` for the whole bar:
|
||||
|
||||
```qml
|
||||
Process {
|
||||
running: true
|
||||
command: ["fluxo", "stream", "cpu", "mem", "gpu", "net", "sys"]
|
||||
stdout: SplitParser {
|
||||
onRead: line => {
|
||||
const ev = JSON.parse(line);
|
||||
// dispatch on ev.module
|
||||
}
|
||||
}
|
||||
onExited: reconnectTimer.restart()
|
||||
}
|
||||
```
|
||||
|
||||
The connection dies when the daemon restarts, so drive `running` back to `true`
|
||||
from a timer — that is a reconnect, not a poll, and only ever fires while the
|
||||
stream is actually down.
|
||||
|
||||
## Debugging
|
||||
|
||||
Use `--loglevel` to control log verbosity (trace, debug, info, warn, error):
|
||||
|
||||
@@ -37,6 +37,10 @@ format = "{interface} ({ip}): {rx:^4.1} MB/s {tx:^4.1} MB/s"
|
||||
# enabled = false # set to false to disable this module at runtime
|
||||
# tokens: {usage}, {temp}, {model}
|
||||
format = "CPU: {usage:^4.1}% {temp:^4.1}C"
|
||||
# Force a temperature sensor instead of auto-detecting it. Case-insensitive
|
||||
# substring of the label shown by `sensors` (e.g. "Tctl", "Package id 0",
|
||||
# "thinkpad CPU").
|
||||
# temp_sensor = "Tctl"
|
||||
|
||||
[memory]
|
||||
# enabled = false # set to false to disable this module at runtime
|
||||
|
||||
@@ -125,6 +125,10 @@ pub struct CpuConfig {
|
||||
#[serde(default = "default_true")]
|
||||
pub enabled: bool,
|
||||
pub format: String,
|
||||
/// Case-insensitive substring of the sensor label to force (e.g. `"Tctl"`,
|
||||
/// `"Package id 0"`). Empty means auto-detect.
|
||||
#[serde(default)]
|
||||
pub temp_sensor: String,
|
||||
}
|
||||
|
||||
impl Default for CpuConfig {
|
||||
@@ -132,6 +136,7 @@ impl Default for CpuConfig {
|
||||
Self {
|
||||
enabled: true,
|
||||
format: "CPU: {usage:>4.1}% {temp:>4.1}C".to_string(),
|
||||
temp_sensor: String::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+19
-2
@@ -254,9 +254,10 @@ pub async fn run_daemon(config_path: Option<PathBuf>) -> Result<()> {
|
||||
{
|
||||
let cfg = config.read().await;
|
||||
let fast_enabled = cfg.cpu.enabled || cfg.memory.enabled || cfg.sys.enabled;
|
||||
let temp_sensor = cfg.cpu.temp_sensor.clone();
|
||||
drop(cfg);
|
||||
if fast_enabled {
|
||||
let mut daemon = HardwareDaemon::new();
|
||||
let mut daemon = HardwareDaemon::new(temp_sensor);
|
||||
let token = cancel_token.clone();
|
||||
spawn_poll_loop_simple!(
|
||||
"fast_hw",
|
||||
@@ -274,7 +275,8 @@ pub async fn run_daemon(config_path: Option<PathBuf>) -> Result<()> {
|
||||
let slow_enabled = cfg.gpu.enabled || cfg.disk.enabled;
|
||||
drop(cfg);
|
||||
if slow_enabled {
|
||||
let mut daemon = HardwareDaemon::new();
|
||||
// Slow path never samples CPU temperature, so the override is moot.
|
||||
let mut daemon = HardwareDaemon::new(String::new());
|
||||
let token = cancel_token.clone();
|
||||
spawn_poll_loop_simple!(
|
||||
"slow_hw",
|
||||
@@ -456,6 +458,21 @@ async fn run_ipc_loop(
|
||||
return;
|
||||
}
|
||||
|
||||
// `stream` holds the connection open and pushes a
|
||||
// line per change instead of answering once.
|
||||
if *module_name == "stream" {
|
||||
let requested: Vec<String> =
|
||||
parts[1..].iter().map(|s| s.to_string()).collect();
|
||||
crate::stream::run_stream(
|
||||
writer,
|
||||
requested,
|
||||
state_clone,
|
||||
config_clone,
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
debug!(module = module_name, args = ?&parts[1..], "Handling IPC request");
|
||||
let response =
|
||||
handle_request(module_name, &parts[1..], &state_clone, &config_clone).await;
|
||||
|
||||
@@ -365,8 +365,16 @@ fn print_overview() {
|
||||
println!(" fluxo daemon [--config <path>] Start the background daemon");
|
||||
println!(" fluxo reload Hot-reload the daemon config");
|
||||
println!(" fluxo <module> [args...] Query or control a module");
|
||||
println!(" fluxo stream <module>... Follow modules as JSON lines");
|
||||
println!(" fluxo help [module] Show this help or module details\n");
|
||||
|
||||
println!("\x1b[1mBARS:\x1b[0m");
|
||||
println!(" Waybar set `signal = N` per module and let the daemon poke it;");
|
||||
println!(" `fluxo <module>` is re-run by the bar on each signal.");
|
||||
println!(" Everything keep one `fluxo stream cpu mem net ...` child running and");
|
||||
println!(" else read its stdout — one JSON line per actual change, so");
|
||||
println!(" there is no polling, no re-exec, and no interval lag.\n");
|
||||
|
||||
println!("\x1b[1mCONFIGURATION:\x1b[0m");
|
||||
println!(" Config file: $XDG_CONFIG_HOME/fluxo/config.toml");
|
||||
println!(" Format tokens in config strings use {{token}} syntax.");
|
||||
|
||||
+34
@@ -44,3 +44,37 @@ pub fn request_data(module: &str, args: &[&str]) -> anyhow::Result<String> {
|
||||
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
/// Subscribe to `modules` and copy the daemon's push stream to stdout.
|
||||
///
|
||||
/// Blocks until the daemon closes the connection or the process is killed —
|
||||
/// this is the long-lived counterpart to [`request_data`], so no read timeout
|
||||
/// is set. Each line is flushed immediately, otherwise a consumer reading our
|
||||
/// stdout through a pipe would see updates only once a block's worth had
|
||||
/// accumulated.
|
||||
pub fn stream_modules(modules: &[String]) -> anyhow::Result<()> {
|
||||
use std::io::{BufRead, BufReader};
|
||||
|
||||
let sock = socket_path();
|
||||
debug!(?modules, "Opening stream to daemon socket: {}", sock);
|
||||
let mut stream = UnixStream::connect(&sock)?;
|
||||
|
||||
let mut request = String::from("stream");
|
||||
for module in modules {
|
||||
request.push(' ');
|
||||
request.push_str(module);
|
||||
}
|
||||
request.push('\n');
|
||||
stream.write_all(request.as_bytes())?;
|
||||
|
||||
let reader = BufReader::new(stream);
|
||||
let mut stdout = std::io::stdout();
|
||||
for line in reader.lines() {
|
||||
let line = line?;
|
||||
stdout.write_all(line.as_bytes())?;
|
||||
stdout.write_all(b"\n")?;
|
||||
stdout.flush()?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
+18
@@ -28,6 +28,7 @@ mod output;
|
||||
mod registry;
|
||||
mod signaler;
|
||||
mod state;
|
||||
mod stream;
|
||||
mod utils;
|
||||
|
||||
use clap::{Parser, Subcommand, ValueEnum};
|
||||
@@ -91,6 +92,17 @@ enum Commands {
|
||||
},
|
||||
/// Reload the daemon configuration
|
||||
Reload,
|
||||
/// Stream module updates as newline-delimited JSON until interrupted
|
||||
///
|
||||
/// Push-based alternative to polling `fluxo <module>` on a timer: one
|
||||
/// connection, one line per actual change. Intended for bars that keep a
|
||||
/// long-lived child process (Quickshell, ags, eww) rather than re-running a
|
||||
/// command on a signal the way Waybar does.
|
||||
Stream {
|
||||
/// Modules to subscribe to (e.g. `cpu mem gpu net sys`)
|
||||
#[arg(required = true)]
|
||||
modules: Vec<String>,
|
||||
},
|
||||
/// Show detailed help for all modules or a specific module
|
||||
Help {
|
||||
/// Optional module name to show detailed help for
|
||||
@@ -142,6 +154,12 @@ fn main() {
|
||||
process::exit(1);
|
||||
}
|
||||
},
|
||||
Commands::Stream { modules } => {
|
||||
if let Err(e) = ipc::stream_modules(modules) {
|
||||
error!("Stream ended: {}", e);
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
Commands::Help { module } => {
|
||||
help::print_help(module.as_deref());
|
||||
}
|
||||
|
||||
+54
-28
@@ -1,6 +1,7 @@
|
||||
//! Screen backlight indicator, driven by `inotify` on
|
||||
//! `/sys/class/backlight/*/actual_brightness`. Falls back to a 5 s poll loop
|
||||
//! to catch any missed events.
|
||||
//! Screen backlight indicator, driven by `poll(2)` on
|
||||
//! `/sys/class/backlight/*/actual_brightness`. The kernel backlight class calls
|
||||
//! `sysfs_notify` on that attribute, so it wakes on `POLLPRI`; inotify does not
|
||||
//! work on sysfs. Falls back to a 5 s timeout to catch any missed wakeups.
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::error::Result;
|
||||
@@ -8,9 +9,9 @@ use crate::modules::WaybarModule;
|
||||
use crate::output::WaybarOutput;
|
||||
use crate::state::{AppReceivers, BacklightState};
|
||||
use crate::utils::{TokenValue, format_template};
|
||||
use notify::{Config as NotifyConfig, Event, RecommendedWatcher, RecursiveMode, Watcher};
|
||||
use std::io::{Read, Seek, SeekFrom};
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::mpsc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::watch;
|
||||
use tracing::{error, info};
|
||||
@@ -85,13 +86,16 @@ impl BacklightDaemon {
|
||||
info!("Monitoring backlight device: {:?}", dir);
|
||||
|
||||
let max_brightness_path = dir.join("max_brightness");
|
||||
let brightness_path = dir.join("actual_brightness");
|
||||
let brightness_path_fallback = dir.join("brightness");
|
||||
let actual_path = dir.join("actual_brightness");
|
||||
let set_path = dir.join("brightness");
|
||||
|
||||
let target_file = if brightness_path.exists() {
|
||||
brightness_path
|
||||
// `brightness` is the requested level; `actual_brightness` is a
|
||||
// hardware readback that can settle a few steps below it (amdgpu
|
||||
// reports 64265/64764 at full), which would render as 99 %.
|
||||
let level_file = if set_path.exists() {
|
||||
set_path
|
||||
} else {
|
||||
brightness_path_fallback
|
||||
actual_path.clone()
|
||||
};
|
||||
|
||||
let get_percentage = || -> u8 {
|
||||
@@ -100,7 +104,7 @@ impl BacklightDaemon {
|
||||
.trim()
|
||||
.parse()
|
||||
.unwrap_or(100.0);
|
||||
let current: f64 = std::fs::read_to_string(&target_file)
|
||||
let current: f64 = std::fs::read_to_string(&level_file)
|
||||
.unwrap_or_default()
|
||||
.trim()
|
||||
.parse()
|
||||
@@ -117,29 +121,51 @@ impl BacklightDaemon {
|
||||
percentage: get_percentage(),
|
||||
});
|
||||
|
||||
let (ev_tx, ev_rx) = mpsc::channel();
|
||||
let mut watcher = RecommendedWatcher::new(
|
||||
move |res: notify::Result<Event>| {
|
||||
if let Ok(event) = res
|
||||
&& event.kind.is_modify()
|
||||
{
|
||||
let _ = ev_tx.send(());
|
||||
}
|
||||
},
|
||||
NotifyConfig::default(),
|
||||
)
|
||||
.unwrap();
|
||||
// Watch the attribute the kernel actually notifies on, even when the
|
||||
// percentage is read from `brightness`.
|
||||
let watch_path = if actual_path.exists() {
|
||||
actual_path
|
||||
} else {
|
||||
dir.join("brightness")
|
||||
};
|
||||
|
||||
if let Err(e) = watcher.watch(&target_file, RecursiveMode::NonRecursive) {
|
||||
error!("Failed to watch backlight file: {}", e);
|
||||
let Ok(mut file) = std::fs::File::open(&watch_path) else {
|
||||
error!("Failed to open backlight file: {:?}", watch_path);
|
||||
return;
|
||||
};
|
||||
|
||||
// A sysfs poll event stays raised until the attribute is re-read, so
|
||||
// every wakeup must drain the fd or poll() spins.
|
||||
let mut scratch = String::new();
|
||||
let mut drain = |file: &mut std::fs::File| {
|
||||
scratch.clear();
|
||||
let _ = file.seek(SeekFrom::Start(0));
|
||||
let _ = file.read_to_string(&mut scratch);
|
||||
};
|
||||
drain(&mut file);
|
||||
|
||||
loop {
|
||||
let mut pfd = libc::pollfd {
|
||||
fd: file.as_raw_fd(),
|
||||
events: libc::POLLPRI | libc::POLLERR,
|
||||
revents: 0,
|
||||
};
|
||||
let rc = unsafe { libc::poll(&mut pfd, 1, 5_000) };
|
||||
|
||||
if rc < 0 {
|
||||
let err = std::io::Error::last_os_error();
|
||||
if err.kind() == std::io::ErrorKind::Interrupted {
|
||||
continue;
|
||||
}
|
||||
error!("poll() on backlight failed: {}", err);
|
||||
return;
|
||||
}
|
||||
|
||||
loop {
|
||||
if ev_rx.recv_timeout(Duration::from_secs(5)).is_ok() {
|
||||
if rc > 0 {
|
||||
drain(&mut file);
|
||||
// Debounce bursts from scroll-driven brightness changes.
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
while ev_rx.try_recv().is_ok() {}
|
||||
drain(&mut file);
|
||||
|
||||
let _ = tx.send(BacklightState {
|
||||
percentage: get_percentage(),
|
||||
|
||||
+56
-11
@@ -55,22 +55,74 @@ impl BtDaemon {
|
||||
self.session = Some(bluer::Session::new().await?);
|
||||
}
|
||||
let session = self.session.as_ref().unwrap();
|
||||
let adapter = session.default_adapter().await?;
|
||||
|
||||
let adapter = match session.default_adapter().await {
|
||||
Ok(adapter) => adapter,
|
||||
Err(e) if is_gone(&e) => return Ok(publish_powered_off(tx)),
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
let adapter_powered = adapter.is_powered().await.unwrap_or(false);
|
||||
|
||||
let mut connected_devices = Vec::new();
|
||||
|
||||
if adapter_powered {
|
||||
match collect_devices(&adapter, state, config).await {
|
||||
Ok(devices) => connected_devices = devices,
|
||||
Err(e) if is_gone(&e) => return Ok(publish_powered_off(tx)),
|
||||
Err(e) => return Err(e.into()),
|
||||
}
|
||||
}
|
||||
|
||||
let _ = tx.send(BtState {
|
||||
adapter_powered,
|
||||
devices: connected_devices,
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// BlueZ tears down its adapter and device objects when the radio is switched
|
||||
/// off, so calls racing with that return `NotFound`. That is a normal
|
||||
/// bluetooth-off transition, not a failure.
|
||||
fn is_gone(err: &bluer::Error) -> bool {
|
||||
err.kind == bluer::ErrorKind::NotFound
|
||||
}
|
||||
|
||||
fn publish_powered_off(tx: &watch::Sender<BtState>) {
|
||||
let _ = tx.send(BtState {
|
||||
adapter_powered: false,
|
||||
devices: vec![],
|
||||
});
|
||||
}
|
||||
|
||||
/// Collect paired+connected audio-sink devices, enriched by their plugins.
|
||||
async fn collect_devices(
|
||||
adapter: &bluer::Adapter,
|
||||
state: &AppReceivers,
|
||||
config: &Config,
|
||||
) -> bluer::Result<Vec<BtDeviceInfo>> {
|
||||
let mut addresses = adapter.device_addresses().await?;
|
||||
addresses.sort();
|
||||
let audio_sink_uuid = bluer::Uuid::from_u128(0x0000110b_0000_1000_8000_00805f9b34fb);
|
||||
|
||||
let mut connected_devices = Vec::new();
|
||||
|
||||
for addr in addresses {
|
||||
let device = adapter.device(addr)?;
|
||||
let device = match adapter.device(addr) {
|
||||
Ok(device) => device,
|
||||
Err(e) if is_gone(&e) => continue,
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
if !device.is_connected().await.unwrap_or(false) {
|
||||
continue;
|
||||
}
|
||||
let uuids = device.uuids().await?.unwrap_or_default();
|
||||
let uuids = match device.uuids().await {
|
||||
Ok(uuids) => uuids.unwrap_or_default(),
|
||||
// Device disappeared between enumeration and query.
|
||||
Err(e) if is_gone(&e) => continue,
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
if !uuids.contains(&audio_sink_uuid) {
|
||||
continue;
|
||||
}
|
||||
@@ -110,15 +162,8 @@ impl BtDaemon {
|
||||
}
|
||||
connected_devices.push(dev_info);
|
||||
}
|
||||
}
|
||||
|
||||
let _ = tx.send(BtState {
|
||||
adapter_powered,
|
||||
devices: connected_devices,
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Ok(connected_devices)
|
||||
}
|
||||
|
||||
static PLUGINS: LazyLock<Vec<Box<dyn BtPlugin>>> =
|
||||
|
||||
+18
-3
@@ -20,6 +20,21 @@ use zbus::{Connection, fdo::PropertiesProxy};
|
||||
/// Renders + toggles DND state. Args: `["show"]` (default) or `["toggle"]`.
|
||||
pub struct DndModule;
|
||||
|
||||
const SWAYNC_SERVICE: &str = "org.erikreider.swaync.control";
|
||||
|
||||
/// Build a SwayNC proxy, but only if the service is actually on the bus.
|
||||
///
|
||||
/// Constructing the proxy unconditionally makes zbus spawn a property-caching
|
||||
/// task that logs a `ServiceUnknown` warning when SwayNC isn't running.
|
||||
async fn swaync_proxy(connection: &Connection) -> Option<SwayncControlProxy<'_>> {
|
||||
let dbus = zbus::fdo::DBusProxy::new(connection).await.ok()?;
|
||||
let name = zbus::names::BusName::try_from(SWAYNC_SERVICE).ok()?;
|
||||
if !dbus.name_has_owner(name).await.unwrap_or(false) {
|
||||
return None;
|
||||
}
|
||||
SwayncControlProxy::new(connection).await.ok()
|
||||
}
|
||||
|
||||
/// Read dunst's `paused` property via raw D-Bus call.
|
||||
async fn dunst_get_paused(connection: &Connection) -> anyhow::Result<bool> {
|
||||
let reply = connection
|
||||
@@ -68,7 +83,7 @@ impl WaybarModule for DndModule {
|
||||
message: format!("DBus connection failed: {}", e),
|
||||
})?;
|
||||
|
||||
if let Ok(proxy) = SwayncControlProxy::new(&connection).await
|
||||
if let Some(proxy) = swaync_proxy(&connection).await
|
||||
&& let Ok(is_dnd) = proxy.dnd().await
|
||||
{
|
||||
let _ = proxy.set_dnd(!is_dnd).await;
|
||||
@@ -145,14 +160,14 @@ impl DndDaemon {
|
||||
|
||||
info!("Connected to D-Bus for DND monitoring");
|
||||
|
||||
if let Ok(proxy) = SwayncControlProxy::new(&connection).await
|
||||
if let Some(proxy) = swaync_proxy(&connection).await
|
||||
&& let Ok(is_dnd) = proxy.dnd().await
|
||||
{
|
||||
debug!("Found SwayNC, using signal-based DND monitoring");
|
||||
let _ = tx.send(DndState { is_dnd });
|
||||
|
||||
if let Ok(props_proxy) = PropertiesProxy::builder(&connection)
|
||||
.destination("org.erikreider.swaync.control")?
|
||||
.destination(SWAYNC_SERVICE)?
|
||||
.path("/org/erikreider/swaync/control")?
|
||||
.build()
|
||||
.await
|
||||
|
||||
+50
-9
@@ -14,14 +14,48 @@ use tokio::sync::watch;
|
||||
pub struct HardwareDaemon {
|
||||
sys: System,
|
||||
components: Components,
|
||||
/// User-forced CPU sensor label substring; empty means auto-detect.
|
||||
temp_sensor: String,
|
||||
gpu_vendor: Option<String>,
|
||||
gpu_poll_counter: u8,
|
||||
disk_poll_counter: u8,
|
||||
}
|
||||
|
||||
/// Sensors that are never the CPU but whose labels can look like one. Unlabeled
|
||||
/// hwmon channels reach us as `"<driver> temp1"`, so a bare `temp1` match would
|
||||
/// happily pick up the WiFi card or the NVMe drive.
|
||||
const NON_CPU_SENSORS: &[&str] = &[
|
||||
"iwlwifi", "wifi", "nvme", "amdgpu", "nouveau", "gpu", "bat", "acpi ac", "composite", "edge",
|
||||
"junction", "mem",
|
||||
];
|
||||
|
||||
/// Score a component label for "is this the CPU package temperature".
|
||||
/// Higher wins; 0 means unusable.
|
||||
fn cpu_sensor_rank(label: &str) -> u8 {
|
||||
if NON_CPU_SENSORS.iter().any(|s| label.contains(s)) {
|
||||
return 0;
|
||||
}
|
||||
// AMD (k10temp) and Intel (coretemp) package sensors are the canonical ones.
|
||||
if label.contains("tctl") || label.contains("tdie") || label.contains("package") {
|
||||
100
|
||||
} else if label.contains("core ") {
|
||||
80
|
||||
} else if label.contains("cpu") {
|
||||
// e.g. thinkpad_acpi's "thinkpad CPU" — right sensor, coarser reading.
|
||||
70
|
||||
} else if label.contains("k10temp") || label.contains("coretemp") {
|
||||
60
|
||||
} else if label.contains("acpitz") {
|
||||
20
|
||||
} else {
|
||||
0
|
||||
}
|
||||
}
|
||||
|
||||
impl HardwareDaemon {
|
||||
/// Build a new daemon with an initial `sysinfo` snapshot.
|
||||
pub fn new() -> Self {
|
||||
/// Build a new daemon with an initial `sysinfo` snapshot. `temp_sensor` is
|
||||
/// the configured CPU sensor override (empty to auto-detect).
|
||||
pub fn new(temp_sensor: String) -> Self {
|
||||
let mut sys = System::new();
|
||||
sys.refresh_cpu_usage();
|
||||
sys.refresh_memory();
|
||||
@@ -29,6 +63,7 @@ impl HardwareDaemon {
|
||||
Self {
|
||||
sys,
|
||||
components,
|
||||
temp_sensor: temp_sensor.to_lowercase(),
|
||||
gpu_vendor: None,
|
||||
gpu_poll_counter: 0,
|
||||
// Start at 9 so (counter + 1) % 10 == 0 on the first tick.
|
||||
@@ -57,18 +92,24 @@ impl HardwareDaemon {
|
||||
.unwrap_or_else(|| "Unknown".to_string());
|
||||
|
||||
let mut cpu_temp = 0.0;
|
||||
let mut best_rank = 0;
|
||||
for component in &self.components {
|
||||
let label = component.label().to_lowercase();
|
||||
if (label.contains("tctl")
|
||||
|| label.contains("cpu")
|
||||
|| label.contains("package")
|
||||
|| label.contains("temp1"))
|
||||
|
||||
let rank = if self.temp_sensor.is_empty() {
|
||||
cpu_sensor_rank(&label)
|
||||
} else if label.contains(&self.temp_sensor) {
|
||||
u8::MAX
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
if rank > best_rank
|
||||
&& let Some(temp) = component.temperature()
|
||||
&& temp > 0.0
|
||||
{
|
||||
cpu_temp = temp as f64;
|
||||
if cpu_temp > 0.0 {
|
||||
break;
|
||||
}
|
||||
best_rank = rank;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -207,6 +207,11 @@ fn get_primary_interface() -> Result<String> {
|
||||
let mask = u32::from_str_radix(parts[7], 16).unwrap_or(0);
|
||||
|
||||
if dest == "00000000" {
|
||||
// Skip ProtonVPN kill-switch interfaces — they exist
|
||||
// only for leak protection and don't carry real traffic.
|
||||
if iface.starts_with("pvpnksintrf") {
|
||||
continue;
|
||||
}
|
||||
defaults.push((mask, metric, iface.to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
+90
-25
@@ -5,24 +5,42 @@
|
||||
//! `watch::Receiver`, evaluates the module when its channel fires, and only
|
||||
//! signals Waybar when the rendered output actually changed. A 50 ms per-signal
|
||||
//! debounce prevents storms during rapid state churn.
|
||||
//!
|
||||
//! Waybar is optional: a bar that reads the daemon over `fluxo stream` (see
|
||||
//! [`crate::stream`]) never wants a signal at all. Discovery is therefore
|
||||
//! written to be cheap *and* to stay cheap when Waybar is simply not installed —
|
||||
//! see [`WaybarSignaler::send_signal`].
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::state::AppReceivers;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use sysinfo::{ProcessesToUpdate, System};
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::time::{Duration, Instant, sleep};
|
||||
use tracing::{debug, warn};
|
||||
|
||||
/// Shortest gap between two signals with the same number.
|
||||
const SIGNAL_DEBOUNCE: Duration = Duration::from_millis(50);
|
||||
|
||||
/// First cooldown applied after a failed Waybar lookup.
|
||||
const DISCOVERY_BACKOFF_MIN: Duration = Duration::from_secs(2);
|
||||
|
||||
/// Ceiling for the discovery cooldown. A Waybar started later is still picked
|
||||
/// up within this long, which is well inside "I just launched my bar" latency.
|
||||
const DISCOVERY_BACKOFF_MAX: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Sends real-time signals to the Waybar process.
|
||||
///
|
||||
/// Resolves Waybar's PID lazily and caches it — the PID is invalidated on
|
||||
/// signal failure (e.g. Waybar was restarted) and rediscovered via `sysinfo`.
|
||||
/// signal failure (e.g. Waybar was restarted) and rediscovered by scanning
|
||||
/// `/proc`. Failed lookups are cached too, under an exponential cooldown.
|
||||
pub struct WaybarSignaler {
|
||||
cached_pid: Option<i32>,
|
||||
sys: System,
|
||||
last_signal_sent: HashMap<i32, Instant>,
|
||||
/// When set, no rescan of `/proc` happens before this instant.
|
||||
discovery_blocked_until: Option<Instant>,
|
||||
/// Current cooldown length, doubling on each consecutive failed lookup.
|
||||
discovery_backoff: Duration,
|
||||
}
|
||||
|
||||
impl WaybarSignaler {
|
||||
@@ -30,51 +48,98 @@ impl WaybarSignaler {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
cached_pid: None,
|
||||
sys: System::new(),
|
||||
last_signal_sent: HashMap::new(),
|
||||
discovery_blocked_until: None,
|
||||
discovery_backoff: DISCOVERY_BACKOFF_MIN,
|
||||
}
|
||||
}
|
||||
|
||||
fn find_waybar_pid(&mut self) -> Option<i32> {
|
||||
self.sys.refresh_processes(ProcessesToUpdate::All, true);
|
||||
for (pid, process) in self.sys.processes() {
|
||||
if process.name() == "waybar" {
|
||||
return Some(pid.as_u32() as i32);
|
||||
/// Scan `/proc` for a process named `waybar`.
|
||||
///
|
||||
/// Reads only `/proc/<pid>/comm`, one short line per process. The previous
|
||||
/// implementation called `sysinfo`'s `refresh_processes(All, true)`, which
|
||||
/// walks every thread of every process and parses `stat`, `status` and
|
||||
/// `cmdline` for each — thousands of file reads per call with a browser
|
||||
/// open, and it ran on *every* signal (see `send_signal`).
|
||||
fn find_waybar_pid() -> Option<i32> {
|
||||
let dir = std::fs::read_dir("/proc").ok()?;
|
||||
for entry in dir.flatten() {
|
||||
let file_name = entry.file_name();
|
||||
let Some(pid) = file_name.to_str().and_then(|s| s.parse::<i32>().ok()) else {
|
||||
continue;
|
||||
};
|
||||
// `comm` is truncated to 15 bytes by the kernel, which "waybar"
|
||||
// comfortably fits inside.
|
||||
if let Ok(comm) = std::fs::read_to_string(format!("/proc/{}/comm", pid))
|
||||
&& comm.trim_end() == "waybar"
|
||||
{
|
||||
return Some(pid);
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn send_signal(&mut self, signal_num: i32) {
|
||||
if let Some(last) = self.last_signal_sent.get(&signal_num)
|
||||
&& last.elapsed() < Duration::from_millis(50)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
let mut valid_pid = false;
|
||||
/// Resolve Waybar's PID, honouring the failed-lookup cooldown.
|
||||
fn resolve_pid(&mut self) -> Option<i32> {
|
||||
if let Some(pid) = self.cached_pid
|
||||
&& unsafe { libc::kill(pid, 0) } == 0
|
||||
{
|
||||
valid_pid = true;
|
||||
return Some(pid);
|
||||
}
|
||||
self.cached_pid = None;
|
||||
|
||||
if let Some(until) = self.discovery_blocked_until
|
||||
&& Instant::now() < until
|
||||
{
|
||||
return None;
|
||||
}
|
||||
|
||||
if !valid_pid {
|
||||
self.cached_pid = self.find_waybar_pid();
|
||||
match Self::find_waybar_pid() {
|
||||
Some(pid) => {
|
||||
debug!("Discovered waybar at PID {}", pid);
|
||||
self.cached_pid = Some(pid);
|
||||
self.discovery_blocked_until = None;
|
||||
self.discovery_backoff = DISCOVERY_BACKOFF_MIN;
|
||||
Some(pid)
|
||||
}
|
||||
None => {
|
||||
// Nothing to signal. Without this cooldown every state change
|
||||
// rescanned /proc, which on a Waybar-less setup meant a full
|
||||
// process walk several times a second, forever.
|
||||
self.discovery_blocked_until = Some(Instant::now() + self.discovery_backoff);
|
||||
self.discovery_backoff =
|
||||
(self.discovery_backoff * 2).min(DISCOVERY_BACKOFF_MAX);
|
||||
debug!(
|
||||
"No waybar process; pausing discovery for {:?}",
|
||||
self.discovery_backoff
|
||||
);
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(pid) = self.cached_pid {
|
||||
fn send_signal(&mut self, signal_num: i32) {
|
||||
// Debounce on attempt, not on success: recording the timestamp only
|
||||
// after a successful send meant a missing Waybar bypassed the debounce
|
||||
// entirely, so nothing throttled the work below.
|
||||
if let Some(last) = self.last_signal_sent.get(&signal_num)
|
||||
&& last.elapsed() < SIGNAL_DEBOUNCE
|
||||
{
|
||||
return;
|
||||
}
|
||||
self.last_signal_sent.insert(signal_num, Instant::now());
|
||||
|
||||
let Some(pid) = self.resolve_pid() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let sig = libc::SIGRTMIN() + signal_num;
|
||||
if unsafe { libc::kill(pid, sig) } == 0 {
|
||||
debug!("Sent SIGRTMIN+{} to waybar (PID: {})", signal_num, pid);
|
||||
self.last_signal_sent.insert(signal_num, Instant::now());
|
||||
} else {
|
||||
warn!("Failed to send SIGRTMIN+{} to waybar", signal_num);
|
||||
self.cached_pid = None;
|
||||
}
|
||||
} else {
|
||||
debug!("Waybar process not found, skipping signal.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+176
@@ -0,0 +1,176 @@
|
||||
//! Push-based module streaming for bars that are not Waybar.
|
||||
//!
|
||||
//! Waybar's model is *signal the bar, the bar re-runs a command*, so the
|
||||
//! one-shot `fluxo <module>` client is the natural fit there. Every other shell
|
||||
//! (Quickshell, ags, eww, …) has to emulate it by polling — spawning a process
|
||||
//! per module per interval. At a 2 s interval over five modules that is 2.5
|
||||
//! forks a second, each one exec'ing and dynamically linking a 13 MB binary to
|
||||
//! read a value the daemon already has in memory, and it still shows values up
|
||||
//! to one interval stale.
|
||||
//!
|
||||
//! This module is the alternative: one long-lived connection, over which the
|
||||
//! daemon writes a JSON line whenever a subscribed module's rendered output
|
||||
//! actually changes.
|
||||
//!
|
||||
//! $ fluxo stream cpu mem net
|
||||
//! {"module":"cpu","text":"7.4|41.0"}
|
||||
//! {"module":"mem","text":"9.41|15.49"}
|
||||
//! {"module":"cpu","text":"6.9|40.5"}
|
||||
//!
|
||||
//! Each line is one complete JSON object with a `module` key naming the source
|
||||
//! and the module's usual fields (`text`, and `tooltip`/`class`/`percentage`
|
||||
//! when set) flattened alongside it. Unlike the one-shot client, `text` is *not*
|
||||
//! passed through [`crate::output::stabilize_text`]: the figure-space and
|
||||
//! zero-width-space padding exists to stop Waybar's proportional font from
|
||||
//! reflowing, and it is noise to a consumer that does its own layout.
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::output::WaybarOutput;
|
||||
use crate::state::AppReceivers;
|
||||
use serde::Serialize;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::time::{Duration, sleep};
|
||||
use tracing::debug;
|
||||
|
||||
/// How often subscribed modules are re-evaluated regardless of channel
|
||||
/// activity.
|
||||
///
|
||||
/// This covers the dispatch-only modules (`power`, `game`, `pool`), which have
|
||||
/// no watch channel to wake us, and doubles as a safety net for any state that
|
||||
/// changes without its channel firing. Unchanged output is never written, so a
|
||||
/// quiet tick costs nothing on the wire.
|
||||
const SWEEP_INTERVAL: Duration = Duration::from_secs(5);
|
||||
|
||||
/// One line of the stream: the module's own fields plus which module they came
|
||||
/// from.
|
||||
#[derive(Serialize)]
|
||||
struct StreamEvent<'a> {
|
||||
module: &'a str,
|
||||
#[serde(flatten)]
|
||||
output: WaybarOutput,
|
||||
}
|
||||
|
||||
/// Render `module_name` and return its serialised [`StreamEvent`] line.
|
||||
async fn render_line(
|
||||
module_name: &str,
|
||||
state: &AppReceivers,
|
||||
config: &Config,
|
||||
) -> Option<String> {
|
||||
let args = crate::registry::signaler_default_args(module_name);
|
||||
let output = crate::registry::dispatch(module_name, config, state, args)
|
||||
.await
|
||||
.ok()?;
|
||||
let mut line = serde_json::to_string(&StreamEvent {
|
||||
module: module_name,
|
||||
output,
|
||||
})
|
||||
.ok()?;
|
||||
line.push('\n');
|
||||
Some(line)
|
||||
}
|
||||
|
||||
/// Generates [`run_stream`] from the central module registry.
|
||||
///
|
||||
/// The shape mirrors [`crate::signaler`]: one cfg-gated `changed()` future per
|
||||
/// watched module, and one `select!` arm per module that re-renders the module
|
||||
/// names that channel feeds.
|
||||
macro_rules! gen_stream {
|
||||
($( { $feature:literal, $field:ident, $state:ty, [$($name:literal),+], [$($sig_name:literal),+], $module:path, $signal:ident, [$($default_arg:literal),*], $config:ident } )*) => {
|
||||
/// Serve one `stream` connection until the client disconnects.
|
||||
///
|
||||
/// `requested` holds the module names the client subscribed to; any name
|
||||
/// the registry does not know simply never produces output.
|
||||
pub async fn run_stream<W: AsyncWriteExt + Unpin>(
|
||||
mut writer: W,
|
||||
requested: Vec<String>,
|
||||
receivers: AppReceivers,
|
||||
config_lock: Arc<RwLock<Config>>,
|
||||
) {
|
||||
debug!(modules = ?requested, "Stream client subscribed");
|
||||
|
||||
// Two handles on the same channels: `watchers` is what we await
|
||||
// changes on (needs `&mut`), `evaluator` is what module rendering
|
||||
// reads from (needs `&`). Splitting them keeps the borrow checker
|
||||
// out of the `select!` below; a `watch::Receiver` clone still sees
|
||||
// the latest published value, so reads are unaffected.
|
||||
let mut watchers = receivers.clone();
|
||||
let evaluator = receivers;
|
||||
|
||||
let mut last: HashMap<String, String> = HashMap::new();
|
||||
|
||||
// Re-render every subscribed module and write the ones that moved.
|
||||
// Returns false once the client is gone.
|
||||
macro_rules! flush_modules {
|
||||
($names:expr) => {{
|
||||
let mut alive = true;
|
||||
for name in $names {
|
||||
if !requested.iter().any(|r| r == name) {
|
||||
continue;
|
||||
}
|
||||
let line = {
|
||||
let config = config_lock.read().await;
|
||||
render_line(name, &evaluator, &config).await
|
||||
};
|
||||
let Some(line) = line else { continue };
|
||||
if last.get(name).is_some_and(|prev| prev == &line) {
|
||||
continue;
|
||||
}
|
||||
last.insert(name.to_string(), line.clone());
|
||||
if writer.write_all(line.as_bytes()).await.is_err()
|
||||
|| writer.flush().await.is_err()
|
||||
{
|
||||
alive = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
alive
|
||||
}};
|
||||
}
|
||||
|
||||
// Prime the client with a full snapshot so a bar renders correctly
|
||||
// the instant it connects, rather than after the first change.
|
||||
let all: Vec<&str> = requested.iter().map(|s| s.as_str()).collect();
|
||||
if !flush_modules!(all.iter().copied()) {
|
||||
debug!("Stream client disconnected during initial snapshot");
|
||||
return;
|
||||
}
|
||||
|
||||
loop {
|
||||
$(
|
||||
#[cfg(not(feature = $feature))]
|
||||
let $field = std::future::pending::<
|
||||
std::result::Result<(), tokio::sync::watch::error::RecvError>,
|
||||
>();
|
||||
#[cfg(feature = $feature)]
|
||||
let $field = watchers.$field.changed();
|
||||
)*
|
||||
|
||||
let alive = tokio::select! {
|
||||
$(
|
||||
res = $field => {
|
||||
if res.is_ok() {
|
||||
flush_modules!([$($sig_name),+])
|
||||
} else {
|
||||
true
|
||||
}
|
||||
}
|
||||
)*
|
||||
_ = sleep(SWEEP_INTERVAL) => {
|
||||
flush_modules!(all.iter().copied())
|
||||
}
|
||||
};
|
||||
|
||||
if !alive {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Stream client disconnected");
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
for_each_watched_module!(gen_stream);
|
||||
Reference in New Issue
Block a user