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

v0.7.13

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-08-27 22:00:13 +02:00
parent 4cc4378588
commit b93a1f2d92
16 changed files with 196 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api.rs
+16
View File
@@ -208,6 +208,7 @@ async fn bootstrap(State(state): State<AppState>) -> Result<Json<Value>, AppErro
async fn build_bootstrap(state: &AppState) -> Result<Value, AppError> {
let settings = state.settings.read().await.clone();
let (received_frames_total, received_frames_by_device) = state.gree.received_frame_stats();
Ok(json!({
"devices": state.db.list_devices()?,
"zones": state.db.list_zones()?,
@@ -222,12 +223,15 @@ async fn build_bootstrap(state: &AppState) -> Result<Value, AppError> {
"uptime_seconds": state.started.elapsed().as_secs(),
"auth_required": !state.config.app_token.trim().is_empty(),
"control_ready": state.initial_device_sync_complete.load(Ordering::Acquire),
"gree_received_frames": received_frames_total,
"gree_received_frames_by_device": received_frames_by_device,
}
}))
}
async fn system_info(State(state): State<AppState>) -> Result<Json<Value>, AppError> {
let devices = state.db.list_devices()?;
let (received_frames_total, received_frames_by_device) = state.gree.received_frame_stats();
Ok(Json(json!({
"version": env!("CARGO_PKG_VERSION"),
"uptime_seconds": state.started.elapsed().as_secs(),
@@ -238,6 +242,8 @@ async fn system_info(State(state): State<AppState>) -> Result<Json<Value>, AppEr
"control_ready": state.initial_device_sync_complete.load(Ordering::Acquire),
"bind": state.config.bind.to_string(),
"gree_interface": if state.config.gree_interface.trim().is_empty() { "auto" } else { state.config.gree_interface.trim() },
"gree_received_frames": received_frames_total,
"gree_received_frames_by_device": received_frames_by_device,
})))
}
@@ -607,6 +613,14 @@ async fn update_zone(State(state): State<AppState>, Path(id): Path<String>, Json
}
async fn apply_zone_control_patch(state: &AppState, id: &str, patch: ZoneControlPatch) -> Result<Zone, AppError> {
// Serialize quick-thermostat changes with the same device lock used by GREE polling and
// manual-takeover detection. Without this, a poll that started just before a Web/HA
// thermostat action could save an older zone snapshot afterwards and resurrect a false
// "physical/pilot" takeover.
let device_id = state.db.get_zone(id)?
.ok_or_else(|| AppError::NotFound(format!("zone {id}")))?
.device_id;
let device_guard = state.lock_device_operation(&device_id).await;
let mut zone = state.db.get_zone(id)?.ok_or_else(|| AppError::NotFound(format!("zone {id}")))?;
let was_enabled = zone.enabled;
let schedules = state.db.list_schedules()?;
@@ -687,6 +701,8 @@ async fn apply_zone_control_patch(state: &AppState, id: &str, patch: ZoneControl
zone.updated_at = Utc::now();
state.db.save_zone(&zone)?;
state.broadcast("zone.updated", serde_json::to_value(&zone)?);
// Release before any device command below; engine::send_command acquires this same lock.
drop(device_guard);
if was_enabled && !zone.enabled {
power_off_zone_device(state, &zone, "zone.quick_disabled").await;
} else if patch.power == Some(false) {
src/engine.rs
+37 -37
View File
@@ -149,6 +149,7 @@ async fn send_command_locked_inner(
// detach) may bypass cache de-duplication so stale state cannot leave a unit powered.
let command = if dedupe_against_cache && device.online && device.communication_failures == 0 { command.changed_from(&device) } else { command };
if command.is_empty() { return Ok(device); }
let controller_command_baseline = device.clone();
let suppress_beep = state.settings.read().await.suppress_device_beep;
let response_started = Instant::now();
@@ -264,21 +265,19 @@ async fn send_command_locked_inner(
state.db.save_device(&device)?;
if !dedupe_against_cache && confirmed_state && !confirmed_requested_state {
if track_controller_command && !command_manual_control_fields(&applied_command).is_empty() {
remember_controller_command(state, device_id, &applied_command).await;
remember_controller_command(state, device_id, &applied_command, &controller_command_baseline).await;
}
state.broadcast("device.updated", serde_json::to_value(&device).unwrap_or_default());
return Err(AppError::Device("device did not confirm the requested forced state change".into()));
}
}
if track_controller_command {
if confirmed_state && confirmed_requested_state {
// A verified full status snapshot supersedes any older unsettled expectation for
// this device, so it must not mask a later real remote change.
clear_pending_controller_command(state, device_id).await;
} else if !command_manual_control_fields(&applied_command).is_empty() {
remember_controller_command(state, device_id, &applied_command).await;
}
if track_controller_command && !command_manual_control_fields(&applied_command).is_empty() {
// Keep a bounded settling history even after the requested state has already been
// observed once. Several GREE modules can briefly publish an older snapshot again
// and then return to the controller-requested state. Without this guard that normal
// firmware bounce can be misclassified as a physical/pilot takeover.
remember_controller_command(state, device_id, &applied_command, &controller_command_baseline).await;
}
state.log("info", "device.command", &format!("Updated {}", device.name), json!({
@@ -476,42 +475,41 @@ fn command_manual_control_fields(command: &DeviceCommand) -> Vec<String> {
fields
}
fn merge_device_command(base: &mut DeviceCommand, update: &DeviceCommand) {
if update.power.is_some() { base.power = update.power; }
if update.mode.is_some() { base.mode = update.mode.clone(); }
if update.target_temperature.is_some() { base.target_temperature = update.target_temperature; }
if update.fan_speed.is_some() { base.fan_speed = update.fan_speed; }
if update.swing_vertical.is_some() { base.swing_vertical = update.swing_vertical; }
if update.swing_horizontal.is_some() { base.swing_horizontal = update.swing_horizontal; }
if update.quiet.is_some() { base.quiet = update.quiet; }
if update.turbo.is_some() { base.turbo = update.turbo; }
if update.light.is_some() { base.light = update.light; }
if update.air.is_some() { base.air = update.air; }
if update.xfan.is_some() { base.xfan = update.xfan; }
if update.health.is_some() { base.health = update.health; }
if update.sleep.is_some() { base.sleep = update.sleep; }
fn command_baseline_from_device(command: &DeviceCommand, device: &Device) -> DeviceCommand {
DeviceCommand {
power: command.power.map(|_| device.power),
mode: command.mode.as_ref().map(|_| device.mode.clone()),
target_temperature: command.target_temperature.map(|_| device.target_temperature),
fan_speed: command.fan_speed.map(|_| device.fan_speed),
quiet: command.quiet.map(|_| device.quiet),
sleep: command.sleep.map(|_| device.sleep),
..Default::default()
}
}
async fn remember_controller_command(state: &AppState, device_id: &str, command: &DeviceCommand) {
async fn remember_controller_command(state: &AppState, device_id: &str, command: &DeviceCommand, baseline_device: &Device) {
let poll_seconds = state.settings.read().await.poll_interval_seconds.max(2);
let ttl = Duration::from_secs(poll_seconds.saturating_mul(2).saturating_add(5).min(120));
let mut pending = state.pending_controller_commands.lock().await;
let expires_at = Instant::now() + ttl;
let baseline = command_baseline_from_device(command, baseline_device);
if let Some(existing) = pending.get_mut(device_id) {
merge_device_command(&mut existing.command, command);
existing.commands.push(command.clone());
existing.baselines.push(baseline);
// The history only spans one settling window; cap it defensively so a noisy device
// cannot grow this allocation without bound.
if existing.commands.len() > 8 { existing.commands.remove(0); }
if existing.baselines.len() > 8 { existing.baselines.remove(0); }
existing.expires_at = expires_at;
} else {
pending.insert(device_id.to_string(), PendingControllerCommand {
command: command.clone(),
commands: vec![command.clone()],
baselines: vec![baseline],
expires_at,
});
}
}
async fn clear_pending_controller_command(state: &AppState, device_id: &str) {
state.pending_controller_commands.lock().await.remove(device_id);
}
fn command_field_matches_device(command: &DeviceCommand, field: &str, device: &Device) -> bool {
match field {
"power" => command.power.map(|value| value == device.power).unwrap_or(false),
@@ -542,14 +540,16 @@ async fn suppress_expected_controller_changes(
}
let Some(expected) = pending.get(&device.id).cloned() else { return fields; };
let filtered = fields.into_iter()
.filter(|field| !command_field_matches_device(&expected.command, field, device))
.filter(|field| {
let matches_recent_controller_state = expected.commands.iter()
.chain(expected.baselines.iter())
.any(|command| command_field_matches_device(command, field, device));
!matches_recent_controller_state
})
.collect();
let expected_fields = command_manual_control_fields(&expected.command);
if !expected_fields.is_empty()
&& expected_fields.iter().all(|field| command_field_matches_device(&expected.command, field, device))
{
pending.remove(&device.id);
}
// Do not clear the settling guard merely because one poll matched the requested state.
// A later status packet can still briefly roll back to the pre-command snapshot. The
// bounded TTL is what ends this ambiguity window.
filtered
}
src/protocol/gree.rs
+36 -1
View File
@@ -1,4 +1,4 @@
use std::{collections::HashSet, net::{Ipv4Addr, SocketAddr, SocketAddrV4}, sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}}, time::Duration};
use std::{collections::{HashMap, HashSet}, net::{Ipv4Addr, SocketAddr, SocketAddrV4}, sync::{Arc, Mutex, atomic::{AtomicBool, AtomicU64, Ordering}}, time::Duration};
use anyhow::{anyhow, bail, Context, Result};
use chrono::Utc;
use serde_json::{json, Value};
@@ -22,6 +22,8 @@ pub struct GreeClient {
interface: Option<String>,
debug_events: Option<broadcast::Sender<ApiEvent>>,
debug_gree_frames: Arc<AtomicBool>,
received_frames_total: Arc<AtomicU64>,
received_frames_by_device: Arc<Mutex<HashMap<String, u64>>>,
buzzer_unsupported: Arc<Mutex<HashSet<String>>>,
quiet_unsupported: Arc<Mutex<HashSet<String>>>,
sleep_unsupported: Arc<Mutex<HashSet<String>>>,
@@ -39,12 +41,43 @@ impl GreeClient {
interface,
debug_events,
debug_gree_frames,
received_frames_total: Arc::new(AtomicU64::new(0)),
received_frames_by_device: Arc::new(Mutex::new(HashMap::new())),
buzzer_unsupported: Arc::new(Mutex::new(HashSet::new())),
quiet_unsupported: Arc::new(Mutex::new(HashSet::new())),
sleep_unsupported: Arc::new(Mutex::new(HashSet::new())),
}
}
pub fn received_frame_stats(&self) -> (u64, HashMap<String, u64>) {
let total = self.received_frames_total.load(Ordering::Relaxed);
let by_device = self.received_frames_by_device.lock()
.map(|counts| counts.clone())
.unwrap_or_default();
(total, by_device)
}
fn record_received_frame(&self, device: &Device) {
let total = self.received_frames_total.fetch_add(1, Ordering::Relaxed).saturating_add(1);
let device_count = self.received_frames_by_device.lock().ok().map(|mut counts| {
let count = counts.entry(device.id.clone()).or_insert(0);
*count = (*count).saturating_add(1);
*count
}).unwrap_or(0);
if let Some(events) = &self.debug_events {
let _ = events.send(ApiEvent {
event: "gree.frame_received".into(),
timestamp: Utc::now(),
data: json!({
"device_id": device.id,
"device_name": device.name,
"total": total,
"device_count": device_count,
}),
});
}
}
fn debug_frame(&self, direction: &str, device: &Device, target: SocketAddr, protocol: u8, payload: &Value) {
if !self.debug_gree_frames.load(Ordering::Relaxed) { return; }
let Some(events) = &self.debug_events else { return; };
@@ -323,6 +356,7 @@ impl GreeClient {
let remaining = scan_deadline.saturating_duration_since(Instant::now());
match timeout(remaining, socket.recv_from(&mut scan_buf)).await {
Ok(Ok((_size, source))) if source.ip() == target.ip() => {
self.record_received_frame(device);
tracing::debug!(device=%device.id, source=%source, "Received scan response immediately before bind");
break;
}
@@ -667,6 +701,7 @@ impl GreeClient {
Err(_) => break,
};
if source.ip() != target.ip() { continue; }
self.record_received_frame(device);
let response: Value = match serde_json::from_slice(&buffer[..size]) {
Ok(value) => value,
Err(err) => { last_decode_error = Some(anyhow!("invalid GREE JSON response: {err}")); continue; }
src/state.rs
+6 -1
View File
@@ -6,7 +6,12 @@ use crate::{config::Config, db::Db, models::{ApiEvent, DeviceCommand, RuntimeSet
#[derive(Debug, Clone)]
pub(crate) struct PendingControllerCommand {
pub command: DeviceCommand,
/// Recent controller-requested values for climate fields. Keep a short history because
/// some GREE modules expose intermediate/out-of-order status snapshots while settling.
pub commands: Vec<DeviceCommand>,
/// Matching pre-command values for the same fields. A brief rollback to one of these
/// values is firmware settling, not necessarily a physical-remote takeover.
pub baselines: Vec<DeviceCommand>,
pub expires_at: Instant,
}