gru/gree-controller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.14.2-connectivity

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-09-15 10:02:49 +02:00
parent 2fd7e47eb9
commit c3f6e773f8
41 changed files with 1295 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api.rs
+3 -1
View File
@@ -10,7 +10,7 @@ use crate::{
GroupControlPatch, HaReading, HistorySettings, HomeAssistantSettings,
HomeAssistantSettingsUpdate, HomeAssistantSettingsView, InfluxDbSettings,
InfluxDbSettingsUpdate, InfluxDbSettingsView, ManualDeviceRequest, NightModeSettings,
NotificationSettings, NotificationSettingsUpdate, NotificationSettingsView, Reading,
NotificationSettings, NotificationSettingsUpdate, NotificationSettingsView, NetworkReading, Reading,
RuntimeSettings, Schedule, SettingsSnapshot, TemporaryQuickThermostat,
TemporaryQuickThermostatRequest, Zone,
ZoneControlPatch, ZoneReading,
@@ -77,6 +77,7 @@ const SPA_ROUTES: &[&str] = &[
"/history/zones",
"/history/devices",
"/history/energy",
"/history/network",
"/history/sensors",
"/history/custom",
];
@@ -156,6 +157,7 @@ pub fn router(state: AppState) -> Router {
.route("/api/readings", get(readings))
.route("/api/history", get(history))
.route("/api/history/energy", get(energy_history))
.route("/api/history/network", get(network_history))
.route("/api/control-plan", get(control_plan))
.route("/api/events", get(events))
.route(
src/api/configuration.rs
+7
View File
@@ -628,6 +628,9 @@ fn normalize_imported_runtime_settings(settings: &mut RuntimeSettings) -> Result
settings.zone_interval_seconds = gree.zone_interval_seconds;
settings.discovery_timeout_ms = gree.discovery_timeout_ms;
settings.discovery_broadcast = gree.discovery_broadcast;
settings.ping_metrics_enabled = gree.ping_metrics_enabled;
settings.ping_interval_seconds = gree.ping_interval_seconds;
settings.ping_sample_count = gree.ping_sample_count;
settings.suppress_device_beep = gree.suppress_device_beep;
settings.compressor_protection_enabled = gree.compressor_protection_enabled;
settings.compressor_protection_seconds = gree.compressor_protection_seconds;
@@ -637,6 +640,10 @@ fn normalize_imported_runtime_settings(settings: &mut RuntimeSettings) -> Result
}
settings.gree_cloud.polling_interval_seconds =
settings.gree_cloud.polling_interval_seconds.clamp(30, 3600);
settings.gree_cloud.connectivity_metrics_interval_seconds =
settings.gree_cloud.connectivity_metrics_interval_seconds.clamp(30, 3600);
settings.gree_cloud.connectivity_metrics_sample_count =
settings.gree_cloud.connectivity_metrics_sample_count.clamp(1, 10);
if settings.gree_cloud.account_id.trim().is_empty() {
settings.gree_cloud.account_id = "default".into();
}
src/api/history.rs
+103
View File
@@ -775,3 +775,106 @@ async fn energy_history(
"latest": latest,
})))
}
#[derive(Debug, Deserialize)]
struct NetworkHistoryQuery {
target_id: Option<String>,
hours: Option<i64>,
limit: Option<u32>,
}
async fn combined_network_history(
state: &AppState,
target_id: Option<&str>,
since: chrono::DateTime<Utc>,
bucket_seconds: i64,
limit: u32,
) -> Result<(Vec<NetworkReading>, String, Option<String>), AppError> {
let influx = state.settings.read().await.influxdb.clone();
let cutoff = Utc::now() - ChronoDuration::days(influx.history_threshold_days.max(1) as i64);
if !influx.enabled || since >= cutoff {
return Ok((
state.db.list_network_history(target_id, since, bucket_seconds, limit)?,
"sqlite".into(),
None,
));
}
let mut warning = None;
let mut values = match influxdb::query_network(
&state.http,
&influx,
target_id,
since,
cutoff,
bucket_seconds,
limit,
)
.await
{
Ok(rows) => rows,
Err(err) => {
warning = Some(err.to_string());
state.log(
"warn",
"influx.query_error",
"InfluxDB connectivity history query failed",
json!({"target_id": target_id, "error": err.to_string()}),
);
state.db.list_network_history(target_id, since, bucket_seconds, limit)?
}
};
if warning.is_none() {
values.extend(state.db.list_network_history(target_id, cutoff, bucket_seconds, limit)?);
}
values.sort_by_key(|row| row.timestamp);
trim_history(&mut values, limit);
Ok((
values,
if warning.is_some() { "sqlite_fallback".into() } else { "influx+sqlite".into() },
warning,
))
}
async fn network_history(
State(state): State<AppState>,
Query(query): Query<NetworkHistoryQuery>,
) -> Result<Json<Value>, AppError> {
let hours = query.hours.unwrap_or(24).clamp(1, 24 * 3650);
let since = Utc::now() - ChronoDuration::hours(hours);
let bucket_seconds = history_bucket_seconds(hours);
let limit = query.limit.unwrap_or(20_000).clamp(1, 50_000);
let target_id = query
.target_id
.as_deref()
.filter(|value| !value.is_empty() && *value != "all");
let (readings, storage, warning) =
combined_network_history(&state, target_id, since, bucket_seconds, limit).await?;
let settings = state.settings.read().await.clone();
let mut targets = Vec::<Value>::new();
for device in state.db.list_devices()? {
if device.enabled && device.connection_type == ConnectionType::Local && !device.simulated {
targets.push(json!({
"id": device.id,
"name": device.name,
"kind": "device",
"source": "local_udp",
}));
}
}
let has_rest = readings.iter().any(|row| row.target_id == "cloud:rest");
let has_mqtt = readings.iter().any(|row| row.target_id == "cloud:mqtt");
if (settings.gree_cloud.enabled && settings.gree_cloud.connectivity_metrics_enabled) || has_rest {
targets.push(json!({"id":"cloud:rest","name":"GREE Cloud REST","kind":"cloud_service","source":"cloud_rest"}));
}
if (settings.gree_cloud.enabled && settings.gree_cloud.connectivity_metrics_enabled) || has_mqtt {
targets.push(json!({"id":"cloud:mqtt","name":"GREE Cloud MQTT","kind":"cloud_service","source":"cloud_mqtt"}));
}
Ok(Json(json!({
"readings": readings,
"targets": targets,
"bucket_seconds": bucket_seconds,
"storage": storage,
"storage_warning": warning,
})))
}
src/api/settings.rs
+21 -1
View File
@@ -11,6 +11,9 @@ fn gree_settings(settings: &RuntimeSettings) -> GreeSettings {
zone_interval_seconds: settings.zone_interval_seconds,
discovery_timeout_ms: settings.discovery_timeout_ms,
discovery_broadcast: settings.discovery_broadcast.clone(),
ping_metrics_enabled: settings.ping_metrics_enabled,
ping_interval_seconds: settings.ping_interval_seconds,
ping_sample_count: settings.ping_sample_count,
suppress_device_beep: settings.suppress_device_beep,
compressor_protection_enabled: settings.compressor_protection_enabled,
compressor_protection_seconds: settings.compressor_protection_seconds,
@@ -24,6 +27,9 @@ fn gree_cloud_settings(settings: &RuntimeSettings) -> GreeCloudSettingsView {
username: settings.gree_cloud.username.clone(),
password_configured: !settings.gree_cloud.password.is_empty(),
polling_interval_seconds: settings.gree_cloud.polling_interval_seconds,
connectivity_metrics_enabled: settings.gree_cloud.connectivity_metrics_enabled,
connectivity_metrics_interval_seconds: settings.gree_cloud.connectivity_metrics_interval_seconds,
connectivity_metrics_sample_count: settings.gree_cloud.connectivity_metrics_sample_count,
installation_id: settings.gree_cloud.installation_id.clone(),
account_id: settings.gree_cloud.account_id.clone(),
last_successful_contact: settings.gree_cloud.last_successful_contact,
@@ -138,6 +144,8 @@ fn normalize_gree_settings(mut input: GreeSettings) -> Result<GreeSettings, AppE
input.poll_interval_seconds = input.poll_interval_seconds.clamp(2, 3600);
input.zone_interval_seconds = input.zone_interval_seconds.clamp(2, 3600);
input.discovery_timeout_ms = input.discovery_timeout_ms.clamp(300, 30_000);
input.ping_interval_seconds = input.ping_interval_seconds.clamp(10, 3600);
input.ping_sample_count = input.ping_sample_count.clamp(1, 10);
input.compressor_protection_seconds = input.compressor_protection_seconds.clamp(30, 1800);
if !(input.discovery_broadcast.eq_ignore_ascii_case("auto")
|| input
@@ -201,6 +209,9 @@ async fn update_gree_settings(
settings.zone_interval_seconds = input.zone_interval_seconds;
settings.discovery_timeout_ms = input.discovery_timeout_ms;
settings.discovery_broadcast = input.discovery_broadcast;
settings.ping_metrics_enabled = input.ping_metrics_enabled;
settings.ping_interval_seconds = input.ping_interval_seconds;
settings.ping_sample_count = input.ping_sample_count;
settings.suppress_device_beep = input.suppress_device_beep;
settings.compressor_protection_enabled = input.compressor_protection_enabled;
settings.compressor_protection_seconds = input.compressor_protection_seconds;
@@ -216,7 +227,10 @@ async fn update_gree_settings(
"GREE settings updated",
json!({
"compressor_protection_enabled": payload.compressor_protection_enabled,
"compressor_protection_seconds": payload.compressor_protection_seconds
"compressor_protection_seconds": payload.compressor_protection_seconds,
"ping_metrics_enabled": payload.ping_metrics_enabled,
"ping_interval_seconds": payload.ping_interval_seconds,
"ping_sample_count": payload.ping_sample_count
}),
);
state.broadcast("settings.gree.updated", serde_json::to_value(&payload)?);
@@ -246,6 +260,9 @@ fn apply_gree_cloud_update(
next.region = input.region.trim().to_string();
next.username = input.username.trim().to_string();
next.polling_interval_seconds = input.polling_interval_seconds.clamp(30, 3600);
next.connectivity_metrics_enabled = input.connectivity_metrics_enabled;
next.connectivity_metrics_interval_seconds = input.connectivity_metrics_interval_seconds.clamp(30, 3600);
next.connectivity_metrics_sample_count = input.connectivity_metrics_sample_count.clamp(1, 10);
if let Some(password) = input.password {
next.password = password;
}
@@ -292,6 +309,9 @@ async fn update_gree_cloud_settings(
"enabled": payload.enabled,
"region": payload.region,
"polling_interval_seconds": payload.polling_interval_seconds,
"connectivity_metrics_enabled": payload.connectivity_metrics_enabled,
"connectivity_metrics_interval_seconds": payload.connectivity_metrics_interval_seconds,
"connectivity_metrics_sample_count": payload.connectivity_metrics_sample_count,
"password_configured": payload.password_configured
}),
);
src/config.rs
+12
View File
@@ -76,6 +76,9 @@ impl Config {
zone_interval_seconds: self.zone_interval_seconds.max(2),
discovery_timeout_ms: self.discovery_timeout_ms.clamp(300, 30_000),
discovery_broadcast: self.discovery_broadcast.clone(),
ping_metrics_enabled: env_bool("GREE_CONTROLLER_PING_METRICS_ENABLED").unwrap_or(true),
ping_interval_seconds: env_u64("GREE_CONTROLLER_PING_INTERVAL_SECONDS").unwrap_or(60).clamp(10, 3600),
ping_sample_count: env_u32("GREE_CONTROLLER_PING_SAMPLE_COUNT").unwrap_or(3).clamp(1, 10),
house_mode: env::var("GREE_CONTROLLER_HOUSE_MODE").unwrap_or_else(|_| "cool".into()),
control_strategy: "setpoint".into(),
outdoor_assist_enabled: env_bool("GREE_CONTROLLER_OUTDOOR_ASSIST_ENABLED")
@@ -137,6 +140,15 @@ impl Config {
/// Environment values explicitly supplied by the service override persisted runtime values.
pub fn apply_runtime_env_overrides(&self, settings: &mut RuntimeSettings) {
if let Some(value) = env_bool("GREE_CONTROLLER_PING_METRICS_ENABLED") {
settings.ping_metrics_enabled = value;
}
if let Some(value) = env_u64("GREE_CONTROLLER_PING_INTERVAL_SECONDS") {
settings.ping_interval_seconds = value.clamp(10, 3600);
}
if let Some(value) = env_u32("GREE_CONTROLLER_PING_SAMPLE_COUNT") {
settings.ping_sample_count = value.clamp(1, 10);
}
if env::var_os("GREE_CONTROLLER_HISTORY_RETENTION_DAYS").is_some() {
settings.history_retention_days = env_u32("GREE_CONTROLLER_HISTORY_RETENTION_DAYS")
.unwrap_or(settings.history_retention_days)
src/db.rs
+2 -1
View File
@@ -1,7 +1,7 @@
use crate::{
models::{
ApiTokenInfo, Automation, ClimateGroup, ConfigurationExport, ConnectionType, Device, DeviceGroup, EnergySourcePreference, EventLog, Flow,
EnergyReading, HaReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading,
EnergyReading, HaReading, NetworkReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading,
},
queries,
};
@@ -29,6 +29,7 @@ include!("db/device_history.rs");
include!("db/zone_history.rs");
include!("db/ha_history.rs");
include!("db/energy_history.rs");
include!("db/network_history.rs");
include!("db/events_tokens.rs");
include!("db/configuration.rs");
include!("db/tests.rs");
src/db/core_devices.rs
+1
View File
@@ -110,6 +110,7 @@ impl Db {
let tx = conn.transaction()?;
tx.execute(queries::DELETE_DEVICE_READINGS, [id])?;
tx.execute(queries::DELETE_DEVICE_ENERGY_READINGS, [id])?;
tx.execute("DELETE FROM network_readings WHERE target_id=?1", [id])?;
tx.execute(queries::DELETE_ZONE_READINGS_BY_DEVICE_ID, [id])?;
tx.execute(queries::DELETE_SCHEDULES_BY_DEVICE_ID, [id])?;
tx.execute(queries::DELETE_ZONES_BY_DEVICE_ID, [id])?;
src/db/network_history.rs
+143
View File
@@ -0,0 +1,143 @@
impl Db {
pub fn add_network_reading(&self, reading: &NetworkReading) -> Result<i64> {
let conn = self.lock()?;
conn.execute(
"INSERT INTO network_readings(target_id,target_kind,timestamp,latency_ms,jitter_ms,packet_loss_pct,sample_count,successful_samples,source) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9)",
params![
reading.target_id,
reading.target_kind,
reading.timestamp.to_rfc3339(),
reading.latency_ms,
reading.jitter_ms,
reading.packet_loss_pct.clamp(0.0, 100.0),
reading.sample_count as i64,
reading.successful_samples as i64,
reading.source,
],
)?;
Ok(conn.last_insert_rowid())
}
pub fn list_network_history(
&self,
target_id: Option<&str>,
since: DateTime<Utc>,
bucket_seconds: i64,
limit: u32,
) -> Result<Vec<NetworkReading>> {
let conn = self.lock()?;
let bucket_seconds = bucket_seconds.max(1);
let limit = limit.clamp(1, 50_000) as i64;
let sql_by_target = r#"
SELECT MIN(id),target_id,MAX(target_kind),MIN(timestamp),AVG(latency_ms),AVG(jitter_ms),AVG(packet_loss_pct),
CAST(ROUND(AVG(sample_count)) AS INTEGER),CAST(ROUND(AVG(successful_samples)) AS INTEGER),MAX(source)
FROM network_readings
WHERE target_id=?1 AND timestamp>=?2
GROUP BY target_id,source,CAST(unixepoch(timestamp)/?3 AS INTEGER)
ORDER BY MIN(timestamp) ASC
LIMIT ?4
"#;
let sql_all = r#"
SELECT MIN(id),target_id,MAX(target_kind),MIN(timestamp),AVG(latency_ms),AVG(jitter_ms),AVG(packet_loss_pct),
CAST(ROUND(AVG(sample_count)) AS INTEGER),CAST(ROUND(AVG(successful_samples)) AS INTEGER),MAX(source)
FROM network_readings
WHERE timestamp>=?1
GROUP BY target_id,source,CAST(unixepoch(timestamp)/?2 AS INTEGER)
ORDER BY MIN(timestamp) ASC
LIMIT ?3
"#;
let mut out = Vec::new();
if let Some(target_id) = target_id {
let mut stmt = conn.prepare(sql_by_target)?;
let rows = stmt.query_map(
params![target_id, since.to_rfc3339(), bucket_seconds, limit],
Self::map_network_reading,
)?;
for row in rows { out.push(row?); }
} else {
let mut stmt = conn.prepare(sql_all)?;
let rows = stmt.query_map(
params![since.to_rfc3339(), bucket_seconds, limit],
Self::map_network_reading,
)?;
for row in rows { out.push(row?); }
}
Ok(out)
}
pub fn network_before(&self, before: DateTime<Utc>, limit: u32) -> Result<Vec<NetworkReading>> {
let conn = self.lock()?;
let mut stmt = conn.prepare(
"SELECT id,target_id,target_kind,timestamp,latency_ms,jitter_ms,packet_loss_pct,sample_count,successful_samples,source FROM network_readings WHERE timestamp<?1 ORDER BY timestamp ASC,id ASC LIMIT ?2",
)?;
let rows = stmt.query_map(
params![before.to_rfc3339(), limit.clamp(1, 5000) as i64],
Self::map_network_reading,
)?;
rows.collect::<rusqlite::Result<Vec<_>>>().map_err(Into::into)
}
pub fn delete_network_batch(&self, rows: &[NetworkReading]) -> Result<u64> {
let mut conn = self.lock()?;
let tx = conn.transaction()?;
let mut changed = 0_u64;
for row in rows {
changed += tx.execute("DELETE FROM network_readings WHERE id=?1", [row.id])? as u64;
}
tx.commit()?;
Ok(changed)
}
pub fn prune_network_readings(&self, retention_days: i64) -> Result<u64> {
let before = Utc::now() - Duration::days(retention_days.max(1));
let conn = self.lock()?;
Ok(conn.execute("DELETE FROM network_readings WHERE timestamp < ?1", [before.to_rfc3339()])? as u64)
}
pub fn compact_network_history(&self, retention_days: i64) -> Result<u64> {
let now = Utc::now();
let one_day = now - Duration::days(1);
let seven_days = now - Duration::days(7);
let retention = now - Duration::days(retention_days.max(1));
let conn = self.lock()?;
let sql = r#"
DELETE FROM network_readings WHERE id IN (
SELECT id FROM (
SELECT id, ROW_NUMBER() OVER (
PARTITION BY target_id,source,CAST(unixepoch(timestamp)/?1 AS INTEGER)
ORDER BY timestamp DESC,id DESC
) AS rn
FROM network_readings WHERE timestamp < ?2 AND timestamp >= ?3
) WHERE rn > 1
)
"#;
let mut changed = 0_u64;
for (bucket, older_than, newer_than) in [
(600_i64, one_day, seven_days),
(1800_i64, seven_days, retention),
] {
if older_than <= newer_than { continue; }
changed += conn.execute(sql, params![bucket, older_than.to_rfc3339(), newer_than.to_rfc3339()])? as u64;
}
conn.execute_batch("PRAGMA optimize;")?;
Ok(changed)
}
fn map_network_reading(row: &rusqlite::Row<'_>) -> rusqlite::Result<NetworkReading> {
let timestamp: String = row.get(3)?;
Ok(NetworkReading {
id: row.get(0)?,
target_id: row.get(1)?,
target_kind: row.get(2)?,
timestamp: DateTime::parse_from_rfc3339(&timestamp)
.map(|value| value.with_timezone(&Utc))
.unwrap_or_else(|_| Utc::now()),
latency_ms: row.get(4)?,
jitter_ms: row.get(5)?,
packet_loss_pct: row.get::<_, f64>(6)?.clamp(0.0, 100.0),
sample_count: row.get::<_, i64>(7)?.max(0) as u32,
successful_samples: row.get::<_, i64>(8)?.max(0) as u32,
source: row.get(9)?,
})
}
}
src/engine.rs
+2 -1
View File
@@ -3,7 +3,7 @@ use crate::{
home_assistant, influxdb,
models::{
Automation, AutomationPlanRule, ClimateGroup, ConnectionStatus, ConnectionType, ControlPlan, ControlPlanEvent, Device, DeviceGroup, EnergyReading, EnergySourcePreference,
DeviceCommand, GroupControlPatch, HaReading, NightModeSettings, Reading, RuntimeSettings,
DeviceCommand, GroupControlPatch, HaReading, NetworkReading, NightModeSettings, Reading, RuntimeSettings,
Schedule, TemporaryQuickThermostat, Zone, ZoneControlPlan, ZoneReading,
},
state::{AppState, ControlPlanSnapshot, PendingControllerCommand},
@@ -30,6 +30,7 @@ include!("engine/groups.rs");
include!("engine/zone_actions.rs");
include!("engine/zone_control.rs");
include!("engine/history.rs");
include!("engine/connectivity.rs");
include!("engine/energy.rs");
include!("engine/temperature.rs");
include!("engine/targets.rs");
src/engine/connectivity.rs
+184
View File
@@ -0,0 +1,184 @@
fn build_network_reading(
target_id: String,
target_kind: &str,
source: &str,
samples: Vec<Option<u64>>,
) -> NetworkReading {
let sample_count = samples.len() as u32;
let successful: Vec<f64> = samples
.iter()
.filter_map(|value| value.map(|v| v as f64))
.collect();
let successful_samples = successful.len() as u32;
let latency_ms = if successful.is_empty() {
None
} else {
Some(successful.iter().sum::<f64>() / successful.len() as f64)
};
let jitter_ms = if successful.len() < 2 {
if successful.is_empty() { None } else { Some(0.0) }
} else {
let diffs = successful
.windows(2)
.map(|pair| (pair[1] - pair[0]).abs())
.collect::<Vec<_>>();
Some(diffs.iter().sum::<f64>() / diffs.len() as f64)
};
let packet_loss_pct = if sample_count == 0 {
0.0
} else {
((sample_count - successful_samples) as f64 / sample_count as f64) * 100.0
};
NetworkReading {
id: 0,
target_id,
target_kind: target_kind.into(),
timestamp: Utc::now(),
latency_ms,
jitter_ms,
packet_loss_pct,
sample_count,
successful_samples,
source: source.into(),
}
}
fn save_network_reading(state: &AppState, reading: NetworkReading) {
match state.db.add_network_reading(&reading) {
Ok(_) => queue_influx_network(state, reading),
Err(err) => tracing::warn!(error=?err, target_id=%reading.target_id, "cannot save connectivity metric"),
}
}
async fn probe_local_device_batch(state: &AppState, device: &Device, sample_count: u32) {
let mut samples = Vec::with_capacity(sample_count as usize);
for sample in 0..sample_count {
let result = state.providers.local().client().probe(device).await;
match result {
Ok(ms) => samples.push(Some(ms)),
Err(err) => {
tracing::debug!(error=?err, device_id=%device.id, "local connectivity probe failed");
samples.push(None);
}
}
if sample + 1 < sample_count {
sleep(Duration::from_millis(100)).await;
}
}
save_network_reading(
state,
build_network_reading(device.id.clone(), "device", "local_udp", samples),
);
}
async fn run_local_connectivity_cycle(state: &AppState, sample_count: u32) -> Result<()> {
let devices = state
.db
.list_devices()?
.into_iter()
.filter(|device| {
device.enabled
&& !device.simulated
&& device.connection_type == ConnectionType::Local
})
.collect::<Vec<_>>();
futures_util::future::join_all(
devices
.iter()
.map(|device| probe_local_device_batch(state, device, sample_count)),
)
.await;
Ok(())
}
async fn cloud_rest_probe(state: &AppState, cloud: &crate::models::GreeCloudSettings) -> Result<u64> {
let mut api = crate::protocol::gree_cloud::GreeCloudApi::for_region(
state.http.clone(),
&cloud.region,
&cloud.username,
&cloud.password,
)?;
let started = Instant::now();
api.login().await?;
Ok(started.elapsed().as_millis().min(u64::MAX as u128) as u64)
}
async fn run_cloud_connectivity_cycle(
state: &AppState,
cloud: &crate::models::GreeCloudSettings,
sample_count: u32,
) -> Result<()> {
if !cloud.enabled || cloud.username.trim().is_empty() || cloud.password.is_empty() {
return Ok(());
}
let devices = state.db.list_devices()?;
let mut rest = Vec::with_capacity(sample_count as usize);
let mut mqtt = Vec::with_capacity(sample_count as usize);
for sample in 0..sample_count {
match cloud_rest_probe(state, cloud).await {
Ok(ms) => rest.push(Some(ms)),
Err(err) => {
tracing::debug!(error=?err, "GREE Cloud REST connectivity probe failed");
rest.push(None);
}
}
match state.providers.cloud().probe_mqtt(cloud, &devices).await {
Ok(ms) => mqtt.push(Some(ms)),
Err(err) => {
tracing::debug!(error=?err, "GREE Cloud MQTT connectivity probe failed");
mqtt.push(None);
}
}
if sample + 1 < sample_count {
sleep(Duration::from_millis(150)).await;
}
}
save_network_reading(
state,
build_network_reading("cloud:rest".into(), "cloud_service", "cloud_rest", rest),
);
save_network_reading(
state,
build_network_reading("cloud:mqtt".into(), "cloud_service", "cloud_mqtt", mqtt),
);
Ok(())
}
async fn local_connectivity_loop(state: AppState) {
sleep(Duration::from_secs(5)).await;
loop {
let settings = state.settings.read().await.clone();
if settings.ping_metrics_enabled {
if let Err(err) = run_local_connectivity_cycle(&state, settings.ping_sample_count.clamp(1, 10)).await {
tracing::warn!(error=?err, "local connectivity metrics cycle failed");
}
sleep(Duration::from_secs(settings.ping_interval_seconds.clamp(10, 3600))).await;
} else {
sleep(Duration::from_secs(10)).await;
}
}
}
async fn cloud_connectivity_loop(state: AppState) {
sleep(Duration::from_secs(10)).await;
loop {
let cloud = state.settings.read().await.gree_cloud.clone();
if cloud.connectivity_metrics_enabled {
if let Err(err) = run_cloud_connectivity_cycle(
&state,
&cloud,
cloud.connectivity_metrics_sample_count.clamp(1, 10),
)
.await
{
tracing::warn!(error=?err, "GREE Cloud connectivity metrics cycle failed");
}
sleep(Duration::from_secs(
cloud.connectivity_metrics_interval_seconds.clamp(30, 3600),
))
.await;
} else {
sleep(Duration::from_secs(15)).await;
}
}
}
src/engine/history.rs
+13
View File
@@ -121,6 +121,19 @@ fn queue_influx_ha(state: &AppState, reading: HaReading) {
});
}
fn queue_influx_network(state: &AppState, reading: NetworkReading) {
let state = state.clone();
tokio::spawn(async move {
let settings = state.settings.read().await.influxdb.clone();
if !settings.enabled {
return;
}
if let Err(err) = influxdb::write_network(&state.http, &settings, &reading).await {
tracing::warn!(error=?err, target_id=%reading.target_id, source=%reading.source, "cannot write connectivity metric to InfluxDB");
}
});
}
fn median_outdoor_temperature(mut values: Vec<f64>) -> Option<f64> {
if values.is_empty() {
return None;
src/engine/runtime.rs
+33
View File
@@ -46,6 +46,15 @@ pub fn start(state: AppState) {
home_assistant_energy_loop(ha_energy_state).await;
});
let local_connectivity_state = state.clone();
tokio::spawn(async move {
local_connectivity_loop(local_connectivity_state).await;
});
let cloud_connectivity_state = state.clone();
tokio::spawn(async move {
cloud_connectivity_loop(cloud_connectivity_state).await;
});
let control_plan_state = state.clone();
tokio::spawn(async move {
control_plan_cache_loop(control_plan_state).await;
@@ -139,6 +148,11 @@ pub fn start(state: AppState) {
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot compact history"),
}
match maintenance_state.db.compact_network_history(compaction_days) {
Ok(count) if count > 0 => tracing::info!(count, "connectivity history samples compacted"),
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot compact connectivity history"),
}
}
if settings.influxdb.enabled {
match archive_old_history(
@@ -172,6 +186,13 @@ pub fn start(state: AppState) {
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot prune energy readings"),
}
match maintenance_state.db.prune_network_readings(retention_days) {
Ok(count) if count > 0 => {
tracing::info!(count, retention_days, "old connectivity readings pruned")
}
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot prune connectivity readings"),
}
}
let event_retention_days = settings.event_log_retention_days.max(1) as i64;
match maintenance_state.db.prune_events(event_retention_days) {
@@ -216,5 +237,17 @@ async fn archive_old_history(state: &AppState, threshold_days: u32) -> Result<u6
break;
}
}
for _ in 0..50 {
let network = state.db.network_before(cutoff, 1_000)?;
if network.is_empty() {
break;
}
influxdb::write_network_batch(&state.http, &settings, &network).await?;
let deleted = state.db.delete_network_batch(&network)?;
moved += deleted;
if deleted == 0 {
break;
}
}
Ok(moved)
}
src/influxdb.rs
+2 -1
View File
@@ -4,12 +4,13 @@ use reqwest::Client;
use serde_json::Value;
use std::collections::HashMap;
use crate::models::{EnergyReading, HaReading, InfluxDbSettings, Reading, ZoneReading};
use crate::models::{EnergyReading, HaReading, InfluxDbSettings, NetworkReading, Reading, ZoneReading};
const DEVICE_MEASUREMENT: &str = "gree_device";
const ZONE_MEASUREMENT: &str = "gree_zone";
const HA_MEASUREMENT: &str = "gree_ha";
const ENERGY_MEASUREMENT: &str = "gree_energy";
const NETWORK_MEASUREMENT: &str = "gree_network";
// Functional source split intentionally keeps items in the existing module namespace.
include!("influxdb/write.rs");
src/influxdb/query.rs
+78
View File
@@ -544,3 +544,81 @@ pub async fn query_energy(
out.sort_by_key(|row| row.timestamp);
Ok(out)
}
pub async fn query_network(
client: &Client,
settings: &InfluxDbSettings,
target_id: Option<&str>,
start: DateTime<Utc>,
stop: DateTime<Utc>,
bucket_seconds: i64,
limit: u32,
) -> Result<Vec<NetworkReading>> {
if settings.version == "1" {
let filter = target_id
.map(|id| format!(" AND \"target_id\"='{}'", influxql_string(id)))
.unwrap_or_default();
let q = format!(
"SELECT mean(\"latency_ms\") AS \"latency_ms\",mean(\"jitter_ms\") AS \"jitter_ms\",mean(\"packet_loss_pct\") AS \"packet_loss_pct\",mean(\"sample_count\") AS \"sample_count\",mean(\"successful_samples\") AS \"successful_samples\" FROM \"{NETWORK_MEASUREMENT}\" WHERE time >= '{}' AND time < '{}'{} GROUP BY time({}s),\"target_id\",\"target_kind\",\"source\" fill(none) LIMIT {}",
start.to_rfc3339(), stop.to_rfc3339(), filter, bucket_seconds.max(1), limit
);
let series = query_v1(client, settings, &q).await?;
let mut out = Vec::new();
for item in series {
let id = item.tags.get("target_id").cloned().unwrap_or_default();
let kind = item.tags.get("target_kind").cloned().unwrap_or_else(|| "device".into());
let source = item.tags.get("source").cloned().unwrap_or_default();
for row in item.rows {
let Some(timestamp) = row_time(&row) else { continue; };
out.push(NetworkReading {
id: 0,
target_id: id.clone(),
target_kind: kind.clone(),
timestamp,
latency_ms: row_num(&row, "latency_ms"),
jitter_ms: row_num(&row, "jitter_ms"),
packet_loss_pct: row_num(&row, "packet_loss_pct").unwrap_or(0.0).clamp(0.0, 100.0),
sample_count: row_num(&row, "sample_count").unwrap_or(0.0).round().max(0.0) as u32,
successful_samples: row_num(&row, "successful_samples").unwrap_or(0.0).round().max(0.0) as u32,
source: source.clone(),
});
}
}
out.sort_by_key(|row| row.timestamp);
out.truncate(limit as usize);
return Ok(out);
}
let tags = target_id
.map(|value| format!(" |> filter(fn: (r) => r.target_id == {})", flux_string(value)))
.unwrap_or_default();
let query = flux_query(
settings,
NETWORK_MEASUREMENT,
&tags,
&["target_id", "target_kind", "source"],
start,
stop,
bucket_seconds,
);
let rows = query_v2(client, settings, &query).await?;
let mut out = Vec::new();
for row in rows.into_iter().take(limit as usize) {
let Some(timestamp) = parse_flux_time(&row) else { continue; };
let Some(id) = row.get("target_id").filter(|v| !v.is_empty()) else { continue; };
out.push(NetworkReading {
id: 0,
target_id: id.clone(),
target_kind: row.get("target_kind").cloned().unwrap_or_else(|| "device".into()),
timestamp,
latency_ms: row_f64(&row, "latency_ms"),
jitter_ms: row_f64(&row, "jitter_ms"),
packet_loss_pct: row_f64(&row, "packet_loss_pct").unwrap_or(0.0).clamp(0.0, 100.0),
sample_count: row_f64(&row, "sample_count").unwrap_or(0.0).round().max(0.0) as u32,
successful_samples: row_f64(&row, "successful_samples").unwrap_or(0.0).round().max(0.0) as u32,
source: row.get("source").cloned().unwrap_or_default(),
});
}
out.sort_by_key(|row| row.timestamp);
Ok(out)
}
src/influxdb/write.rs
+46
View File
@@ -1,4 +1,50 @@
pub async fn write_network(
client: &Client,
settings: &InfluxDbSettings,
reading: &NetworkReading,
) -> Result<()> {
if !settings.enabled { return Ok(()); }
let mut fields = Vec::new();
push_float(&mut fields, "latency_ms", reading.latency_ms);
push_float(&mut fields, "jitter_ms", reading.jitter_ms);
push_float(&mut fields, "packet_loss_pct", Some(reading.packet_loss_pct.clamp(0.0, 100.0)));
push_int(&mut fields, "sample_count", reading.sample_count as i64);
push_int(&mut fields, "successful_samples", reading.successful_samples as i64);
let line = line_protocol(
NETWORK_MEASUREMENT,
&[("target_id", &reading.target_id), ("target_kind", &reading.target_kind), ("source", &reading.source)],
fields,
reading.timestamp,
)?;
write_line(client, settings, line).await
}
pub async fn write_network_batch(
client: &Client,
settings: &InfluxDbSettings,
readings: &[NetworkReading],
) -> Result<()> {
if !settings.enabled || readings.is_empty() { return Ok(()); }
let mut lines = Vec::with_capacity(readings.len());
for reading in readings {
let mut fields = Vec::new();
push_float(&mut fields, "latency_ms", reading.latency_ms);
push_float(&mut fields, "jitter_ms", reading.jitter_ms);
push_float(&mut fields, "packet_loss_pct", Some(reading.packet_loss_pct.clamp(0.0, 100.0)));
push_int(&mut fields, "sample_count", reading.sample_count as i64);
push_int(&mut fields, "successful_samples", reading.successful_samples as i64);
lines.push(line_protocol(
NETWORK_MEASUREMENT,
&[("target_id", &reading.target_id), ("target_kind", &reading.target_kind), ("source", &reading.source)],
fields,
reading.timestamp,
)?);
}
write_lines(client, settings, lines.join("\n")).await
}
pub async fn write_energy(
client: &Client,
settings: &InfluxDbSettings,
src/models/defaults.rs
+10
View File
@@ -88,6 +88,16 @@ fn default_cloud_region() -> String {
fn default_cloud_poll_interval_seconds() -> u64 {
60
}
fn default_ping_interval_seconds() -> u64 {
60
}
fn default_ping_sample_count() -> u32 {
3
}
fn default_cloud_metrics_interval_seconds() -> u64 {
300
}
fn default_influx_version() -> String {
"2".into()
}
src/models/history.rs
+21
View File
@@ -40,6 +40,27 @@ pub struct HaReading {
pub temperature: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetworkReading {
pub id: i64,
/// Stable endpoint id: local device id or a controller-level Cloud service id.
pub target_id: String,
/// `device` for Local/LAN units, `cloud_service` for REST/MQTT diagnostics.
pub target_kind: String,
pub timestamp: DateTime<Utc>,
/// Mean successful round-trip latency for this probe batch.
pub latency_ms: Option<f64>,
/// Mean absolute difference between consecutive successful RTT samples.
pub jitter_ms: Option<f64>,
/// Failed probes divided by all probes, in percent (0..100).
pub packet_loss_pct: f64,
pub sample_count: u32,
pub successful_samples: u32,
/// `local_udp`, `cloud_rest` or `cloud_mqtt`.
pub source: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventLog {
pub id: i64,
src/models/runtime.rs
+17
View File
@@ -10,6 +10,13 @@ pub struct GreeCloudSettings {
pub password: String,
#[serde(default = "default_cloud_poll_interval_seconds")]
pub polling_interval_seconds: u64,
/// Periodic Cloud REST/MQTT quality measurements for History. Disabled by default.
#[serde(default)]
pub connectivity_metrics_enabled: bool,
#[serde(default = "default_cloud_metrics_interval_seconds")]
pub connectivity_metrics_interval_seconds: u64,
#[serde(default = "default_ping_sample_count")]
pub connectivity_metrics_sample_count: u32,
/// Random, non-personal installation identifier used in the HTTP User-Agent.
#[serde(default)]
pub installation_id: String,
@@ -31,6 +38,9 @@ impl Default for GreeCloudSettings {
username: String::new(),
password: String::new(),
polling_interval_seconds: default_cloud_poll_interval_seconds(),
connectivity_metrics_enabled: false,
connectivity_metrics_interval_seconds: default_cloud_metrics_interval_seconds(),
connectivity_metrics_sample_count: default_ping_sample_count(),
installation_id: String::new(),
account_id: "default".into(),
last_successful_contact: None,
@@ -47,6 +57,13 @@ pub struct RuntimeSettings {
pub zone_interval_seconds: u64,
pub discovery_timeout_ms: u64,
pub discovery_broadcast: String,
/// Background Local/LAN round-trip quality worker.
#[serde(default = "default_true")]
pub ping_metrics_enabled: bool,
#[serde(default = "default_ping_interval_seconds")]
pub ping_interval_seconds: u64,
#[serde(default = "default_ping_sample_count")]
pub ping_sample_count: u32,
/// Global seasonal mode. Zones follow this by default. Values: cool/heat/off; off pauses house-level thermostat control.
#[serde(default = "default_house_mode")]
pub house_mode: String,
src/models/settings_api.rs
+15
View File
@@ -10,6 +10,12 @@ pub struct GreeSettings {
pub zone_interval_seconds: u64,
pub discovery_timeout_ms: u64,
pub discovery_broadcast: String,
#[serde(default = "default_true")]
pub ping_metrics_enabled: bool,
#[serde(default = "default_ping_interval_seconds")]
pub ping_interval_seconds: u64,
#[serde(default = "default_ping_sample_count")]
pub ping_sample_count: u32,
pub suppress_device_beep: bool,
pub compressor_protection_enabled: bool,
pub compressor_protection_seconds: u64,
@@ -23,6 +29,12 @@ pub struct GreeCloudSettingsUpdate {
#[serde(default)]
pub password: Option<String>,
pub polling_interval_seconds: u64,
#[serde(default)]
pub connectivity_metrics_enabled: bool,
#[serde(default = "default_cloud_metrics_interval_seconds")]
pub connectivity_metrics_interval_seconds: u64,
#[serde(default = "default_ping_sample_count")]
pub connectivity_metrics_sample_count: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -32,6 +44,9 @@ pub struct GreeCloudSettingsView {
pub username: String,
pub password_configured: bool,
pub polling_interval_seconds: u64,
pub connectivity_metrics_enabled: bool,
pub connectivity_metrics_interval_seconds: u64,
pub connectivity_metrics_sample_count: u32,
pub installation_id: String,
pub account_id: String,
pub last_successful_contact: Option<DateTime<Utc>>,
src/protocol/gree_cloud_mqtt.rs
+30 -1
View File
@@ -5,7 +5,7 @@ use std::{
atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering},
Arc,
},
time::{Duration, SystemTime, UNIX_EPOCH},
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tokio::{
io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
@@ -81,6 +81,7 @@ enum WireCommand {
pub struct MqttConnection {
tx: mpsc::Sender<WireCommand>,
connected: Arc<AtomicBool>,
events: broadcast::Sender<MqttEvent>,
}
impl MqttConnection {
@@ -168,6 +169,7 @@ impl MqttConnection {
Ok(Self {
tx,
connected,
events,
})
}
@@ -211,6 +213,33 @@ impl MqttConnection {
.map_err(|_| anyhow!("GREE Cloud MQTT writer stopped"))
}
pub async fn ping_round_trip(&self) -> Result<u64> {
if !self.is_connected() {
bail!("GREE Cloud MQTT is not connected");
}
let mut events = self.events.subscribe();
let started = Instant::now();
timeout(MQTT_QUEUE_TIMEOUT, self.tx.send(WireCommand::Ping))
.await
.context("GREE Cloud MQTT ping queue timed out")?
.map_err(|_| anyhow!("GREE Cloud MQTT writer stopped"))?;
let response = timeout(Duration::from_secs(5), async {
loop {
match events.recv().await {
Ok(MqttEvent::Traffic { kind: "PINGRESP" }) => break Ok(()),
Ok(MqttEvent::Disconnected { reason }) => break Err(anyhow!("{}", reason)),
Ok(_) => {}
Err(broadcast::error::RecvError::Lagged(_)) => {}
Err(broadcast::error::RecvError::Closed) => break Err(anyhow!("GREE Cloud MQTT event stream closed")),
}
}
})
.await
.context("GREE Cloud MQTT PINGRESP timed out")?;
response?;
Ok(started.elapsed().as_millis().min(u64::MAX as u128) as u64)
}
pub async fn disconnect(&self) {
// Mark disconnected first so no new work can enter the queue while shutdown is in
// progress. A wedged/full writer queue must never prevent process termination.
src/provider.rs
+6
View File
@@ -614,6 +614,12 @@ impl GreeCloudProvider {
.ok_or_else(|| anyhow!("GREE Cloud MQTT is disconnected"))
}
pub async fn probe_mqtt(&self, settings: &GreeCloudSettings, devices: &[Device]) -> Result<u64> {
self.ensure_connected(settings, devices).await?;
let session = self.current_session().await?;
session.mqtt.ping_round_trip().await
}
pub async fn ensure_connected(&self, settings: &GreeCloudSettings, devices: &[Device]) -> Result<()> {
if !settings.enabled {
bail!("GREE Cloud is disabled");
src/queries/schema.rs
+20
View File
@@ -129,6 +129,24 @@ CREATE INDEX IF NOT EXISTS energy_readings_device_time_idx
CREATE INDEX IF NOT EXISTS energy_readings_source_time_idx
ON energy_readings(device_id, source, timestamp DESC);
CREATE TABLE IF NOT EXISTS network_readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
target_id TEXT NOT NULL,
target_kind TEXT NOT NULL,
timestamp TEXT NOT NULL,
latency_ms REAL,
jitter_ms REAL,
packet_loss_pct REAL NOT NULL CHECK(packet_loss_pct >= 0 AND packet_loss_pct <= 100),
sample_count INTEGER NOT NULL,
successful_samples INTEGER NOT NULL,
source TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS network_readings_target_time_idx
ON network_readings(target_id, timestamp DESC);
CREATE INDEX IF NOT EXISTS network_readings_source_time_idx
ON network_readings(source, timestamp DESC);
CREATE TABLE IF NOT EXISTS event_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
@@ -163,5 +181,7 @@ INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (7, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (8, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (9, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
"#;