gru/gree-controller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0437ef7b6fa98046e5f5902f4c769bb24f8a4aa7
gree-controller/src/provider.rs
T

1936 lines
81 KiB
Rust
Raw Blame History

use crate::{
models::{ApiEvent, ConnectionStatus, ConnectionType, Device, DeviceCommand, GreeCloudSettings},
protocol::{
crypto::{decrypt_v1, decrypt_v2, encrypt_v1, encrypt_v2},
gree::{BindResult, GreeClient},
gree_cloud::{parent_mac, CloudCredentials, GreeCloudApi},
gree_cloud_mqtt::{broker_for_region, MqttConnection, MqttDeviceEnvelope, MqttEvent, MQTT_PORT},
},
};
use anyhow::{anyhow, bail, Context, Result};
use chrono::{DateTime, Utc};
use rand::Rng;
use serde::Serialize;
use serde_json::{json, Value};
use std::{
collections::{BTreeMap, HashMap, HashSet},
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc,
},
time::{Duration, Instant},
};
use tokio::{
sync::{broadcast, oneshot, Mutex, Notify, RwLock, Semaphore},
time::{sleep, timeout},
};
const CLOUD_COMMAND_TIMEOUT: Duration = Duration::from_secs(2);
const CLOUD_STATUS_TIMEOUT: Duration = Duration::from_secs(10);
const CLOUD_COMPAT_STATUS_TIMEOUT: Duration = Duration::from_secs(6);
const CLOUD_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
const CLOUD_MAX_CONCURRENT_REQUESTS: usize = 8;
const CLOUD_QUEUE_WAIT_TIMEOUT: Duration = Duration::from_secs(2);
const CLOUD_DEVICE_LOCK_TIMEOUT: Duration = Duration::from_secs(2);
const KNOWN_CLOUD_PROPERTIES: &[&str] = &[
"Pow", "Mod", "Dwet", "DwatSen", "Dfltr", "DwatFul", "Dmod", "SetTem", "TemSen",
"TemUn", "TemRec", "HalfTemEn", "SetDeciTem", "Add0.5", "WdSpd", "Air", "Blo",
"Health", "SwhSlp", "SlpMod", "Lig", "SwingLfRig", "SwUpDn", "Quiet", "Tur", "StHt",
"SvSt", "HeatCoolType", "hid", "ElcAll", "CompressorFqy",
];
// Property set used by older GREE Wi-Fi modules before the dehumidifier,
// half-degree and cloud energy fields were added. Some V1.x modules silently
// drop a whole status request when it contains an unknown column, so probing
// with the modern superset can look exactly like a broken MQTT connection.
// This list matches the long-standing greeclimate status schema used by those
// modules and intentionally excludes newer optional fields.
const LEGACY_CLOUD_PROPERTIES: &[&str] = &[
"Pow", "Mod", "SetTem", "TemSen", "TemUn", "TemRec", "WdSpd", "Air", "Blo",
"Health", "SwhSlp", "SlpMod", "Lig", "SwingLfRig", "SwUpDn", "Quiet", "Tur",
"StHt", "SvSt", "HeatCoolType",
];
#[derive(Clone)]
pub struct LocalProvider {
client: GreeClient,
}
impl LocalProvider {
pub fn new(client: GreeClient) -> Self {
Self { client }
}
pub fn client(&self) -> &GreeClient {
&self.client
}
pub async fn bind(&self, device: &Device) -> Result<BindResult> {
debug_assert_eq!(device.connection_type, ConnectionType::Local);
self.client.bind(device).await
}
pub async fn poll(&self, device: &mut Device) -> Result<()> {
debug_assert_eq!(device.connection_type, ConnectionType::Local);
self.client.poll(device).await
}
pub async fn command(
&self,
device: &Device,
command: &DeviceCommand,
suppress_beep: bool,
) -> Result<DeviceCommand> {
debug_assert_eq!(device.connection_type, ConnectionType::Local);
self.client.command(device, command, suppress_beep).await
}
}
#[derive(Debug, Clone, Serialize, Default)]
pub struct CloudDiagnostics {
pub connection_state: String,
pub mqtt_state: String,
pub last_connect: Option<DateTime<Utc>>,
pub last_disconnect: Option<DateTime<Utc>>,
pub reconnect_count: u64,
pub last_message_timestamp: Option<DateTime<Utc>>,
pub last_status_timestamp: Option<DateTime<Utc>>,
pub last_request_timestamp: Option<DateTime<Utc>>,
pub last_response_timestamp: Option<DateTime<Utc>>,
pub last_response_time_ms: Option<u64>,
pub last_request_kind: Option<String>,
pub last_request_topic: Option<String>,
pub last_mqtt_topic: Option<String>,
pub requests_sent: u64,
pub responses_received: u64,
pub request_timeouts: u64,
pub selected_cipher_version: Option<u8>,
pub status_profile: Option<String>,
pub wire_device_id: Option<String>,
pub rest_device_online: Option<bool>,
pub fresh_discovery_key_used: Option<bool>,
pub saved_key_changed: Option<bool>,
pub broker_host: Option<String>,
pub subscribed_topics: Vec<String>,
pub raw_device_properties: BTreeMap<String, Value>,
pub parsed_device_properties: BTreeMap<String, Value>,
pub unknown_properties: BTreeMap<String, Value>,
pub energy_related_properties: BTreeMap<String, Value>,
pub last_error: Option<String>,
}
#[derive(Debug, Clone)]
pub struct CloudPushEvent {
pub cloud_device_id: Option<String>,
pub parent_mac: String,
pub properties: BTreeMap<String, Value>,
pub connected: Option<bool>,
pub cipher_version: Option<u8>,
}
#[derive(Clone)]
struct RegisteredDevice {
device_id: String,
key: String,
parent_mac: String,
cipher_version: u8,
}
#[derive(Clone)]
struct CloudSession {
credentials: CloudCredentials,
mqtt: MqttConnection,
broker: String,
connected_at: DateTime<Utc>,
/// Exact MAC spelling returned by GREE REST, keyed by normalized device id.
/// MQTT topic names are case-sensitive, so this must not be reconstructed from
/// the application's upper-case display/stable identifier.
wire_macs: HashMap<String, String>,
/// Fresh per-device key returned by the current REST discovery. Keep this only
/// in backend memory; it is never serialized to diagnostics or frontend APIs.
/// The reference greeclimate cloud path recreates CloudDevice from this fresh
/// key on every setup instead of relying on a previously persisted value.
wire_keys: HashMap<String, String>,
}
#[derive(Debug, Clone)]
struct ParsedCloudMessage {
properties: BTreeMap<String, Value>,
cipher_version: u8,
}
struct CloudInner {
http: reqwest::Client,
session: RwLock<Option<CloudSession>>,
connecting: AtomicBool,
connect_notify: Notify,
raw_events: broadcast::Sender<MqttEvent>,
push_events: broadcast::Sender<CloudPushEvent>,
api_events: Option<broadcast::Sender<ApiEvent>>,
debug_cloud_requests: Arc<AtomicBool>,
debug_cloud_mqtt: Arc<AtomicBool>,
registered: RwLock<HashMap<String, RegisteredDevice>>,
pending: Mutex<HashMap<String, oneshot::Sender<ParsedCloudMessage>>>,
command_locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
diagnostics: RwLock<HashMap<String, CloudDiagnostics>>,
last_mqtt_message: RwLock<Option<DateTime<Utc>>>,
reconnects: AtomicU64,
request_limit: Semaphore,
}
#[derive(Clone)]
pub struct GreeCloudProvider {
inner: Arc<CloudInner>,
}
impl GreeCloudProvider {
#[cfg(test)]
pub fn new(http: reqwest::Client) -> Self {
Self::new_with_debug(
http,
None,
Arc::new(AtomicBool::new(false)),
Arc::new(AtomicBool::new(false)),
)
}
pub fn new_with_debug(
http: reqwest::Client,
api_events: Option<broadcast::Sender<ApiEvent>>,
debug_cloud_requests: Arc<AtomicBool>,
debug_cloud_mqtt: Arc<AtomicBool>,
) -> Self {
let (raw_events, _) = broadcast::channel(512);
let (push_events, _) = broadcast::channel(512);
let inner = Arc::new(CloudInner {
http,
session: RwLock::new(None),
connecting: AtomicBool::new(false),
connect_notify: Notify::new(),
raw_events,
push_events,
api_events,
debug_cloud_requests,
debug_cloud_mqtt,
registered: RwLock::new(HashMap::new()),
pending: Mutex::new(HashMap::new()),
command_locks: Mutex::new(HashMap::new()),
diagnostics: RwLock::new(HashMap::new()),
last_mqtt_message: RwLock::new(None),
reconnects: AtomicU64::new(0),
request_limit: Semaphore::new(CLOUD_MAX_CONCURRENT_REQUESTS),
});
let provider = Self { inner };
provider.spawn_message_router();
provider
}
pub fn subscribe_push(&self) -> broadcast::Receiver<CloudPushEvent> {
self.inner.push_events.subscribe()
}
pub async fn diagnostics(&self, device_id: &str) -> CloudDiagnostics {
self.inner
.diagnostics
.read()
.await
.get(device_id)
.cloned()
.unwrap_or_default()
}
pub async fn is_connected(&self) -> bool {
self.inner
.session
.read()
.await
.as_ref()
.is_some_and(|session| session.mqtt.is_connected())
}
pub async fn shutdown(&self) {
// Shutdown must stay bounded even if a Cloud writer or broker is wedged. Dropping
// pending response senders wakes request waiters immediately instead of making the
// HTTP server's graceful shutdown wait for their device timeouts.
let session = match timeout(Duration::from_secs(1), self.inner.session.write()).await {
Ok(mut guard) => guard.take(),
Err(_) => {
tracing::warn!("GREE Cloud session lock did not drain during shutdown");
None
}
};
if let Ok(mut pending) = timeout(Duration::from_secs(1), self.inner.pending.lock()).await {
pending.clear();
}
if let Some(session) = session {
let _ = timeout(Duration::from_secs(1), session.mqtt.disconnect()).await;
}
}
pub async fn unregister_device(&self, device_id: &str) {
let (cloud_ids, no_registered_devices) = {
let mut registered = self.inner.registered.write().await;
let ids = registered
.iter()
.filter(|(_, item)| item.device_id == device_id)
.map(|(cloud_id, _)| cloud_id.clone())
.collect::<Vec<_>>();
registered.retain(|_, item| item.device_id != device_id);
let empty = registered.is_empty();
(ids, empty)
};
let mut pending = self.inner.pending.lock().await;
for cloud_id in cloud_ids {
pending.remove(&cloud_id);
}
drop(pending);
self.inner.command_locks.lock().await.remove(device_id);
self.inner.diagnostics.write().await.remove(device_id);
// The broker can continue publishing retained/status traffic for topics from the
// previous subscription until the MQTT session is closed. Once the last registered
// Cloud device is removed there is nothing useful to receive, so close the session
// immediately instead of leaving a stale subscription alive until process restart.
if no_registered_devices {
let session = self.inner.session.write().await.take();
if let Some(session) = session {
session.mqtt.disconnect().await;
tracing::info!("GREE Cloud MQTT disconnected; no registered devices remain");
}
}
}
pub async fn poll(
&self,
settings: &GreeCloudSettings,
all_devices: &[Device],
device: &mut Device,
) -> Result<()> {
self.register_device(device).await?;
self.ensure_connected(settings, all_devices).await?;
let lock = self.command_lock(&device.id).await;
let _guard = timeout(CLOUD_DEVICE_LOCK_TIMEOUT, lock.lock())
.await
.context("GREE Cloud device operation queue timed out")?;
let session = self.current_session().await?;
let parent = cloud_wire_parent(&session, device);
session.mqtt.subscribe_device(&parent).await?;
self.set_subscriptions(device, &session.broker, &parent).await;
let response = self.poll_status_compatible(device, &session).await?;
apply_cloud_capability_snapshot(device, &response.properties);
apply_cloud_properties(device, &response.properties);
device.protocol_version = response.cipher_version;
device.connection_status = ConnectionStatus::Online;
device.online = true;
device.communication_failures = 0;
device.last_error = None;
device.last_seen = Some(Utc::now());
device.last_cloud_sync = Some(Utc::now());
device.updated_at = Utc::now();
self.update_diagnostics_from_message(device, &response).await;
Ok(())
}
async fn poll_status_compatible(
&self,
device: &Device,
session: &CloudSession,
) -> Result<ParsedCloudMessage> {
let current_cipher = if device.protocol_version == 2 { 2 } else { 1 };
let current_wire = cloud_wire_mac(session, device);
let diagnostics = self.inner.diagnostics.read().await.get(&device.id).cloned();
let legacy_hint = is_legacy_cloud_firmware(&device.firmware)
|| diagnostics
.as_ref()
.and_then(|item| item.status_profile.as_deref())
== Some("legacy");
if !legacy_hint {
// One bounded request per poll. If a modern profile times out, mark the next
// attempt as legacy instead of running several compatibility probes back-to-back
// while higher layers are waiting on the same device lock.
let payload = json!({"t":"status", "cols": KNOWN_CLOUD_PROPERTIES});
match self
.request_with_options(
device,
session,
payload,
CLOUD_STATUS_TIMEOUT,
Some(current_cipher),
Some(&current_wire),
Some("full"),
)
.await
{
Ok(response) => {
self.remember_status_compatibility(device, &current_wire, "full")
.await;
return Ok(response);
}
Err(err) if !is_timeout_error(&err) => return Err(err),
Err(err) => {
self.inner
.diagnostics
.write()
.await
.entry(device.id.clone())
.or_default()
.status_profile = Some("legacy".into());
return Err(err).context(
"GREE Cloud status timed out; legacy compatibility will be tried on the next poll",
);
}
}
}
// Old modules need compatibility probing, but never probe all variants in one
// operation. Rotate one safe status-read variant per polling interval. This keeps an
// offline unit bounded to a single ~6s network wait and prevents it from monopolizing
// application/zone/device locks for 25-30 seconds at a time.
let step = diagnostics
.as_ref()
.map(|item| item.request_timeouts % 4)
.unwrap_or(0);
let alternate_cipher = if current_cipher == 2 { 1 } else { 2 };
let alternate_wire = alternate_wire_mac_case(&current_wire);
let (cipher, wire, profile) = match step {
1 => (alternate_cipher, current_wire.clone(), "legacy-cipher-probe"),
2 => (
current_cipher,
alternate_wire.clone().unwrap_or_else(|| current_wire.clone()),
"legacy-case-probe",
),
3 => (
alternate_cipher,
alternate_wire.unwrap_or_else(|| current_wire.clone()),
"legacy-case-cipher-probe",
),
_ => (current_cipher, current_wire.clone(), "legacy"),
};
let parent = parent_mac_preserve_case(&wire);
session.mqtt.subscribe_device(&parent).await?;
let payload = json!({"t":"status", "cols": LEGACY_CLOUD_PROPERTIES});
let response = self
.request_with_options(
device,
session,
payload,
CLOUD_COMPAT_STATUS_TIMEOUT,
Some(cipher),
Some(&wire),
Some(profile),
)
.await?;
self.remember_status_compatibility(device, &wire, "legacy")
.await;
if wire != current_wire {
self.set_subscriptions(device, &session.broker, &parent).await;
}
Ok(response)
}
async fn remember_status_compatibility(
&self,
device: &Device,
wire_mac: &str,
profile: &str,
) {
let stable = cloud_id(device);
if let Some(session) = self.inner.session.write().await.as_mut() {
session.wire_macs.insert(stable, wire_mac.to_string());
}
let mut diagnostics = self.inner.diagnostics.write().await;
diagnostics
.entry(device.id.clone())
.or_default()
.status_profile = Some(profile.to_string());
}
pub async fn command(
&self,
settings: &GreeCloudSettings,
all_devices: &[Device],
device: &mut Device,
command: &DeviceCommand,
suppress_beep: bool,
) -> Result<DeviceCommand> {
self.register_device(device).await?;
self.ensure_connected(settings, all_devices).await?;
let lock = self.command_lock(&device.id).await;
let _guard = timeout(CLOUD_DEVICE_LOCK_TIMEOUT, lock.lock())
.await
.context("GREE Cloud device operation queue timed out")?;
let session = self.current_session().await?;
let parent = cloud_wire_parent(&session, device);
session.mqtt.subscribe_device(&parent).await?;
self.set_subscriptions(device, &session.broker, &parent).await;
let commands = self.build_command_sequence(device, command, suppress_beep).await?;
let mut applied = DeviceCommand::default();
for (opt, values, fragment) in commands {
let payload = json!({"t":"cmd", "opt": opt, "p": values});
// greeclimate treats a 2s no-ACK as uncertain success. Preserve that semantic,
// then let the engine's bounded status verification decide the final UI state.
match self.request(device, &session, payload, CLOUD_COMMAND_TIMEOUT).await {
Ok(response) => {
self.update_diagnostics_from_message(device, &response).await;
fragment.apply(device);
merge_command(&mut applied, &fragment);
}
Err(err) if is_timeout_error(&err) => {
tracing::debug!(device=%device.id, "GREE Cloud command did not ACK within 2s; status verification will confirm it");
fragment.apply(device);
merge_command(&mut applied, &fragment);
}
Err(err) => return Err(err),
}
}
Ok(applied)
}
async fn build_command_sequence(
&self,
device: &Device,
command: &DeviceCommand,
suppress_beep: bool,
) -> Result<Vec<(Vec<String>, Vec<Value>, DeviceCommand)>> {
let mut out = Vec::new();
if let Some(mode) = command.mode.as_deref() {
let value = mode_to_wire(mode)?;
out.push((vec!["Mod".into()], vec![json!(value)], DeviceCommand { mode: Some(mode.into()), ..Default::default() }));
}
if let Some(target) = command.target_temperature {
let raw = self.raw_properties(&device.id).await;
let half_enabled = raw.get("HalfTemEn").and_then(value_i64) == Some(1);
let whole = target.floor() as i64;
let half = if target.fract() >= 0.5 { 1_i64 } else { 0_i64 };
let mut opt = vec!["SetTem".into(), "TemRec".into()];
let mut values = vec![json!(whole), json!(half)];
if half_enabled {
opt.push("SetDeciTem".into());
values.push(json!((target * 10.0).round() as i64));
opt.push("Add0.5".into());
values.push(json!(half));
}
out.push((opt, values, DeviceCommand { target_temperature: Some(target), ..Default::default() }));
}
let mut simple = Vec::new();
if let Some(value) = command.fan_speed { simple.push(("WdSpd", json!(value.min(5)), DeviceCommand { fan_speed: Some(value.min(5)), ..Default::default() })); }
if let Some(value) = command.swing_vertical { simple.push(("SwUpDn", json!(if value { 1 } else { 0 }), DeviceCommand { swing_vertical: Some(value), ..Default::default() })); }
if let Some(value) = command.swing_horizontal { simple.push(("SwingLfRig", json!(if value { 1 } else { 0 }), DeviceCommand { swing_horizontal: Some(value), ..Default::default() })); }
if let Some(value) = command.quiet {
let wire = if value { device.quiet_wire_value.unwrap_or(2).max(1) } else { 0 };
simple.push(("Quiet", json!(wire), DeviceCommand { quiet: Some(value), ..Default::default() }));
}
if let Some(value) = command.turbo { simple.push(("Tur", json!(u8::from(value)), DeviceCommand { turbo: Some(value), ..Default::default() })); }
if let Some(value) = command.light { simple.push(("Lig", json!(u8::from(value)), DeviceCommand { light: Some(value), ..Default::default() })); }
if let Some(value) = command.air { simple.push(("Air", json!(u8::from(value)), DeviceCommand { air: Some(value), ..Default::default() })); }
if let Some(value) = command.xfan { simple.push(("Blo", json!(u8::from(value)), DeviceCommand { xfan: Some(value), ..Default::default() })); }
if let Some(value) = command.health { simple.push(("Health", json!(u8::from(value)), DeviceCommand { health: Some(value), ..Default::default() })); }
if let Some(value) = command.sleep {
simple.push(("SwhSlp", json!(u8::from(value)), DeviceCommand { sleep: Some(value), ..Default::default() }));
simple.push(("SlpMod", json!(u8::from(value)), DeviceCommand::default()));
}
for (name, value, fragment) in simple {
out.push((vec![name.into()], vec![value], fragment));
}
if let Some(value) = command.power {
out.push((vec!["Pow".into()], vec![json!(u8::from(value))], DeviceCommand { power: Some(value), ..Default::default() }));
}
// greeclimate adds Buzzer_ON_OFF=1 to every state command when its shared
// buzzer preference is disabled. Some devices ignore the field, which is the
// reference behavior; it is not a separately reported Cloud capability.
if suppress_beep {
for (opt, values, _) in &mut out {
opt.push("Buzzer_ON_OFF".into());
values.push(json!(1));
}
}
Ok(out)
}
async fn raw_properties(&self, device_id: &str) -> BTreeMap<String, Value> {
self.inner
.diagnostics
.read()
.await
.get(device_id)
.map(|d| d.raw_device_properties.clone())
.unwrap_or_default()
}
async fn command_lock(&self, device_id: &str) -> Arc<Mutex<()>> {
let mut locks = self.inner.command_locks.lock().await;
locks
.entry(device_id.to_string())
.or_insert_with(|| Arc::new(Mutex::new(())))
.clone()
}
async fn register_device(&self, device: &Device) -> Result<()> {
if device.connection_type != ConnectionType::GreeCloud {
bail!("GREE Cloud provider received a Local device");
}
let cloud_id = cloud_id(device);
// Once a Cloud session exists, prefer the key and exact wire MAC from the
// current REST discovery. Calling register_device() before every poll/command
// must not overwrite a refreshed key with an older value persisted in SQLite.
let session = self.inner.session.read().await;
let session_key = session
.as_ref()
.and_then(|session| session.wire_keys.get(&cloud_id))
.cloned();
let session_parent = session
.as_ref()
.map(|session| cloud_wire_parent(session, device));
drop(session);
let key = session_key
.or_else(|| device.key.clone().filter(|value| !value.is_empty()))
.ok_or_else(|| anyhow!("GREE Cloud device key is missing"))?;
self.inner.registered.write().await.insert(
cloud_id,
RegisteredDevice {
device_id: device.id.clone(),
key,
parent_mac: session_parent.unwrap_or_else(|| cloud_parent(device)),
cipher_version: if device.protocol_version == 2 { 2 } else { 1 },
},
);
Ok(())
}
async fn current_session(&self) -> Result<CloudSession> {
self.inner
.session
.read()
.await
.as_ref()
.filter(|session| session.mqtt.is_connected())
.cloned()
.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");
}
if self.is_connected().await {
return Ok(());
}
// Create the notification future before racing for the connection leadership so a
// very fast connect/notify cannot be missed by concurrent callers.
let connection_finished = self.inner.connect_notify.notified();
if self
.inner
.connecting
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_err()
{
timeout(CLOUD_CONNECT_TIMEOUT + Duration::from_secs(5), connection_finished)
.await
.context("timed out waiting for GREE Cloud connection attempt")?;
return if self.is_connected().await {
Ok(())
} else {
Err(anyhow!("GREE Cloud MQTT connection failed"))
};
}
let result = match timeout(
CLOUD_CONNECT_TIMEOUT + Duration::from_secs(15),
self.connect_once(settings, devices),
)
.await
{
Ok(result) => result,
Err(_) => Err(anyhow!("GREE Cloud connection attempt timed out")),
};
self.inner.connecting.store(false, Ordering::Release);
self.inner.connect_notify.notify_waiters();
result
}
async fn connect_once(&self, settings: &GreeCloudSettings, devices: &[Device]) -> Result<()> {
let broker = broker_for_region(&settings.region)
.ok_or_else(|| anyhow!("unknown GREE Cloud region: {}", settings.region))?
.to_string();
tracing::info!(broker=%broker, "GREE Cloud MQTT connecting");
// Re-login on every new broker connection exactly like the HA integration. The token
// is then reused for the whole MQTT session and is never exposed through diagnostics.
let mut api = GreeCloudApi::for_region(
self.inner.http.clone(),
&settings.region,
&settings.username,
&settings.password,
)?;
let login_started = Instant::now();
self.emit_cloud_request(json!({
"phase": "sent",
"operation": "login",
"transport": "REST",
"region": settings.region,
}));
let credentials = match api.login().await {
Ok(credentials) => {
self.emit_cloud_request(json!({
"phase": "response",
"operation": "login",
"transport": "REST",
"region": settings.region,
"duration_ms": login_started.elapsed().as_millis().min(u64::MAX as u128) as u64,
"status": "ok",
}));
credentials
}
Err(err) => {
self.emit_cloud_request(json!({
"phase": "error",
"operation": "login",
"transport": "REST",
"region": settings.region,
"duration_ms": login_started.elapsed().as_millis().min(u64::MAX as u128) as u64,
"error": sanitize_error(&err.to_string()),
}));
return Err(err).context("GREE Cloud login failed");
}
};
tracing::info!("GREE Cloud login success");
// Preserve the exact MAC spelling returned by the Cloud API. MQTT topic names are
// case-sensitive; normalizing them to upper case (as the UI/stable ID does) can
// produce a perfectly connected MQTT session with zero device responses. This
// mirrors greeclimate, which passes CloudDeviceInfo.mac through unchanged.
let (wire_macs, wire_keys, rest_online): (
HashMap<String, String>,
HashMap<String, String>,
HashMap<String, bool>,
) = match api.get_all_devices().await {
Ok(discovered) => {
let mut macs = HashMap::new();
let mut keys = HashMap::new();
let mut online = HashMap::new();
for item in discovered {
let wire = item.mac.trim().replace([':', '-'], "");
let stable = normalize_mac(&wire);
macs.insert(stable.clone(), wire);
keys.insert(stable.clone(), item.key);
online.insert(stable, item.online);
}
(macs, keys, online)
}
Err(err) => {
tracing::warn!(error=%sanitize_error(&err.to_string()), "GREE Cloud MQTT device refresh failed; using persisted device metadata as fallback");
(HashMap::new(), HashMap::new(), HashMap::new())
}
};
let mqtt = MqttConnection::connect(
&broker,
MQTT_PORT,
credentials.user_id,
&credentials.token,
CLOUD_CONNECT_TIMEOUT,
self.inner.raw_events.clone(),
)
.await?;
let mut parents = HashSet::new();
let mut registration_failures = HashSet::new();
for device in devices
.iter()
.filter(|d| d.enabled && d.connection_type == ConnectionType::GreeCloud)
{
match self.register_device(device).await {
Ok(()) => {
let stable = cloud_id(device);
let wire = wire_macs
.get(&stable)
.cloned()
.unwrap_or_else(|| stable.to_ascii_lowercase());
let parent = parent_mac_preserve_case(&wire);
let fresh_key = wire_keys.get(&stable);
let saved_key_changed = fresh_key
.zip(device.key.as_ref())
.map(|(fresh, saved)| fresh != saved);
if let Some(registered) = self.inner.registered.write().await.get_mut(&stable) {
if let Some(fresh_key) = fresh_key {
registered.key = fresh_key.clone();
}
registered.parent_mac = parent.clone();
}
{
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.wire_device_id = Some(wire.clone());
item.rest_device_online = rest_online.get(&stable).copied();
item.fresh_discovery_key_used = Some(fresh_key.is_some());
item.saved_key_changed = saved_key_changed;
}
self.emit_cloud_request(json!({
"operation": "device_registration",
"phase": "ready",
"device_id": device.id,
"cloud_device_id": stable,
"wire_device_id": wire,
"wire_parent_mac": parent,
"rest_online": rest_online.get(&cloud_id(device)).copied(),
"fresh_discovery_key_used": fresh_key.is_some(),
"saved_key_changed": saved_key_changed,
}));
parents.insert(cloud_wire_parent_from_map(&wire_macs, device));
}
Err(err) => {
let message = sanitize_error(&err.to_string());
tracing::warn!(device=%device.id, error=%message, "GREE Cloud device registration skipped");
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.connection_state = "offline".into();
item.mqtt_state = "connected".into();
item.last_error = Some(message);
registration_failures.insert(device.id.clone());
}
}
}
for parent in &parents {
mqtt.subscribe_device(parent).await?;
}
let reconnect_count = self.inner.reconnects.fetch_add(1, Ordering::AcqRel);
*self.inner.session.write().await = Some(CloudSession {
credentials,
mqtt,
broker: broker.clone(),
connected_at: Utc::now(),
wire_macs,
wire_keys,
});
let now = Utc::now();
self.emit_cloud_mqtt(json!({
"direction": "connected",
"broker": broker,
"port": MQTT_PORT,
"parents": parents.len(),
}));
let mut diagnostics = self.inner.diagnostics.write().await;
for device in devices.iter().filter(|d| d.connection_type == ConnectionType::GreeCloud) {
let item = diagnostics.entry(device.id.clone()).or_default();
item.mqtt_state = "connected".into();
item.last_connect = Some(now);
item.reconnect_count = reconnect_count;
item.broker_host = Some(broker.clone());
if !registration_failures.contains(&device.id) {
item.connection_state = "connected".into();
item.last_error = None;
}
}
tracing::info!(broker=%broker, "GREE Cloud MQTT connected");
tracing::info!(count=parents.len(), "GREE Cloud MQTT subscriptions restored");
Ok(())
}
async fn request(
&self,
device: &Device,
session: &CloudSession,
inner_payload: Value,
wait: Duration,
) -> Result<ParsedCloudMessage> {
self.request_with_options(
device,
session,
inner_payload,
wait,
None,
None,
None,
)
.await
}
async fn request_with_options(
&self,
device: &Device,
session: &CloudSession,
inner_payload: Value,
wait: Duration,
cipher_override: Option<u8>,
wire_cloud_id_override: Option<&str>,
status_profile: Option<&str>,
) -> Result<ParsedCloudMessage> {
let _permit = timeout(CLOUD_QUEUE_WAIT_TIMEOUT, self.inner.request_limit.acquire())
.await
.context("GREE Cloud request queue timed out")??;
let cloud_id = cloud_id(device);
let wire_cloud_id = wire_cloud_id_override
.map(|value| value.trim().replace([':', '-'], ""))
.unwrap_or_else(|| cloud_wire_mac(session, device));
let wire_parent = parent_mac_preserve_case(&wire_cloud_id);
let key = session
.wire_keys
.get(&cloud_id)
.map(String::as_str)
.or(device.key.as_deref())
.ok_or_else(|| anyhow!("GREE Cloud device key is missing"))?;
let cipher = cipher_override.unwrap_or_else(|| if device.protocol_version == 2 { 2 } else { 1 });
let plaintext = serde_json::to_vec(&inner_payload)?;
let (pack, tag) = if cipher == 2 {
let encrypted = encrypt_v2(key, &plaintext)?;
(encrypted.ciphertext, Some(encrypted.tag))
} else {
(encrypt_v1(key, &plaintext)?, None)
};
let cid = rand::thread_rng()
.gen_range(1_000_000_000_u64..=9_999_999_999_u64)
.to_string();
let request_kind = inner_payload
.get("t")
.and_then(Value::as_str)
.unwrap_or("request")
.to_string();
let envelope = MqttDeviceEnvelope {
cid: cid.clone(),
i: 0,
pack,
t: "pack".into(),
tcid: wire_cloud_id.clone(),
uid: json!(session.credentials.user_id),
tag,
ts: None,
};
let payload = serde_json::to_vec(&envelope)?;
let (tx, rx) = oneshot::channel();
{
let mut pending = self.inner.pending.lock().await;
if pending.insert(cloud_id.clone(), tx).is_some() {
bail!("another GREE Cloud request is already pending for this device");
}
}
let topic = format!("request/{wire_parent}");
let started_at = Utc::now();
let started = Instant::now();
{
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.last_request_timestamp = Some(started_at);
item.last_request_kind = Some(request_kind.clone());
item.last_request_topic = Some(topic.clone());
if let Some(profile) = status_profile {
item.status_profile = Some(profile.to_string());
}
item.requests_sent = item.requests_sent.saturating_add(1);
}
self.emit_cloud_request(json!({
"phase": "sent",
"operation": request_kind,
"device_id": device.id,
"device_name": device.name,
"cloud_device_id": cloud_id,
"topic": topic,
"cid": cid,
"cipher_version": cipher,
"status_profile": status_profile,
"wire_device_id": wire_cloud_id,
"payload": inner_payload,
}));
self.emit_cloud_mqtt(json!({
"direction": "tx",
"topic": topic,
"device_id": device.id,
"cloud_device_id": cloud_id,
"cid": cid,
"tcid": envelope.tcid,
"qos": 1,
"bytes": payload.len(),
}));
if let Err(err) = session.mqtt.publish(topic.clone(), payload).await {
self.inner.pending.lock().await.remove(&cloud_id);
let duration_ms = started.elapsed().as_millis().min(u64::MAX as u128) as u64;
self.emit_cloud_request(json!({
"phase": "error",
"operation": request_kind,
"device_id": device.id,
"topic": topic,
"duration_ms": duration_ms,
"error": sanitize_error(&err.to_string()),
}));
return Err(err);
}
tracing::debug!(device=%device.id, kind=%request_kind, "GREE Cloud command/status request sent");
let response = timeout(wait, rx).await;
self.inner.pending.lock().await.remove(&cloud_id);
let duration_ms = started.elapsed().as_millis().min(u64::MAX as u128) as u64;
match response {
Ok(Ok(message)) => {
let now = Utc::now();
{
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.last_response_timestamp = Some(now);
item.last_response_time_ms = Some(duration_ms);
item.responses_received = item.responses_received.saturating_add(1);
}
self.emit_cloud_request(json!({
"phase": "response",
"operation": request_kind,
"device_id": device.id,
"cloud_device_id": cloud_id,
"topic": topic,
"duration_ms": duration_ms,
"properties": message.properties.keys().cloned().collect::<Vec<_>>(),
}));
Ok(message)
}
Ok(Err(_)) => {
self.emit_cloud_request(json!({
"phase": "error",
"operation": request_kind,
"device_id": device.id,
"topic": topic,
"duration_ms": duration_ms,
"error": "response waiter cancelled",
}));
bail!("GREE Cloud response waiter was cancelled")
}
Err(_) => {
{
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.request_timeouts = item.request_timeouts.saturating_add(1);
item.last_error = Some(format!("GREE Cloud {request_kind} request timed out"));
}
self.emit_cloud_request(json!({
"phase": "timeout",
"operation": request_kind,
"device_id": device.id,
"cloud_device_id": cloud_id,
"topic": topic,
"duration_ms": duration_ms,
}));
bail!("GREE Cloud request timed out")
}
}
}
fn emit_cloud_request(&self, data: Value) {
if !self.inner.debug_cloud_requests.load(Ordering::Relaxed) {
return;
}
if let Some(events) = &self.inner.api_events {
let _ = events.send(ApiEvent {
event: "gree_cloud.request".into(),
timestamp: Utc::now(),
data,
});
}
}
fn emit_cloud_mqtt(&self, data: Value) {
if !self.inner.debug_cloud_mqtt.load(Ordering::Relaxed) {
return;
}
if let Some(events) = &self.inner.api_events {
let _ = events.send(ApiEvent {
event: "gree_cloud.mqtt".into(),
timestamp: Utc::now(),
data,
});
}
}
pub async fn runtime_status(&self) -> Value {
let session = self.inner.session.read().await.clone();
let diagnostics = self.inner.diagnostics.read().await;
let last_mqtt_message = self.inner.last_mqtt_message.read().await.clone();
let mut last_device_response = None;
let mut last_response_time_ms = None;
let mut requests_sent = 0_u64;
let mut responses_received = 0_u64;
let mut request_timeouts = 0_u64;
for item in diagnostics.values() {
if item.last_response_timestamp > last_device_response {
last_device_response = item.last_response_timestamp;
last_response_time_ms = item.last_response_time_ms;
}
requests_sent = requests_sent.saturating_add(item.requests_sent);
responses_received = responses_received.saturating_add(item.responses_received);
request_timeouts = request_timeouts.saturating_add(item.request_timeouts);
}
json!({
"broker_host": session.as_ref().map(|value| value.broker.clone()),
"mqtt_connected_since": session.as_ref().map(|value| value.connected_at.clone()),
"last_mqtt_message": last_mqtt_message,
"last_device_response": last_device_response,
"last_response_time_ms": last_response_time_ms,
"requests_sent": requests_sent,
"responses_received": responses_received,
"request_timeouts": request_timeouts,
"reconnect_count": self.inner.reconnects.load(Ordering::Relaxed),
})
}
fn spawn_message_router(&self) {
let provider = self.clone();
let mut rx = self.inner.raw_events.subscribe();
tokio::spawn(async move {
loop {
match rx.recv().await {
Ok(MqttEvent::Message { topic, payload }) => {
if let Err(err) = provider.handle_raw_message(&topic, &payload).await {
tracing::warn!(error=?err, "GREE Cloud MQTT payload error");
}
}
Ok(MqttEvent::Traffic { kind }) => {
*provider.inner.last_mqtt_message.write().await = Some(Utc::now());
provider.emit_cloud_mqtt(json!({
"direction": "rx_protocol",
"kind": kind,
}));
}
Ok(MqttEvent::Disconnected { reason }) => {
provider.handle_disconnect(&reason).await;
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
tracing::warn!(skipped, "GREE Cloud MQTT event receiver lagged");
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
});
}
async fn handle_disconnect(&self, reason: &str) {
self.emit_cloud_mqtt(json!({
"direction": "disconnect",
"reason": sanitize_error(reason),
}));
*self.inner.session.write().await = None;
let now = Utc::now();
for item in self.inner.diagnostics.write().await.values_mut() {
item.connection_state = "cloud_disconnected".into();
item.mqtt_state = "disconnected".into();
item.last_disconnect = Some(now);
item.last_error = Some(sanitize_error(reason));
}
tracing::warn!(reason=%sanitize_error(reason), "GREE Cloud MQTT disconnected");
}
async fn handle_raw_message(&self, topic: &str, payload: &[u8]) -> Result<()> {
let now = Utc::now();
*self.inner.last_mqtt_message.write().await = Some(now);
self.emit_cloud_mqtt(json!({
"direction": "rx",
"topic": topic,
"bytes": payload.len(),
}));
// A broker message can race with device deletion. If the last Cloud device has
// already been unregistered, any in-flight message belongs to a stale subscription
// and must be ignored rather than reported as a payload error every few seconds.
let registered = self.inner.registered.read().await.clone();
if registered.is_empty() {
tracing::debug!(topic, "ignoring GREE Cloud MQTT payload without registered devices");
return Ok(());
}
if topic.starts_with("connect/") {
let parent = topic.split('/').nth(1).unwrap_or_default().to_string();
self.emit_cloud_mqtt(json!({
"direction": "device_connect",
"topic": topic,
"parent_mac": parent,
}));
let _ = self.inner.push_events.send(CloudPushEvent {
cloud_device_id: None,
parent_mac: parent,
properties: BTreeMap::new(),
connected: Some(true),
cipher_version: None,
});
return Ok(());
}
let envelope: MqttDeviceEnvelope = serde_json::from_slice(payload).context("invalid GREE Cloud MQTT JSON")?;
if envelope.pack.is_empty() {
bail!("GREE Cloud MQTT payload has no encrypted pack");
}
self.emit_cloud_mqtt(json!({
"direction": "rx_envelope",
"topic": topic,
"cid": envelope.cid.clone(),
"tcid": envelope.tcid.clone(),
"cipher": if envelope.tag.is_some() { 2 } else { 1 },
}));
let parent_from_topic = topic.split('/').nth(1).unwrap_or_default().to_string();
// Match the reference client primarily by subscribed parent topic. Some GREE
// responses use a parent/variant tcid that is not byte-for-byte equal to the
// discovery child id; restricting candidates to tcid caused valid frames to be
// discarded and the status request to time out. Prefer exact tcid first, then
// try every registered child under the topic parent.
let normalized_tcid = normalize_mac(&envelope.tcid);
let mut candidates: Vec<(String, RegisteredDevice)> = registered
.iter()
.filter(|(_, item)| item.parent_mac.eq_ignore_ascii_case(&parent_from_topic))
.map(|(id, item)| (id.clone(), item.clone()))
.collect();
candidates.sort_by_key(|(id, _)| if !normalized_tcid.is_empty() && id.eq_ignore_ascii_case(&normalized_tcid) { 0 } else { 1 });
if candidates.is_empty() {
bail!("GREE Cloud MQTT message does not match a registered device");
}
let mut last_error = None;
for (cloud_id, registered) in candidates {
let preferred = if envelope.tag.is_some() { 2 } else { registered.cipher_version };
let (plaintext, cipher_version) = match decrypt_cloud_payload(&registered.key, &envelope, preferred) {
Ok(value) => (value, preferred),
Err(first) => {
let alternate = if preferred == 2 { 1 } else { 2 };
match decrypt_cloud_payload(&registered.key, &envelope, alternate) {
Ok(value) => (value, alternate),
Err(second) => {
last_error = Some(format!("{first}; alternate cipher: {second}"));
continue;
}
}
}
};
let parsed: Value = serde_json::from_slice(&plaintext).context("invalid decrypted GREE Cloud JSON")?;
validate_cloud_response(&parsed)?;
let properties = properties_from_payload(&parsed)?;
let message = ParsedCloudMessage {
properties: properties.clone(),
cipher_version,
};
if let Some(item) = self.inner.registered.write().await.get_mut(&cloud_id) {
item.cipher_version = cipher_version;
}
// The reference client accepts both response/* and status/* frames while a
// sequential request is pending. There is only one in-flight request per device;
// final command state is independently verified by the engine.
let waiter = self.inner.pending.lock().await.remove(&cloud_id);
if let Some(waiter) = waiter {
let _ = waiter.send(message.clone());
}
let _ = self.inner.push_events.send(CloudPushEvent {
cloud_device_id: Some(cloud_id.clone()),
parent_mac: registered.parent_mac.clone(),
properties: properties.clone(),
connected: Some(true),
cipher_version: Some(cipher_version),
});
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(registered.device_id.clone()).or_default();
item.connection_state = "online".into();
item.mqtt_state = "connected".into();
item.last_message_timestamp = Some(now);
item.last_mqtt_topic = Some(topic.to_string());
if !properties.is_empty() {
item.last_status_timestamp = Some(now);
for (key, value) in &properties {
item.raw_device_properties.insert(key.clone(), value.clone());
}
item.parsed_device_properties = parsed_properties(&item.raw_device_properties);
item.unknown_properties = item.raw_device_properties
.iter()
.filter(|(key, _)| !KNOWN_CLOUD_PROPERTIES.contains(&key.as_str()))
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
item.energy_related_properties = item.raw_device_properties
.iter()
.filter(|(key, _)| matches!(key.as_str(), "ElcAll" | "ElcAllConsumption" | "CompressorFqy"))
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
}
item.selected_cipher_version = Some(cipher_version);
item.last_error = None;
drop(diagnostics);
self.emit_cloud_mqtt(json!({
"direction": "rx_decrypted",
"topic": topic,
"device_id": registered.device_id,
"cloud_device_id": cloud_id,
"cipher_version": cipher_version,
"properties": properties,
}));
return Ok(());
}
bail!("cannot decrypt GREE Cloud MQTT payload: {}", last_error.unwrap_or_else(|| "unknown cipher error".into()))
}
async fn update_diagnostics_from_message(&self, device: &Device, message: &ParsedCloudMessage) {
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.connection_state = "online".into();
item.mqtt_state = "connected".into();
item.last_message_timestamp = Some(Utc::now());
item.last_status_timestamp = Some(Utc::now());
item.selected_cipher_version = Some(message.cipher_version);
for (key, value) in &message.properties {
item.raw_device_properties.insert(key.clone(), value.clone());
}
item.parsed_device_properties = parsed_properties(&item.raw_device_properties);
item.unknown_properties = item.raw_device_properties
.iter()
.filter(|(key, _)| !KNOWN_CLOUD_PROPERTIES.contains(&key.as_str()))
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
item.energy_related_properties = item.raw_device_properties
.iter()
.filter(|(key, _)| matches!(key.as_str(), "ElcAll" | "ElcAllConsumption" | "CompressorFqy"))
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
item.last_error = None;
}
async fn set_subscriptions(&self, device: &Device, broker: &str, wire_parent: &str) {
let topics = vec![
format!("response/{wire_parent}/#"),
format!("status/{wire_parent}/#"),
format!("connect/{wire_parent}"),
];
{
let mut diagnostics = self.inner.diagnostics.write().await;
let item = diagnostics.entry(device.id.clone()).or_default();
item.broker_host = Some(broker.into());
item.subscribed_topics = topics.clone();
}
self.emit_cloud_mqtt(json!({
"direction": "subscribe",
"device_id": device.id,
"parent_mac": wire_parent,
"topics": topics,
"qos": 1,
}));
}
}
#[derive(Clone)]
pub struct ProviderDispatcher {
local: LocalProvider,
cloud: GreeCloudProvider,
}
impl ProviderDispatcher {
#[cfg(test)]
pub fn new(local: GreeClient, http: reqwest::Client) -> Self {
Self {
local: LocalProvider::new(local),
cloud: GreeCloudProvider::new(http),
}
}
pub fn new_with_debug(
local: GreeClient,
http: reqwest::Client,
api_events: Option<broadcast::Sender<ApiEvent>>,
debug_cloud_requests: Arc<AtomicBool>,
debug_cloud_mqtt: Arc<AtomicBool>,
) -> Self {
Self {
local: LocalProvider::new(local),
cloud: GreeCloudProvider::new_with_debug(
http,
api_events,
debug_cloud_requests,
debug_cloud_mqtt,
),
}
}
pub fn local(&self) -> &LocalProvider {
&self.local
}
pub fn cloud(&self) -> &GreeCloudProvider {
&self.cloud
}
#[cfg(test)]
pub fn provider_name(&self, device: &Device) -> &'static str {
match device.connection_type {
ConnectionType::Local => "local",
ConnectionType::GreeCloud => "gree_cloud",
}
}
}
fn decrypt_cloud_payload(key: &str, envelope: &MqttDeviceEnvelope, cipher: u8) -> Result<Vec<u8>> {
match cipher {
2 => decrypt_v2(
key,
&envelope.pack,
envelope.tag.as_deref().ok_or_else(|| anyhow!("GCM payload is missing tag"))?,
),
_ => decrypt_v1(key, &envelope.pack),
}
}
fn validate_cloud_response(payload: &Value) -> Result<()> {
if let Some(code) = payload.get("r").and_then(value_i64) {
if code != 0 && code != 200 {
let message = payload
.get("msg")
.and_then(Value::as_str)
.unwrap_or("GREE Cloud command rejected");
bail!("GREE Cloud response error {code}: {message}");
}
}
if matches!(payload.get("t").and_then(Value::as_str), Some("error" | "err")) {
let message = payload
.get("msg")
.or_else(|| payload.get("error"))
.and_then(Value::as_str)
.unwrap_or("GREE Cloud command rejected");
bail!("GREE Cloud response error: {message}");
}
Ok(())
}
fn properties_from_payload(payload: &Value) -> Result<BTreeMap<String, Value>> {
if payload.get("t").and_then(Value::as_str) != Some("dat") {
return Ok(BTreeMap::new());
}
let cols = payload.get("cols").and_then(Value::as_array).ok_or_else(|| anyhow!("GREE Cloud dat payload has no cols"))?;
let dat = payload.get("dat").and_then(Value::as_array).ok_or_else(|| anyhow!("GREE Cloud dat payload has no dat"))?;
if cols.len() != dat.len() {
bail!("GREE Cloud dat payload column/value count mismatch");
}
Ok(cols
.iter()
.zip(dat)
.filter_map(|(key, value)| key.as_str().map(|key| (key.to_string(), value.clone())))
.collect())
}
fn apply_cloud_capability_snapshot(device: &mut Device, props: &BTreeMap<String, Value>) {
device.capabilities.temperature_step = if props
.get("HalfTemEn")
.and_then(value_i64)
.is_some_and(|value| value == 1)
{
0.5
} else {
1.0
};
device.capabilities.vertical_swing = props.contains_key("SwUpDn");
device.capabilities.horizontal_swing = props.contains_key("SwingLfRig");
device.supports_light = Some(props.contains_key("Lig"));
device.supports_quiet = Some(props.contains_key("Quiet"));
device.supports_turbo = Some(props.contains_key("Tur"));
device.supports_air = Some(props.contains_key("Air"));
device.supports_xfan = Some(props.contains_key("Blo"));
device.supports_health = Some(props.contains_key("Health"));
device.supports_sleep = Some(props.contains_key("SwhSlp"));
if props.contains_key("Buzzer_ON_OFF") || props.contains_key("BuzzerCtrl") {
device.supports_buzzer_control = Some(true);
}
device.supports_energy_meter = Some(props.contains_key("ElcAll"));
}
pub fn apply_cloud_properties(device: &mut Device, props: &BTreeMap<String, Value>) {
if let Some(value) = props.get("Pow").and_then(value_i64) { device.power = value != 0; }
if let Some(value) = props.get("Mod").and_then(value_i64) { if let Some(mode) = wire_to_mode(value) { device.mode = mode.into(); } }
if let Some(value) = props.get("WdSpd").and_then(value_i64) { device.fan_speed = value.clamp(0, 5) as u8; }
if let Some(value) = props.get("SwUpDn").and_then(value_i64) { device.swing_vertical = value != 0; }
if let Some(value) = props.get("SwingLfRig").and_then(value_i64) { device.swing_horizontal = value != 0; }
if let Some(value) = props.get("Quiet").and_then(value_i64) { device.quiet = value != 0; if value > 0 { device.quiet_wire_value = Some(value.clamp(1, 255) as u8); } }
if let Some(value) = props.get("Tur").and_then(value_i64) { device.turbo = value != 0; }
if let Some(value) = props.get("Lig").and_then(value_i64) { device.light = value != 0; }
if let Some(value) = props.get("Air").and_then(value_i64) { device.air = value != 0; }
if let Some(value) = props.get("Blo").and_then(value_i64) { device.xfan = value != 0; }
if let Some(value) = props.get("Health").and_then(value_i64) { device.health = value != 0; }
if let Some(value) = props.get("SwhSlp").and_then(value_i64) { device.sleep = value != 0; }
if let Some(raw) = props.get("TemSen").and_then(value_f64) {
if let Some((temperature, offset)) = decode_indoor_temperature(raw) {
device.current_temperature = Some(temperature);
device.temperature_sensor_offset = Some(offset);
}
}
let outdoor_raw = props
.get("OutEnvTem")
.and_then(value_f64)
.or_else(|| props.get("TemsSenOut").and_then(value_f64));
if let Some(raw) = outdoor_raw {
if let Some(temperature) = decode_outdoor_temperature(raw, device.temperature_sensor_offset) {
device.outdoor_temperature = Some(temperature);
}
}
if let Some(half_enabled) = props.get("HalfTemEn").and_then(value_i64) {
device.capabilities.temperature_step = if half_enabled == 1 { 0.5 } else { 1.0 };
}
if props.contains_key("SwUpDn") { device.capabilities.vertical_swing = true; }
if props.contains_key("SwingLfRig") { device.capabilities.horizontal_swing = true; }
if let Some(value) = props.get("SetDeciTem").and_then(value_f64) {
device.target_temperature = value / 10.0;
} else if let Some(value) = props.get("SetTem").and_then(value_f64) {
let half = props.get("TemRec").and_then(value_f64).unwrap_or(0.0);
device.target_temperature = value + if half > 0.0 { 0.5 } else { 0.0 };
}
if props.contains_key("Lig") { device.supports_light = Some(true); }
if props.contains_key("Quiet") { device.supports_quiet = Some(true); }
if props.contains_key("Tur") { device.supports_turbo = Some(true); }
if props.contains_key("Air") { device.supports_air = Some(true); }
if props.contains_key("Blo") { device.supports_xfan = Some(true); }
if props.contains_key("Health") { device.supports_health = Some(true); }
if props.contains_key("SwhSlp") { device.supports_sleep = Some(true); }
if props.contains_key("Buzzer_ON_OFF") || props.contains_key("BuzzerCtrl") { device.supports_buzzer_control = Some(true); }
if props.contains_key("ElcAll") { device.supports_energy_meter = Some(true); }
if let Some(value) = props.get("ElcAll").and_then(value_f64) { device.total_energy_kwh = Some(value * 0.1); }
if let Some(value) = props.get("CompressorFqy").and_then(value_f64) { device.compressor_frequency_hz = Some(value); }
if let Some(hid) = props.get("hid").and_then(Value::as_str) {
if device.firmware.is_empty() { device.firmware = firmware_from_hid(hid).unwrap_or_default(); }
}
}
fn parsed_properties(props: &BTreeMap<String, Value>) -> BTreeMap<String, Value> {
let mut parsed = BTreeMap::new();
if let Some(value) = props.get("Pow").and_then(value_i64) { parsed.insert("power".into(), json!(value != 0)); }
if let Some(value) = props.get("Mod").and_then(value_i64).and_then(wire_to_mode) { parsed.insert("mode".into(), json!(value)); }
let mut sensor_offset = None;
if let Some(raw) = props.get("TemSen").and_then(value_f64) {
if let Some((temperature, offset)) = decode_indoor_temperature(raw) {
sensor_offset = Some(offset);
parsed.insert("indoor_temperature_c".into(), json!(temperature));
}
}
let outdoor_raw = props
.get("OutEnvTem")
.and_then(value_f64)
.or_else(|| props.get("TemsSenOut").and_then(value_f64));
if let Some(raw) = outdoor_raw {
if let Some(temperature) = decode_outdoor_temperature(raw, sensor_offset) {
parsed.insert("outdoor_temperature_c".into(), json!(temperature));
}
}
if let Some(value) = props.get("ElcAll").and_then(value_f64) { parsed.insert("total_energy_kwh".into(), json!(value * 0.1)); }
if let Some(value) = props.get("CompressorFqy").and_then(value_f64) { parsed.insert("compressor_frequency_hz".into(), json!(value)); }
parsed
}
fn merge_command(target: &mut DeviceCommand, source: &DeviceCommand) {
if source.power.is_some() { target.power = source.power; }
if source.mode.is_some() { target.mode = source.mode.clone(); }
if source.target_temperature.is_some() { target.target_temperature = source.target_temperature; }
if source.fan_speed.is_some() { target.fan_speed = source.fan_speed; }
if source.swing_vertical.is_some() { target.swing_vertical = source.swing_vertical; }
if source.swing_horizontal.is_some() { target.swing_horizontal = source.swing_horizontal; }
if source.quiet.is_some() { target.quiet = source.quiet; }
if source.turbo.is_some() { target.turbo = source.turbo; }
if source.light.is_some() { target.light = source.light; }
if source.air.is_some() { target.air = source.air; }
if source.xfan.is_some() { target.xfan = source.xfan; }
if source.health.is_some() { target.health = source.health; }
if source.sleep.is_some() { target.sleep = source.sleep; }
}
fn mode_to_wire(mode: &str) -> Result<i64> {
match mode {
"auto" => Ok(0), "cool" => Ok(1), "dry" => Ok(2), "fan" => Ok(3), "heat" => Ok(4),
_ => bail!("unsupported GREE Cloud mode: {mode}"),
}
}
fn decode_indoor_temperature(raw: f64) -> Option<(f64, bool)> {
if raw == 0.0 || !raw.is_finite() {
return None;
}
// GREE TemSen uses a +40 C wire offset on the modules that report values such as
// 64 for 24 C. Older/local-compatible firmware can also report the real value.
let offset = raw > 40.0;
let temperature = if offset { raw - 40.0 } else { raw };
(-40.0..=80.0).contains(&temperature).then_some((temperature, offset))
}
fn decode_outdoor_temperature(raw: f64, sensor_offset: Option<bool>) -> Option<f64> {
if raw == 0.0 || !raw.is_finite() {
return None;
}
// Follow the same rule already used by the LAN parser: once TemSen proves the +40
// encoding, apply it to OutEnvTem/TemsSenOut as well. Without TemSen, >50 is the
// conservative outdoor sentinel for the encoded form.
let offset = sensor_offset.unwrap_or(raw > 50.0);
let temperature = if offset { raw - 40.0 } else { raw };
(-60.0..=80.0).contains(&temperature).then_some(temperature)
}
fn wire_to_mode(mode: i64) -> Option<&'static str> {
match mode { 0 => Some("auto"), 1 => Some("cool"), 2 => Some("dry"), 3 => Some("fan"), 4 => Some("heat"), _ => None }
}
fn value_i64(value: &Value) -> Option<i64> {
value.as_i64().or_else(|| value.as_u64().and_then(|v| i64::try_from(v).ok())).or_else(|| value.as_f64().map(|v| v as i64)).or_else(|| value.as_str().and_then(|v| v.parse().ok()))
}
fn value_f64(value: &Value) -> Option<f64> {
value.as_f64().or_else(|| value.as_i64().map(|v| v as f64)).or_else(|| value.as_u64().map(|v| v as f64)).or_else(|| value.as_str().and_then(|v| v.parse().ok()))
}
fn parent_mac_preserve_case(mac: &str) -> String {
let compact = mac.trim().replace([':', '-'], "");
if compact.len() > 12 && compact.ends_with("00") {
compact[..compact.len() - 2].to_string()
} else {
compact
}
}
fn cloud_wire_mac(session: &CloudSession, device: &Device) -> String {
let stable = cloud_id(device);
session
.wire_macs
.get(&stable)
.cloned()
// Old databases only contain the normalized ID. GREE's API commonly returns
// lower-case MACs, and MQTT topics are case-sensitive, so lower-case is the
// safest reference-compatible fallback until the next REST refresh succeeds.
.unwrap_or_else(|| stable.to_ascii_lowercase())
}
fn cloud_wire_parent(session: &CloudSession, device: &Device) -> String {
parent_mac_preserve_case(&cloud_wire_mac(session, device))
}
fn cloud_wire_parent_from_map(wire_macs: &HashMap<String, String>, device: &Device) -> String {
let stable = cloud_id(device);
let wire = wire_macs
.get(&stable)
.cloned()
.unwrap_or_else(|| stable.to_ascii_lowercase());
parent_mac_preserve_case(&wire)
}
fn alternate_wire_mac_case(mac: &str) -> Option<String> {
let compact = mac.trim().replace([':', '-'], "");
let upper = compact.to_ascii_uppercase();
if upper != compact {
return Some(upper);
}
let lower = compact.to_ascii_lowercase();
(lower != compact).then_some(lower)
}
fn is_legacy_cloud_firmware(firmware: &str) -> bool {
let value = firmware.trim().to_ascii_uppercase();
value.starts_with("V1.") || value.starts_with("1.")
}
fn normalize_mac(value: &str) -> String { value.trim().replace([':', '-'], "").to_ascii_uppercase() }
fn cloud_id(device: &Device) -> String { normalize_mac(device.cloud_device_id.as_deref().unwrap_or(&device.mac)) }
fn cloud_parent(device: &Device) -> String { device.cloud_parent_mac.clone().unwrap_or_else(|| parent_mac(&cloud_id(device))) }
fn is_timeout_error(error: &anyhow::Error) -> bool { error.to_string().to_ascii_lowercase().contains("timed out") }
fn sanitize_error(value: &str) -> String {
let lower = value.to_ascii_lowercase();
if lower.contains("password") || lower.contains("token") || lower.contains("authorization") { "GREE Cloud authentication/transport error".into() } else { value.chars().take(300).collect() }
}
fn firmware_from_hid(hid: &str) -> Option<String> {
let marker = hid.rfind('V')?;
let value = hid.get(marker + 1..)?.strip_suffix(".bin").unwrap_or(&hid[marker + 1..]);
(!value.is_empty()).then(|| value.to_string())
}
pub async fn cloud_reconnect_loop(provider: GreeCloudProvider, settings: Arc<RwLock<crate::models::RuntimeSettings>>, db: crate::db::Db) {
let mut attempt = 0_u32;
loop {
let cloud = settings.read().await.gree_cloud.clone();
if !cloud.enabled {
attempt = 0;
sleep(Duration::from_secs(15)).await;
continue;
}
if provider.is_connected().await {
attempt = 0;
sleep(Duration::from_secs(10)).await;
continue;
}
let devices = match db.list_devices() {
Ok(items) => items,
Err(err) => { tracing::warn!(error=?err, "cannot list devices for GREE Cloud reconnect"); sleep(Duration::from_secs(10)).await; continue; }
};
if !devices.iter().any(|d| d.enabled && d.connection_type == ConnectionType::GreeCloud) {
sleep(Duration::from_secs(15)).await;
continue;
}
tracing::info!(attempt, "GREE Cloud MQTT reconnect");
match provider.ensure_connected(&cloud, &devices).await {
Ok(()) => {
attempt = 0;
let now = Utc::now();
let mut runtime = settings.write().await;
runtime.gree_cloud.last_successful_contact = Some(now);
if let Err(err) = db.save_runtime_settings(&runtime) {
tracing::warn!(error=?err, "cannot persist GREE Cloud contact timestamp");
}
}
Err(err) => {
attempt = attempt.saturating_add(1).min(8);
tracing::warn!(error=%sanitize_error(&err.to_string()), attempt, "GREE Cloud MQTT reconnect failed");
}
}
let base = (1_u64 << attempt.min(6)).min(60);
let jitter = rand::thread_rng().gen_range(0..=base.min(10));
sleep(Duration::from_secs(base + jitter)).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn status_parser_rejects_malformed_column_count() {
assert!(properties_from_payload(&json!({"t":"dat","cols":["Pow"],"dat":[]})).is_err());
}
#[test]
fn cloud_temperature_wire_offset_matches_gree_encoding() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
device.current_temperature = None;
device.outdoor_temperature = None;
device.temperature_sensor_offset = None;
let props = BTreeMap::from([
("TemSen".into(), json!(64)),
("OutEnvTem".into(), json!(66)),
]);
apply_cloud_properties(&mut device, &props);
assert_eq!(device.current_temperature, Some(24.0));
assert_eq!(device.outdoor_temperature, Some(26.0));
assert_eq!(device.temperature_sensor_offset, Some(true));
let parsed = parsed_properties(&props);
assert_eq!(parsed.get("indoor_temperature_c"), Some(&json!(24.0)));
assert_eq!(parsed.get("outdoor_temperature_c"), Some(&json!(26.0)));
}
#[test]
fn cloud_temperature_without_wire_offset_is_preserved() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
device.temperature_sensor_offset = None;
let props = BTreeMap::from([
("TemSen".into(), json!(23)),
("OutEnvTem".into(), json!(18)),
]);
apply_cloud_properties(&mut device, &props);
assert_eq!(device.current_temperature, Some(23.0));
assert_eq!(device.outdoor_temperature, Some(18.0));
assert_eq!(device.temperature_sensor_offset, Some(false));
}
#[test]
fn cloud_energy_scale_is_tenths_of_kwh() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
let props = BTreeMap::from([("ElcAll".into(), json!(1521))]);
apply_cloud_properties(&mut device, &props);
assert_eq!(device.total_energy_kwh, Some(152.1));
}
#[test]
fn command_mode_mapping_matches_reference() {
assert_eq!(mode_to_wire("auto").unwrap(), 0);
assert_eq!(mode_to_wire("heat").unwrap(), 4);
assert!(mode_to_wire("unsupported").is_err());
}
#[test]
fn cloud_wire_parent_preserves_rest_mac_case_for_mqtt_topics() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
device.cloud_device_id = Some("9424B80C5DB0".into());
let wire = HashMap::from([
("9424B80C5DB0".to_string(), "9424b80c5db0".to_string()),
]);
assert_eq!(cloud_wire_parent_from_map(&wire, &device), "9424b80c5db0");
}
#[test]
fn legacy_cloud_firmware_uses_compatibility_status_profile() {
assert!(is_legacy_cloud_firmware("V1.2.1"));
assert!(is_legacy_cloud_firmware("1.21"));
assert!(!is_legacy_cloud_firmware("V3.4.M"));
}
#[test]
fn mqtt_wire_case_probe_keeps_mac_bytes_and_changes_only_case() {
assert_eq!(
alternate_wire_mac_case("502cc699c117").as_deref(),
Some("502CC699C117")
);
assert_eq!(
alternate_wire_mac_case("502CC699C117").as_deref(),
Some("502cc699c117")
);
}
#[test]
fn legacy_status_profile_excludes_newer_optional_columns() {
assert!(LEGACY_CLOUD_PROPERTIES.contains(&"Pow"));
assert!(LEGACY_CLOUD_PROPERTIES.contains(&"SetTem"));
assert!(!LEGACY_CLOUD_PROPERTIES.contains(&"HalfTemEn"));
assert!(!LEGACY_CLOUD_PROPERTIES.contains(&"ElcAll"));
assert!(!LEGACY_CLOUD_PROPERTIES.contains(&"CompressorFqy"));
}
#[tokio::test]
async fn mqtt_payload_is_ignored_after_all_cloud_devices_are_removed() {
let provider = GreeCloudProvider::new(reqwest::Client::new());
let result = provider
.handle_raw_message("status/stale-parent/device", b"not-json-anymore")
.await;
assert!(result.is_ok());
}
#[tokio::test]
async fn mqtt_payload_for_unknown_parent_still_errors_when_devices_are_registered() {
let provider = GreeCloudProvider::new(reqwest::Client::new());
provider.inner.registered.write().await.insert(
"AABBCCDDEEFF".into(),
RegisteredDevice {
device_id: "cloud-test".into(),
key: "0123456789abcdef".into(),
parent_mac: "AABBCCDDEE".into(),
cipher_version: 1,
},
);
let err = provider
.handle_raw_message(
"status/1122334455/device",
br#"{"pack":"x","tcid":"112233445566"}"#,
)
.await
.unwrap_err();
assert!(err
.to_string()
.contains("does not match a registered device"));
}
#[tokio::test]
async fn provider_dispatcher_selects_transport_from_connection_type() {
let local = GreeClient::new(
"test-controller".into(),
None,
None,
Arc::new(AtomicBool::new(false)),
);
let providers = ProviderDispatcher::new(local, reqwest::Client::new());
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::Local;
assert_eq!(providers.provider_name(&device), "local");
device.connection_type = ConnectionType::GreeCloud;
assert_eq!(providers.provider_name(&device), "gree_cloud");
}
#[tokio::test]
async fn cloud_provider_rejects_local_device_before_network_io() {
let provider = GreeCloudProvider::new(reqwest::Client::new());
let device = Device::simulated_default();
let error = provider.register_device(&device).await.unwrap_err();
assert!(error.to_string().contains("Local device"));
}
#[test]
fn full_cloud_snapshot_can_disable_unreported_capabilities() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
let props = BTreeMap::from([
("Pow".into(), json!(1)),
("Lig".into(), json!(0)),
("HalfTemEn".into(), json!(1)),
]);
apply_cloud_capability_snapshot(&mut device, &props);
assert_eq!(device.supports_light, Some(true));
assert_eq!(device.supports_turbo, Some(false));
assert!(!device.capabilities.vertical_swing);
assert_eq!(device.capabilities.temperature_step, 0.5);
}
#[test]
fn partial_cloud_push_does_not_clear_known_capabilities() {
let mut device = Device::simulated_default();
device.connection_type = ConnectionType::GreeCloud;
device.supports_light = Some(true);
device.capabilities.vertical_swing = true;
apply_cloud_properties(&mut device, &BTreeMap::from([("Pow".into(), json!(1))]));
assert_eq!(device.supports_light, Some(true));
assert!(device.capabilities.vertical_swing);
}
#[test]
fn cloud_error_response_is_rejected() {
assert!(validate_cloud_response(&json!({"r": 1, "msg": "rejected"})).is_err());
assert!(validate_cloud_response(&json!({"t": "error", "msg": "bad"})).is_err());
assert!(validate_cloud_response(&json!({"r": 0})).is_ok());
}
#[tokio::test]
async fn cloud_command_order_matches_reference() {
let provider = GreeCloudProvider::new(reqwest::Client::new());
let mut device = Device::simulated_default();
device.id = "cloud-test".into();
device.connection_type = ConnectionType::GreeCloud;
device.cloud_device_id = Some("AABBCCDDEEFF".into());
device.key = Some("0123456789abcdef".into());
provider.inner.diagnostics.write().await.insert(
device.id.clone(),
CloudDiagnostics {
raw_device_properties: BTreeMap::from([
("HalfTemEn".into(), json!(1)),
("TemUn".into(), json!(0)),
]),
..Default::default()
},
);
let command = DeviceCommand {
power: Some(true),
mode: Some("cool".into()),
target_temperature: Some(23.5),
fan_speed: Some(3),
..Default::default()
};
let sequence = provider
.build_command_sequence(&device, &command, false)
.await
.unwrap();
assert_eq!(sequence.first().unwrap().0, vec!["Mod".to_string()]);
assert_eq!(sequence.last().unwrap().0, vec!["Pow".to_string()]);
let temperature = sequence
.iter()
.find(|(opt, _, _)| opt.first().is_some_and(|value| value == "SetTem"))
.unwrap();
assert_eq!(
temperature.0.iter().map(String::as_str).collect::<Vec<_>>(),
vec!["SetTem", "TemRec", "SetDeciTem", "Add0.5"]
);
assert_eq!(temperature.1[2], json!(235));
let sleep_sequence = provider
.build_command_sequence(
&device,
&DeviceCommand { sleep: Some(true), ..Default::default() },
false,
)
.await
.unwrap();
assert_eq!(sleep_sequence.len(), 2);
assert_eq!(sleep_sequence[0].0, vec!["SwhSlp".to_string()]);
assert_eq!(sleep_sequence[1].0, vec!["SlpMod".to_string()]);
let muted = provider
.build_command_sequence(&device, &command, true)
.await
.unwrap();
assert!(muted.iter().all(|(opt, values, _)| {
opt.last().is_some_and(|name| name == "Buzzer_ON_OFF")
&& values.last() == Some(&json!(1))
}));
}
}
Reference in New Issue View Git Blame Copy Permalink