Skip to content

fix: repair mis-resolved merge conflicts in main.rs and decoder.rs #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mreymann merged 1 commit into from
Aug 6, 2026
Merged

fix: repair mis-resolved merge conflicts in main.rs and decoder.rs #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 36 additions & 72 deletions src/decoder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,64 +137,6 @@ pub fn handle_service_data(

// --- BTHome Decoder ---

/// The BTHome v2 device-information byte, i.e. the first byte of the service
/// data. See <https://bthome.io/format/>.
struct BthomeDeviceInfo {
/// Bits 5-7. This decoder only understands version 2.
version: u8,
/// Bit 0. When set, everything after this byte is ciphertext.
encrypted: bool,
}

BlePacketType::Other => {
// Every phone, watch and TV in range lands here. Now that each
// advertisement is processed rather than only the first one per
// device, saying so on every packet drowns out the readings.
}
}
}

/// Number of value bytes that follow a BTHome object id, or `None` if the
/// length cannot be determined.
///
/// This is what keeps the parser in sync: an object we do not interpret still
/// has to be skipped by exactly the right number of bytes, otherwise the next
/// value byte gets read as an object id. Lengths are from
/// <https://bthome.io/format/>. `tail` is the payload after the object id and
/// is only needed for the few variable-length objects.
fn bthome_value_len(object_id: u8, tail: &[u8]) -> Option<usize> {
let len = match object_id {
// 1-byte sensor values: packet id, battery, count, humidity, moisture,
// UV index, temperature, channel, light level, settings revision.
0x00 | 0x01 | 0x09 | 0x2E | 0x2F | 0x46 | 0x57 | 0x58 | 0x59 | 0x60 | 0x64 | 0x65 => 1,
// 2-byte sensor values.
0x02 | 0x03 | 0x06 | 0x07 | 0x08 | 0x0C | 0x0D | 0x0E | 0x12 | 0x13 | 0x14 | 0x40
| 0x41 | 0x43 | 0x44 | 0x45 | 0x47 | 0x48 | 0x49 | 0x4A | 0x51 | 0x52 | 0x56 | 0x5A
| 0x5D | 0x5E | 0x5F | 0x61 => 2,
// 3-byte sensor values: pressure, illuminance, energy, power, duration, gas.
0x04 | 0x05 | 0x0A | 0x0B | 0x42 | 0x4B => 3,
// 4-byte sensor values.
0x4C | 0x4D | 0x4E | 0x4F | 0x50 | 0x55 | 0x5B | 0x5C | 0x62 | 0x63 => 4,
// Binary sensors and the button event are one byte each.
0x0F..=0x11 | 0x15..=0x2D | 0x3A => 1,
// Command and dimmer events carry a step count for some event types.
0x3B => match *tail.first()? {
0x03 | 0x04 => 2,
_ => 1,
},
0x3C => match *tail.first()? {
0x01 | 0x02 => 2,
_ => 1,
},
// Text and raw are length-prefixed.
0x53 | 0x54 => 1 + usize::from(*tail.first()?),
_ => return None,
};
Some(len)
}

// --- BTHome Decoder ---

/// The BTHome v2 device-information byte, i.e. the first byte of the service
/// data. See <https://bthome.io/format/>.
struct BthomeDeviceInfo {
Expand Down Expand Up @@ -262,33 +204,52 @@ fn decode_bthome(payload: &[u8], options: DecodeOptions<'_>) -> Option<SensorDat
let info = BthomeDeviceInfo::parse(info_byte);

if info.version != 2 {
println!(
" [!] BTHome v{} is not supported (device info 0x{info_byte:02X})",
log::debug!(
"BTHome v{} is not supported (device info 0x{info_byte:02X})",
info.version
);
return None;
}

if info.encrypted {
// Reading the ciphertext as if it were objects yields plausible-looking
// nonsense, so refuse instead of inventing measurements.
println!(" [!] BTHome payload is encrypted and no bind key is configured");
return None;
}
// Held so that `rest` can borrow from it for the rest of the function.
let decrypted;
let mut rest = if info.encrypted {
let Some(key) = options.bindkey else {
// Reading the ciphertext as if it were objects yields
// plausible-looking nonsense, so refuse instead of inventing
// measurements.
log::debug!("BTHome payload is encrypted and no bind key is configured");
return None;
};

let mut result = SensorData::default();
decrypted = match crypto::decrypt_bthome(payload, options.mac, key) {
Ok(plaintext) => plaintext,
Err(e) => {
log::warn!("could not decrypt BTHome payload: {e}");
return None;
}
};
decrypted.as_slice()
} else {
plaintext
};

let mut result = SensorData {
format: SensorFormat::BtHome,
..Default::default()
};

while let Some((&object_id, tail)) = rest.split_first() {
let Some(len) = bthome_value_len(object_id, tail) else {
println!(
" [!] Unknown BTHome object id 0x{object_id:02X}; stopping here rather than \
guessing its length"
log::debug!(
"unknown BTHome object id 0x{object_id:02X}; stopping here rather than guessing \
its length"
);
break;
};

if tail.len() < len {
println!(" [!] Truncated BTHome object 0x{object_id:02X}");
log::debug!("truncated BTHome object 0x{object_id:02X}");
break;
}

Expand Down Expand Up @@ -470,7 +431,10 @@ fn decode_mijia(payload: &[u8], options: DecodeOptions<'_>) -> Result<SensorData
.ok_or("frame ends before the first data object")?
};

let mut result = SensorData::default();
let mut result = SensorData {
format: SensorFormat::MiBeacon,
..Default::default()
};

// Each object is a little-endian u16 id, a length byte, then that many value
// bytes. Because the length is explicit, an object this decoder does not
Expand Down
221 changes: 82 additions & 139 deletions src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
use bluer::{Adapter, AdapterEvent, Address, DeviceProperty, Result};
use clap::Parser;
use futures::StreamExt;
use std::collections::HashMap;
use log::LevelFilter;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, mpsc};
use tokio::time::sleep;
use uuid::Uuid;
use std::time::Duration;

mod config;
mod crypto;
mod decoder;
mod exec;
mod metrics;
Expand Down Expand Up @@ -114,9 +114,6 @@ impl Args {
}
}

/// The service data of the advertisement last seen from a device.
type ServiceData = HashMap<Uuid, Vec<u8>>;

#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let args = Args::parse();
Expand All @@ -134,98 +131,86 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let adapter = session.default_adapter().await?;
adapter.set_powered(true).await?;

// The advertisement each device last sent, so that a property change which
// carries no new reading (an RSSI update, say) is not reported twice.
let mut last_service_data: HashMap<Address, ServiceData> = HashMap::new();
let last_ble_packet = Arc::new(Mutex::new(Instant::now()));
let (tx, mut rx) = mpsc::unbounded_channel::<AdapterEvent>();
log::info!(
"scanning on {} ({} configured sensor(s){}, watchdog {}s, cooldown {}s)",
adapter.name(),
config.configured_sensors(),
if config.only_known() {
", ignoring the rest"
} else {
""
},
args.watchdog,
args.cooldown
);

// One registry shared by the scan loop, the /metrics endpoint and the
// Pushgateway loop; only built when something actually wants it.
let registry = (args.metrics_addr.is_some() || args.pushgateway_url.is_some())
.then(|| Arc::new(Registry::new()));

if let Some(addr) = args.metrics_addr {
let registry = Arc::clone(
registry
.as_ref()
.expect("registry exists with --metrics-addr"),
);
tokio::spawn(async move {
if let Err(e) = metrics::serve(addr, registry).await {
log::error!("metrics endpoint stopped: {e}");
}
});
}

//
// 🔄 Discovery + watchdog task
//
{
let adapter = adapter.clone();
let tx = tx.clone();
let last_ble_packet = last_ble_packet.clone();
let watchdog = args.watchdog;
let cooldown = args.cooldown;
if let Some(url) = &args.pushgateway_url {
let target = PushTarget::parse(url)?;
let registry = Arc::clone(
registry
.as_ref()
.expect("registry exists with --pushgateway-url"),
);
let interval = Duration::from_secs(args.push_interval.max(1));
tokio::spawn(metrics::push_periodically(target, registry, interval));
}

tokio::spawn(async move {
let mut restart_counter: u64 = 1;

loop {
println!("🔍 (Re)starting discovery...");

// discover_devices_with_changes() re-emits DeviceAdded every time
// a device's properties change, which is what turns this into a
// continuous reader: discover_devices() only reports each device
// once, so every advertisement after the first was invisible.
let mut events = match adapter.discover_devices_with_changes().await {
Ok(ev) => ev,
Err(e) => {
eprintln!("❌ Failed to start discovery: {e}");
sleep(Duration::from_secs(cooldown)).await;
continue;
}
};

loop {
tokio::select! {
evt = events.next() => {
match evt {
Some(AdapterEvent::DeviceAdded(addr)) => {
let _ = tx.send(AdapterEvent::DeviceAdded(addr));
}
Some(AdapterEvent::DeviceRemoved(addr)) => {
let _ = tx.send(AdapterEvent::DeviceRemoved(addr));
}
Some(_) => {}
None => {
println!("⚠️ Discovery stream ended — restarting...");
break;
}
}
}

_ = sleep(Duration::from_secs(5)) => {
let elapsed = last_ble_packet.lock().await.elapsed();
if elapsed > Duration::from_secs(watchdog) {
println!(
"⏱ Watchdog: no BLE packets for {elapsed:?}, restarting discovery (count {restart_counter})..."
);
restart_counter += 1;

// Drop the current stream (equivalent to disable_le_scan)
drop(events);

// Wait before restarting (equivalent to Python's 5s delay)
sleep(Duration::from_secs(cooldown)).await;

break;
}
}
}
}
let settings = scan::Settings {
watchdog: Duration::from_secs(args.watchdog),
cooldown: Duration::from_secs(args.cooldown),
format: args.format,
config,
exec: args
.exec
.clone()
.map(|program| Hook::new(program, Duration::from_secs(args.exec_interval))),
metrics: registry,
};

tokio::select! {
result = scan::run(&adapter, settings) => result?,
() = shutdown_requested() => log::info!("shutting down"),
}

//
// 📡 Event processing loop
//
while let Some(evt) = rx.recv().await {
match evt {
AdapterEvent::DeviceAdded(addr) => {
if let Err(e) =
handle_device(&adapter, addr, &mut last_service_data, &last_ble_packet).await
{
eprintln!("Error handling device {addr}: {e}");
Ok(())
}

/// Resolves once the process is asked to stop, so discovery is torn down cleanly
/// instead of being killed mid-scan. SIGTERM matters here because that is what
/// systemd sends.
async fn shutdown_requested() {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};

match signal(SignalKind::terminate()) {
Ok(mut terminate) => {
tokio::select! {
() = wait_for_ctrl_c() => {}
_ = terminate.recv() => {}
}
}
AdapterEvent::DeviceRemoved(addr) => {
last_service_data.remove(&addr);
Err(e) => {
log::warn!("cannot listen for SIGTERM: {e}");
wait_for_ctrl_c().await;
}
}
}
Expand All @@ -234,52 +219,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
wait_for_ctrl_c().await;
}

async fn handle_device(
adapter: &Adapter,
addr: Address,
last_service_data: &mut HashMap<Address, ServiceData>,
last_ble_packet: &Mutex<Instant>,
) -> Result<()> {
let device = adapter.device(addr)?;

// One D-Bus round-trip for every property, instead of one round-trip each for
// the name, the RSSI and the service data. At one advertisement per second
// per sensor that is the difference the Pi Zero W notices.
let mut name = None;
let mut rssi = None;
let mut service_data = None;
for property in device.all_properties().await? {
match property {
DeviceProperty::Name(value) => name = Some(value),
DeviceProperty::Rssi(value) => rssi = Some(value),
DeviceProperty::ServiceData(value) => service_data = Some(value),
_ => {}
}
}

// No service data at all: not something this tool can read.
let Some(service_data) = service_data else {
return Ok(());
};

// Identical to the advertisement we already reported, so some other property
// changed. Sensors bump a packet counter on every broadcast, so a genuinely
// new reading always differs here.
if last_service_data.get(&addr) == Some(&service_data) {
return Ok(());
async fn wait_for_ctrl_c() {
if let Err(e) = tokio::signal::ctrl_c().await {
log::warn!("cannot listen for Ctrl-C: {e}");
// Never resolve, so the scan keeps running instead of exiting at once.
std::future::pending::<()>().await;
}

if let Some(decoded) = decoder::handle_service_data(&service_data) {
let name = name.as_deref().unwrap_or("<unknown>");
let rssi = rssi.map_or_else(|| "n/a".to_string(), |value| value.to_string());
println!("📡 {addr} ({name}), RSSI={rssi}");
println!(" 🔍 Got sensor reading: {decoded:?}");

// ✅ Reset watchdog timer only on actual sensor readings
*last_ble_packet.lock().await = Instant::now();
}

last_service_data.insert(addr, service_data);

Ok(())
}
Loading