Dreaded_X/automation_rs
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Switched to tracing

Browse Source
This commit is contained in:
Dreaded_X
2022-12-29 06:14:01 +01:00
parent 3c0f4bf3b3
commit 06e511758b
10 changed files with 145 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/config.rs
View File

@@ -1,6 +1,6 @@
use std::{fs, error::Error, collections::HashMap};
use log::{debug, trace};
use tracing::{debug, trace};
use rumqttc::AsyncClient;
use serde::Deserialize;
@@ -103,11 +103,11 @@ impl Device {
pub fn into(self, identifier: String, client: AsyncClient) -> DeviceBox {
match self {
Device::IkeaOutlet { info, mqtt, kettle } => {
trace!("\tIkeaOutlet [{} in {:?}]", info.name, info.room);
trace!(id = identifier, "IkeaOutlet [{} in {:?}]", info.name, info.room);
Box::new(IkeaOutlet::new(identifier, info, mqtt, kettle, client))
},
Device::WakeOnLAN { info, mqtt, mac_address } => {
trace!("\tWakeOnLan [{} in {:?}]", info.name, info.room);
trace!(id = identifier, "WakeOnLan [{} in {:?}]", info.name, info.room);
Box::new(WakeOnLAN::new(identifier, info, mqtt, mac_address, client))
},
}

11
src/devices.rs
View File

@@ -7,7 +7,7 @@ pub use self::wake_on_lan::WakeOnLAN;
use std::collections::HashMap;
use google_home::{GoogleHomeDevice, traits::OnOff};
use log::trace;
use tracing::{trace, debug, span, Level};
use crate::{mqtt::OnMqtt, presence::OnPresence};
@@ -51,6 +51,7 @@ impl Devices {
}
pub fn add_device(&mut self, device: DeviceBox) {
debug!(id = device.get_id(), "Adding device");
self.devices.insert(device.get_id(), device);
}
@@ -69,9 +70,9 @@ impl Devices {
impl OnMqtt for Devices {
fn on_mqtt(&mut self, message: &rumqttc::Publish) {
trace!("OnMqtt for devices");
self.as_on_mqtts().iter_mut().for_each(|(id, listener)| {
trace!("OnMqtt: {id}");
let _span = span!(Level::TRACE, "on_mqtt").entered();
trace!(id, "Handling");
listener.on_mqtt(message);
})
}
@@ -79,9 +80,9 @@ impl OnMqtt for Devices {
impl OnPresence for Devices {
fn on_presence(&mut self, presence: bool) {
trace!("OnPresence for devices");
self.as_on_presences().iter_mut().for_each(|(id, device)| {
trace!("OnPresence: {id}");
let _span = span!(Level::TRACE, "on_presence").entered();
trace!(id, "Handling");
device.on_presence(presence);
})
}

18
src/devices/ikea_outlet.rs
View File

@@ -4,7 +4,7 @@ use google_home::errors::ErrorCode;
use google_home::{GoogleHomeDevice, device, types::Type, traits};
use rumqttc::{AsyncClient, Publish};
use serde::{Deserialize, Serialize};
use log::{debug, trace, warn};
use tracing::{debug, trace, warn};
use tokio::task::JoinHandle;
use crate::config::{KettleConfig, InfoConfig, MqttDeviceConfig};
@@ -79,7 +79,7 @@ impl OnMqtt for IkeaOutlet {
let new_state = match StateMessage::try_from(message) {
Ok(state) => state,
Err(err) => {
warn!("Failed to parse message: {err}");
warn!(id = self.identifier, "Failed to parse message: {err}");
return;
}
}.state == "ON";
@@ -94,7 +94,7 @@ impl OnMqtt for IkeaOutlet {
handle.abort();
}
trace!("Updating state: {} => {}", self.last_known_state, new_state);
debug!(id = self.identifier, "Updating state to {new_state}");
self.last_known_state = new_state;
// If this is a kettle start a timeout for turning it of again
@@ -107,7 +107,7 @@ impl OnMqtt for IkeaOutlet {
let timeout = match kettle.timeout.clone() {
Some(timeout) => timeout,
None => {
trace!("Outlet is a kettle without timeout");
trace!(id = self.identifier, "Outlet is a kettle without timeout");
return;
},
};
@@ -117,16 +117,17 @@ impl OnMqtt for IkeaOutlet {
// get dropped
let client = self.client.clone();
let topic = self.mqtt.topic.clone();
let id = self.identifier.clone();
self.handle = Some(
tokio::spawn(async move {
debug!("Starting timeout ({timeout}s) for kettle...");
debug!(id, "Starting timeout ({timeout}s) for kettle...");
tokio::time::sleep(Duration::from_secs(timeout)).await;
// @TODO We need to call set_on(false) in order to turn the device off
// again, how are we going to do this?
debug!("Turning kettle off!");
debug!(id, "Turning kettle off!");
set_on(client, topic, false).await;
})
);
);
}
}
}
@@ -135,10 +136,11 @@ impl OnPresence for IkeaOutlet {
fn on_presence(&mut self, presence: bool) {
// Turn off the outlet when we leave the house
if !presence {
debug!(id = self.identifier, "Turning device off");
let client = self.client.clone();
let topic = self.mqtt.topic.clone();
tokio::spawn(async move {
set_on(client, topic, false).await;
set_on(client, topic, false).await;
});
}
}

13
src/devices/wake_on_lan.rs
View File

@@ -1,5 +1,5 @@
use google_home::{GoogleHomeDevice, types::Type, device, traits::{self, Scene}, errors::{ErrorCode, DeviceError}};
use log::{debug, warn};
use tracing::{debug, warn};
use rumqttc::{AsyncClient, Publish};
use serde::Deserialize;
@@ -56,7 +56,7 @@ impl OnMqtt for WakeOnLAN {
let payload = match StateMessage::try_from(message) {
Ok(state) => state,
Err(err) => {
warn!("Failed to parse message: {err}");
warn!(id = self.identifier, "Failed to parse message: {err}");
return;
}
};
@@ -96,23 +96,24 @@ impl traits::Scene for WakeOnLAN {
// @TODO In the future send the wake on lan package directly, this is kind of annoying
// if we are inside of docker, so for now just call a webhook that does it for us
let mac_address = self.mac_address.clone();
let id = self.identifier.clone();
tokio::spawn(async move {
debug!("Activating Computer: {}", mac_address);
debug!(id, "Activating Computer: {}", mac_address);
let req = match reqwest::get(format!("http://10.0.0.2:9000/start-pc?mac={mac_address}")).await {
Ok(req) => req,
Err(err) => {
warn!("Failed to call webhook: {err}");
warn!(id, "Failed to call webhook: {err}");
return;
}
};
if req.status() != 200 {
warn!("Failed to call webhook: {}", req.status());
warn!(id, "Failed to call webhook: {}", req.status());
}
});
Ok(())
} else {
debug!("Trying to deactive computer, this is not currently supported");
debug!(id = self.identifier, "Trying to deactive computer, this is not currently supported");
// We do not support deactivating this scene
Err(ErrorCode::DeviceError(DeviceError::ActionNotAvailable))
}

15
src/main.rs
View File

@@ -6,20 +6,22 @@ use axum::{Router, Json, routing::post, http::StatusCode};
use automation::{config::Config, presence::Presence, ntfy::Ntfy};
use dotenv::dotenv;
use rumqttc::{MqttOptions, Transport, AsyncClient};
use env_logger::Builder;
use log::{error, info, debug, LevelFilter};
use tracing::{error, info, metadata::LevelFilter};
use automation::{devices::Devices, mqtt::Mqtt};
use google_home::{GoogleHome, Request};
use tracing_subscriber::EnvFilter;
#[tokio::main]
async fn main() {
dotenv().ok();
// Setup logger
Builder::new()
.filter_module("automation", LevelFilter::Trace)
.parse_default_env()
let filter = EnvFilter::builder()
.with_default_directive(LevelFilter::INFO.into())
.from_env_lossy();
tracing_subscriber::fmt()
.with_env_filter(filter)
.init();
let config = std::env::var("AUTOMATION_CONFIG").unwrap_or("./config/config.toml".to_owned());
@@ -67,7 +69,6 @@ async fn main() {
device_config.into(identifier, client.clone())
})
.for_each(|device| {
debug!("Adding device {}", device.get_id());
devices.write().unwrap().add_device(device);
});

8
src/mqtt.rs
View File

@@ -1,8 +1,7 @@
use std::sync::{Weak, RwLock};
use log::{error, debug};
use tracing::{error, debug, trace, span, Level};
use rumqttc::{Publish, Event, Incoming, EventLoop};
use log::trace;
use tokio::task::JoinHandle;
pub trait OnMqtt {
@@ -21,8 +20,6 @@ impl Mqtt {
}
fn notify(&mut self, message: Publish) {
trace!("Listener count: {}", self.listeners.len());
self.listeners.retain(|listener| {
if let Some(listener) = listener.upgrade() {
listener.write().unwrap().on_mqtt(&message);
@@ -46,7 +43,8 @@ impl Mqtt {
let notification = self.eventloop.poll().await;
match notification {
Ok(Event::Incoming(Incoming::Publish(p))) => {
trace!("{:?}", p);
// Could cause problems in async
let _span = span!(Level::TRACE, "mqtt_message").entered();
self.notify(p);
},
Ok(..) => continue,

2
src/ntfy.rs
View File

@@ -1,6 +1,6 @@
use std::collections::HashMap;
use log::{warn, error};
use tracing::{warn, error};
use reqwest::StatusCode;
use serde::Serialize;
use serde_repr::*;

5
src/presence.rs
View File

@@ -1,6 +1,6 @@
use std::{sync::{Weak, RwLock}, collections::HashMap};
use log::{debug, warn, trace};
use tracing::{debug, warn, trace, span, Level};
use rumqttc::{AsyncClient, Publish};
use serde::{Serialize, Deserialize};
@@ -75,7 +75,8 @@ impl OnMqtt for Presence {
debug!("Overall presence updated: {overall_presence}");
self.overall_presence = overall_presence;
trace!("Listener count: {}", self.listeners.len());
// This has problems in async
let _span = span!(Level::TRACE, "presence_update").entered();
self.listeners.retain(|listener| {
if let Some(listener) = listener.upgrade() {